Cesar Mendivil
8b458a3ddf
- Add Next.js app structure with base configs, linting, and formatting - Implement LiveKit Meet page, types, and utility functions - Add Docker, Compose, and deployment scripts for backend and token server - Provide E2E and smoke test scaffolding with Puppeteer and Playwright helpers - Include CSS modules and global styles for UI - Add postMessage and studio integration utilities - Update package.json with dependencies and scripts for development and testing
217 lines
9.5 KiB
Python
217 lines
9.5 KiB
Python
#!/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)
|