"""Browser tests for ani-web (Playwright, headless Chromium). Run via tests/run.py or standalone: python tests/test_browser.py Covers what HTTP tests can't: the login UI, the user chip, and real video playback behind auth. Needs: pip install playwright && playwright install chromium """ import asyncio import http.cookiejar import json import sys import urllib.request from playwright.async_api import async_playwright BASE = "http://127.0.0.1:8399" LAUNCH_ARGS = ["--autoplay-policy=no-user-gesture-required"] # This headless Chromium stops generating animation frames on idle pages, # which stalls rAF-based Playwright waits (wait_for_function's default # polling, click actionability). Poll by timer instead; force the one click. POLL = 100 async def _login(page, user, pw): await page.goto(BASE + "/") await page.wait_for_selector("input#pw") assert "create account" not in await page.content(), "self-signup UI must be absent" await page.fill("#user", user) await page.fill("#pw", pw) # Enter instead of clicking #go: this playwright/chromium combo's # "element stable" actionability check hangs non-deterministically; # fill() leaves focus on the field, so a bare keypress submits. await page.keyboard.press("Enter") await page.wait_for_selector("#q", timeout=15000) await page.wait_for_function( f"document.getElementById('me').textContent.includes('{user}')", timeout=5000, polling=POLL) async def test_login_ui(): async with async_playwright() as p: browser = await p.chromium.launch(args=LAUNCH_ARGS) page = await (await browser.new_context()).new_page() errors = [] page.on("pageerror", lambda e: errors.append(str(e))) # unauth -> login form await page.goto(BASE + "/") await page.wait_for_selector("input#pw") assert "/login" in page.url, page.url # wrong password -> inline error await page.fill("#user", "admin") await page.fill("#pw", "wrong") await page.keyboard.press("Enter") await page.wait_for_function("document.querySelector('#err').textContent.length > 0", timeout=10000, polling=POLL) assert "wrong" in await page.inner_text("#err") # correct -> home with user chip await page.fill("#pw", "adminpw12345") await page.keyboard.press("Enter") await _login_settled(page) assert await page.inner_text("#me") == "hi, admin" assert "Seed Anime" in await page.inner_text("#main"), "seeded history must show" assert not errors, errors await browser.close() print("ok test_login_ui") async def _login_settled(page): await page.wait_for_selector("#q", timeout=15000) await page.wait_for_function( "document.getElementById('me').textContent.length > 0", timeout=5000, polling=POLL) async def test_video_plays_behind_auth(): async with async_playwright() as p: browser = await p.chromium.launch(args=LAUNCH_ARGS) page = await (await browser.new_context()).new_page() errors = [] page.on("pageerror", lambda e: errors.append(str(e))) page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None) await _login(page, "alice", "alicepw12345") await page.fill("#q", "chainsaw") await page.keyboard.press("Enter") await page.wait_for_selector("a.card", timeout=30000) await page.click("a.card", force=True) await page.wait_for_selector("video", timeout=30000) await page.wait_for_function( "document.querySelector('video').readyState >= 4 && document.querySelector('video').currentTime > 1", timeout=120000, polling=250) # 120s: cold CDN/upstream starts can be slow real_errors = [e for e in errors if "401" not in e] assert not real_errors, real_errors await browser.close() print("ok test_video_plays_behind_auth") async def test_history_page_and_delete(): # self-contained: seed erin's own entries over HTTP first jar = http.cookiejar.CookieJar() op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) op.open(urllib.request.Request(BASE + "/login", data=json.dumps({"username": "erin", "password": "erinpw12345"}).encode(), headers={"Content-Type": "application/json"}, method="POST")) for e in ({"title": "Hist A", "slug_id": "hist-a", "season_id": 1, "ep_id": 7, "lang": "eng", "ep": 7, "pos": 5}, {"title": "Hist B", "slug_id": "hist-b", "season_id": 2, "ep_id": 9, "lang": "jpn", "ep": 9, "pos": 30, "dur": 600}): op.open(urllib.request.Request(BASE + "/api/history", data=json.dumps(e).encode(), headers={"Content-Type": "application/json"}, method="POST")) async with async_playwright() as p: browser = await p.chromium.launch(args=LAUNCH_ARGS) page = await (await browser.new_context()).new_page() errors = [] page.on("pageerror", lambda e: errors.append(str(e))) await _login(page, "erin", "erinpw12345") await page.goto(BASE + "/?history") await page.wait_for_selector(".card.hist", timeout=15000) await page.wait_for_function( "document.querySelectorAll('.card.hist').length == 2", timeout=15000, polling=POLL) await page.click(".card.hist .del", force=True) await page.wait_for_function( "document.querySelectorAll('.card.hist').length == 1", timeout=15000, polling=POLL) await page.wait_for_selector(".toast", timeout=5000) assert not errors, errors await browser.close() print("ok test_history_page_and_delete") async def test_stream_failure_retry(): async with async_playwright() as p: browser = await p.chromium.launch(args=LAUNCH_ARGS) page = await (await browser.new_context()).new_page() async def abort_stream(route): await route.abort() # note: **/stream/* (single star) — must NOT match the # /api/episode/{id}/stream token endpoint, only the /stream/{token} proxy await page.route("**/stream/*", abort_stream) await _login(page, "admin", "adminpw12345") await page.fill("#q", "chainsaw") await page.keyboard.press("Enter") await page.wait_for_selector("a.card", timeout=30000) await page.click("a.card", force=True) # detail auto-plays ep1; with /stream aborted hls.js exhausts its # retries and surfaces the fatal error -> failure overlay with Retry await page.wait_for_selector(".pl-overlay button", timeout=90000) await page.unroute("**/stream/*", abort_stream) await page.click(".pl-overlay button", force=True) await page.wait_for_function( "document.querySelector('video').readyState >= 4 && document.querySelector('video').currentTime > 1", timeout=120000, polling=250) await browser.close() print("ok test_stream_failure_retry") if __name__ == "__main__": if len(sys.argv) > 1: BASE = sys.argv[1] asyncio.run(test_login_ui()) asyncio.run(test_video_plays_behind_auth()) asyncio.run(test_history_page_and_delete()) asyncio.run(test_stream_failure_retry())