#!/usr/bin/env python3 """ playwright_py_runner.py Runner E2E simple usando Playwright (Python). Funciona en modo local (lanza Chromium) y puede usarse para ver logs en tiempo real, tomar capturas y ejecutar pasos configurables. Uso: # instalar dependencias (una sola vez) python3 -m pip install --user playwright requests python3 -m playwright install chromium # ejecutar y pedir al backend crear una sesión python3 packages/broadcast-panel/e2e/playwright_py_runner.py --create-session --backend http://localhost:4000 --room test-room --username tester --out /tmp/py-playwright-shot.png Opciones (resumen): --url URL URL del frontend (por defecto http://localhost:5176) (usado si no se crea sesión) --out PATH Ruta para guardar captura (por defecto /tmp/py-playwright-shot.png) --headful Abrir navegador en modo no headless (útil para debugging) --wait N Tiempo en segundos a esperar entre pasos (default 1) --ws WS_ENDPOINT Conectar a Playwright run-server remoto (ws://host:port) --create-session Hacer POST a BACKEND/api/session y abrir la redirectUrl devuelta --backend URL Backend base (por defecto http://localhost:4000) --room NAME Nombre de la sala para crear session (por defecto 'test-room') --username NAME Username para crear session (por defecto 'e2e-py') El script imprime logs sencillos por stdout (progreso de pasos y errores). """ import sys import argparse import time import json import os from pathlib import Path try: from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError except Exception as e: print("ERROR: Playwright Python no está instalado. Ejecuta:\n python3 -m pip install --user playwright\n python3 -m playwright install chromium\n") raise # Try to import requests, fallback to urllib try: import requests # type: ignore HAS_REQUESTS = True except Exception: import urllib.request as _urllib_request HAS_REQUESTS = False def create_session_via_backend(backend_base: str, room: str, username: str, ttl: int | None = None, headers: dict | None = None): url = backend_base.rstrip('/') + '/api/session' payload = { 'room': room, 'username': username } if ttl: payload['ttl'] = int(ttl) print(f"[backend] Creating session at {url} with payload {payload}") try: if HAS_REQUESTS: resp = requests.post(url, json=payload, headers=headers or {}, timeout=15) try: data = resp.json() except Exception: data = { 'status_code': resp.status_code, 'text': resp.text } return resp.status_code, data else: req = _urllib_request.Request(url, data=json.dumps(payload).encode('utf-8'), headers={'Content-Type':'application/json', **(headers or {})}, method='POST') with _urllib_request.urlopen(req, timeout=15) as r: b = r.read() try: data = json.loads(b.decode('utf-8')) except Exception: data = { 'text': b.decode('utf-8', errors='replace') } return r.getcode(), data except Exception as e: print('[backend] Request failed:', e) return None, { 'error': str(e) } def run(url: str, out_path: str, headful: bool, wait_seconds: float, ws_endpoint: str | None = None, create_session: bool = False, backend: str | None = None, room: str | None = None, username: str | None = None): print(f"[runner] Starting Playwright runner (headful={headful})") # If requested, create session first and override navigation URL if create_session: backend_base = backend or 'http://localhost:4000' status, data = create_session_via_backend(backend_base, room or 'test-room', username or 'e2e-py') print('[backend] create response:', status, data) if status and status >= 200 and status < 300: # Prefer redirectUrl, then studioUrl, else fall back to token url or provided url nav = data.get('redirectUrl') or data.get('studioUrl') or data.get('url') or url print(f"[runner] Will navigate to backend-provided URL: {nav}") url = nav else: print('[runner] Backend session creation failed; will navigate to provided URL instead') with sync_playwright() as p: browser = None try: # If a ws endpoint is provided, try to connect to remote Playwright run-server if ws_endpoint: print(f"[runner] Attempting to connect to Playwright server at {ws_endpoint}") try: browser = p.connect(ws_endpoint=ws_endpoint) print('[runner] Connected to remote Playwright server') except Exception as conn_err: print('[runner] Remote connect failed, falling back to local launch:', conn_err) browser = p.chromium.launch(headless=not headful) else: browser = p.chromium.launch(headless=not headful) context = browser.new_context() page = context.new_page() page.on("console", lambda msg: print(f"[page console] {msg.type}: {msg.text}")) page.on("pageerror", lambda exc: print(f"[page error] {exc}")) print(f"[runner] Navigating to {url}") page.goto(url, wait_until="networkidle", timeout=30000) print(f"[runner] Page loaded: {page.url}") time.sleep(wait_seconds) # Try common flows: click 'Entrar al Estudio' if exists try_click_text(page, 'Entrar al Estudio') time.sleep(wait_seconds) # If creation modal flow (try alternative selectors) if try_click_text(page, 'Crear transmisión'): print('[runner] Opened create modal (Crear transmisión)') time.sleep(wait_seconds) try_click_text(page, 'Omitir por ahora') time.sleep(0.5) # try to fill first input with 'Transmitir' try: inp = page.query_selector('input') if inp: inp.fill('Transmitir') print('[runner] Filled input with "Transmitir"') except Exception as e: print('[runner] Input fill error', e) try_click_text(page, 'Empezar ahora') time.sleep(wait_seconds) try_click_text(page, 'Entrar al Estudio') time.sleep(wait_seconds) # Final: take screenshot outp = Path(out_path) outp.parent.mkdir(parents=True, exist_ok=True) print(f"[runner] Taking screenshot -> {outp}") page.screenshot(path=str(outp), full_page=True) print(f"[runner] Screenshot saved: {outp}") # Optionally log current cookies / localStorage try: cookies = context.cookies() print(f"[runner] Cookies: {len(cookies)} entries") except Exception: pass context.close() return True except PlaywrightTimeoutError as te: print('[runner][error] Timeout', te) return False except Exception as ex: print('[runner][error] Exception', ex) return False finally: try: if browser: browser.close() except Exception: pass def try_click_text(page, text: str) -> bool: try: locator = page.locator(f"text={text}") if locator.count() > 0: locator.first.click(timeout=5000) print(f"[runner] Clicked text: '{text}'") return True else: print(f"[runner] Text not found: '{text}'") return False except Exception as e: print(f"[runner] Click error for '{text}': {e}") return False if __name__ == '__main__': ap = argparse.ArgumentParser() default_out = os.path.join(os.path.dirname(__file__), 'out', 'py-playwright-shot.png') os.makedirs(os.path.dirname(default_out), exist_ok=True) ap.add_argument('--url', default='http://localhost:5176', help='URL del frontend (por defecto http://localhost:5176)') ap.add_argument('--out', default=default_out, help='Ruta de captura') ap.add_argument('--headful', action='store_true', help='Abrir navegador en modo no headless') ap.add_argument('--wait', type=float, default=1.0, help='Segundos de espera entre pasos') ap.add_argument('--ws', default=None, help='WebSocket endpoint del Playwright run-server (ej. ws://localhost:3003)') ap.add_argument('--create-session', action='store_true', help='Pedir al backend que cree una session y navegar a la redirectUrl') ap.add_argument('--backend', default='http://localhost:4000', help='Backend base URL') ap.add_argument('--room', default='test-room', help='Room name para crear session') ap.add_argument('--username', default='e2e-py', help='Username para crear session') args = ap.parse_args() try: ok = run(args.url, args.out, args.headful, args.wait, args.ws, args.create_session, args.backend, args.room, args.username) if ok: print('[runner] Finished successfully') sys.exit(0) else: print('[runner] Finished with errors') sys.exit(2) except Exception as e: import traceback print('[runner][fatal] Unhandled exception:') traceback.print_exc() sys.exit(3)