"""Browser tests for ani-web (Playwright, headless Chromium). Run via pytest (see conftest.py for the throwaway server). Covers what HTTP tests can't: the login UI, the user chip, and real video playback behind auth. These tests scrape and play LIVE upstream content, so they fail whenever a provider is mid-rotation or a title is taken down — that is not a code regression. They therefore stay OFF by default (the default suite must be deterministic); run them deliberately with ANIWEB_LIVE_TESTS=1. """ import asyncio import http.cookiejar import json import os import urllib.request import pytest if os.environ.get("ANIWEB_LIVE_TESTS") != "1": pytest.skip("live-upstream e2e — run with ANIWEB_LIVE_TESTS=1", allow_module_level=True) async_playwright = pytest.importorskip("playwright.async_api").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=30000) # 30s: full-suite load slows the home render await page.wait_for_function( f"document.getElementById('me').textContent.includes('{user}')", timeout=5000, polling=POLL) def test_login_ui(): asyncio.run(_login_ui()) async def _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 await page.query_selector('link[rel="icon"]'), "favicon missing" assert "Seed Anime" in await page.inner_text("#main"), "seeded history must show" assert not errors, errors await browser.close() async def _login_settled(page): await page.wait_for_selector("#q", timeout=30000) # 30s: full-suite load slows the home render await page.wait_for_function( "document.getElementById('me').textContent.length > 0", timeout=5000, polling=POLL) def test_video_plays_behind_auth(): asyncio.run(_video_plays_behind_auth()) async def _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() def test_history_page_and_delete(): asyncio.run(_history_page_and_delete()) async def _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", "X-Aniweb": "1"}, method="POST")) for n in range(1, 6): # 5 entries: makes the home "history →" link render e = {"title": f"Hist {n}", "slug_id": f"hist-{n}", "season_id": n, "ep_id": n, "lang": "eng", "ep": n, "pos": 5, "dur": 600} op.open(urllib.request.Request(BASE + "/api/history", data=json.dumps(e).encode(), headers={"Content-Type": "application/json", "X-Aniweb": "1"}, 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") # regression: the home history link must go to the ?history query, not the /history path await page.goto(BASE + "/") await page.wait_for_selector("a:has-text('history →')", timeout=15000) assert (await page.get_attribute("a:has-text('history →')", "href")).endswith("/?history") await page.click("a:has-text('history →')", force=True) await page.wait_for_selector(".card.hist", timeout=15000) await page.wait_for_function( "document.querySelectorAll('.card.hist').length == 5", timeout=15000, polling=POLL) await page.click(".card.hist .del", force=True) await page.wait_for_function( "document.querySelectorAll('.card.hist').length == 4", timeout=15000, polling=POLL) await page.wait_for_selector(".toast", timeout=5000) assert not errors, errors await browser.close() def test_stream_failure_retry(): asyncio.run(_stream_failure_retry()) async def _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() 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. Wait only for playlist metadata (duration # comes from the VOD playlist, NOT from segment bytes) — this must # work even when the CDN is slow/flaky on segments await page.wait_for_function( "document.querySelector('video')?.duration > 0", timeout=120000, polling=250) # wipe this show's history: the initial play() write (and any entry # left by earlier tests) would otherwise mask whether the error-time # save happened — from here on the only write can be that save slug = await page.evaluate("new URLSearchParams(location.search).get('anime')") h0 = await page.evaluate("fetch('api/history?all=1').then(r => r.json())") for e in (x for x in h0 if x["slug_id"] == slug): await page.evaluate( "([s, ep]) => fetch('api/history?slug_id=' + encodeURIComponent(s) + '&ep_id=' + ep," " {method:'DELETE', headers:{'X-Aniweb':'1'}})", [e["slug_id"], e["ep_id"]]) # seek to 6s: below the 10s periodic-save threshold (and above the # 3s minimum), so NO periodic save can fire — only an error-time # save could persist ~6s. The seek reflects immediately even though # the bytes never arrive (we kill /stream next) await page.evaluate("document.querySelector('video').currentTime = 6") # kill the stream: hls.js's frag loads abort, retries exhaust, # fatal error -> playerFail, which must save the position # (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 page.wait_for_selector(".pl-overlay button", timeout=90000) # pre-fix nothing was written after the DELETE above hist = await page.evaluate("fetch('api/history?all=1').then(r => r.json())") assert any(e["slug_id"] == slug and e["pos"] >= 5 for e in hist), hist # and the player is recoverable: Retry re-invokes play(), which # starts with a fresh token fetch (assert that — not segment bytes, # which a hanging CDN would make environmental; real playback is # covered by the other video tests) await page.unroute("**/stream/*", abort_stream) async with page.expect_request("**/api/episode/*/stream*", timeout=30000): await page.click(".pl-overlay button", force=True) await browser.close() def test_recent_searches(): asyncio.run(_recent_searches()) async def _recent_searches(): # UI: focus the (empty) search box -> dropdown of saved searches; # typing filters; clicking one runs the search. jar = http.cookiejar.CookieJar() op = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) op.open(urllib.request.Request(BASE + "/login", data=json.dumps({"username": "alice", "password": "alicepw12345"}).encode(), headers={"Content-Type": "application/json", "X-Aniweb": "1"}, method="POST")) op.open(urllib.request.Request(BASE + "/api/searches", data=json.dumps({"q": "chainsaw"}).encode(), headers={"Content-Type": "application/json", "X-Aniweb": "1"}, 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, "alice", "alicepw12345") await page.focus("#q") await page.wait_for_selector(".recent.open", timeout=10000) assert "chainsaw" in await page.inner_text(".recent") # typing narrows the list to the matching entry await page.fill("#q", "cha") await page.wait_for_function( "document.querySelectorAll('.recent button').length === 1", timeout=5000, polling=POLL) await page.click(".recent button", force=True) await page.wait_for_selector("a.card", timeout=30000) assert "q=chainsaw" in page.url, page.url assert not errors, errors await browser.close() def test_live_search(): asyncio.run(_live_search()) async def _live_search(): # typing (no Enter) renders results in place; Enter still commits to ?q= # so results stay shareable/reloadable. 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, "admin", "adminpw12345") await page.fill("#q", "chainsaw") # debounced live results must appear WITHOUT pressing Enter await page.wait_for_selector("main a.card", timeout=30000) assert "q=chainsaw" not in page.url, page.url # Enter commits the query to the URL await page.keyboard.press("Enter") await page.wait_for_function( "location.search.includes('q=chainsaw')", timeout=15000, polling=POLL) assert not errors, errors await browser.close() def test_auto_next_countdown(): asyncio.run(_auto_next_countdown()) async def _auto_next_countdown(): # playing a full episode to its natural end is too slow; simulate the # `ended` event and verify the countdown overlay + cancel path. 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, "admin", "adminpw12345") # straight to the 12-ep series (the first search card is the 1-ep # movie — no next episode there) await page.goto(BASE + "/?anime=chainsaw-man-922") await page.wait_for_selector("video", timeout=30000) await page.wait_for_function( "document.querySelector('video').readyState >= 4", timeout=120000, polling=250) # cold CDN/upstream starts can be slow await page.evaluate( "document.querySelector('video').dispatchEvent(new Event('ended'))") await page.wait_for_selector(".pl-overlay .btn", timeout=10000) assert "Next: E2" in await page.inner_text(".pl-overlay") # cancel (x): overlay gone, still on the same episode await page.click(".pl-overlay .cnt-x", force=True) await page.wait_for_function( "!document.querySelector('.pl-overlay')", timeout=5000, polling=POLL) assert "E1" in await page.inner_text(".ep.active") assert not errors, errors await browser.close() def test_episode_watch_state(): asyncio.run(_episode_watch_state()) async def _episode_watch_state(): # completed episodes dimmed with a check, first unwatched highlighted. # seed the first rendered ep as finished (id read off the list), # reload, and check the classes. 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, "admin", "adminpw12345") await page.goto(BASE + "/?anime=chainsaw-man-922") await page.wait_for_selector("#eps .ep", timeout=30000) e1_id = await page.evaluate("document.querySelector('#eps .ep').dataset.ep") await page.evaluate( "(id) => fetch('api/history', {method: 'POST'," " headers: {'Content-Type': 'application/json', 'X-Aniweb': '1'}," " body: JSON.stringify({title: 'Chainsaw Man', slug_id: 'chainsaw-man-922'," " season_id: 922, ep_id: +id, lang: 'eng', ep: 1, pos: 2350, dur: 2400})})", e1_id) await page.reload() await page.wait_for_selector("#eps .ep.done", timeout=30000) assert "\u2713" in await page.inner_text("#eps .ep.done") await page.wait_for_selector("#eps .ep.nextup", timeout=10000) assert not errors, errors await browser.close() def test_resume_saved_position(): asyncio.run(_resume_saved_position()) async def _resume_saved_position(): # end-to-end resume: seed a saved position, abort the first # /api/history fetch (a failed boot fetch leaves history = []), and the # page must resume near the saved time — not restart at 0. POSTs are # aborted too, so the dying page's pagehide save can't clobber the seed. 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, "bob", "bobpw12345") await page.goto(BASE + "/?anime=chainsaw-man-922") await page.wait_for_selector("#eps .ep", timeout=30000) e1_id = await page.evaluate("document.querySelector('#eps .ep').dataset.ep") # seed exactly the way savePos writes (same session, same fields) status = await page.evaluate( "(id) => fetch('api/history', {method: 'POST'," \ " headers: {'Content-Type': 'application/json', 'X-Aniweb': '1'}," \ " body: JSON.stringify({title: detail.name, slug_id: detail.slug_id," " season_id: seasonId, ep_id: +id, lang: 'jpn', ep: 1," " pos: 400, dur: 1500})}).then(r => r.status)", e1_id) assert status == 200 state = {"aborted": 0} async def hist_route(route): if route.request.method == "POST": await route.abort() return if state["aborted"] == 0: # the reloaded page's boot fetch state["aborted"] += 1 await route.abort() return await route.continue_() await page.route("**/api/history*", hist_route) await page.reload() # resumes near 400 (keyframe-aligned), not at 0 await page.wait_for_function( "(document.querySelector('video')?.currentTime || 0) > 350", timeout=120000, polling=250) got = await page.evaluate("fetch('api/history?all=1').then(r => r.json())") row = [x for x in got if x["slug_id"] == "chainsaw-man-922" and x["ep_id"] == int(e1_id)] # the row still holds the seeded position (a boot-write of # pos: saved?.pos || 0 used to clobber it when the fetch failed) assert row and row[0]["pos"] == 400, row assert not errors, errors await browser.close() def test_autoplay_blocked_shows_tap_to_play(): asyncio.run(_autoplay_blocked()) async def _autoplay_blocked(): # Simulates the browser's autoplay policy (headless Chromium here allows # autoplay, so the policy is faked): the first, gesture-less play() # rejects; a later user-initiated one goes through. Regression: the # "Loading…" overlay (which only clears on 'playing') used to sit over # the unreachable native controls forever in that state. async with async_playwright() as p: browser = await p.chromium.launch() context = await browser.new_context() await context.add_init_script(""" const origPlay = HTMLMediaElement.prototype.play; HTMLMediaElement.prototype.play = function () { if (window.__firstPlayBlocked) { window.__firstPlayBlocked = false; return Promise.reject(new DOMException('blocked', 'NotAllowedError')); } return origPlay.call(this); }; window.__firstPlayBlocked = true; """) page = await context.new_page() errors = [] page.on("pageerror", lambda e: errors.append(str(e))) await _login(page, "admin", "adminpw12345") await page.goto(BASE + "/?anime=chainsaw-man-922") await page.wait_for_selector("#eps .ep", timeout=30000) # blocked autoplay: the dead "Loading…" must become a tappable overlay. # The fake block only rejects the FIRST play(); the app's 3s safety # net then retriggers doPlay unblocked, so "already playing" is also a # valid end state. 60s: under full-suite load the CDN can be slow. await page.wait_for_function( "(document.querySelector('video')?.currentTime || 0) > 0.5 || " "!!(document.querySelector('.pl-overlay .btn') && " " document.querySelector('.pl-overlay').textContent.includes('Tap to play'))", timeout=60000, polling=POLL) # tapping (the gesture) starts real playback; if buffering swapped the # overlay away in the meantime the video is already playing if await page.query_selector(".pl-overlay .btn"): try: await page.click(".pl-overlay .btn", force=True, timeout=10000) except Exception: pass # a lost race fails loudly at the final wait below await page.wait_for_function( "(document.querySelector('video')?.currentTime || 0) > 0.5", timeout=120000, polling=250) assert not errors, errors await browser.close() def test_stale_history_remap_board(): asyncio.run(_stale_history_remap_board()) async def _stale_history_remap_board(): # provider switch (anidb -> hianime) left rows whose anidb slug 200s on # hianime as a DIFFERENT anime (jujutsu-kaisen-2552 = Burning Kabaddi). # The history page must flag it for remap instead of linking to that show. 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", "X-Aniweb": "1"}, method="POST")) op.open(urllib.request.Request(BASE + "/api/history", data=json.dumps({"title": "Jujutsu Kaisen", "slug_id": "jujutsu-kaisen-2552", "season_id": 2552, "ep_id": 35996, "lang": "jpn", "ep": 1, "pos": 42}).encode(), headers={"Content-Type": "application/json", "X-Aniweb": "1"}, 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_function( "document.querySelector('a.card.remap')", timeout=60000, polling=POLL) stale = page.locator("a.card.remap", has_text="Jujutsu Kaisen") assert "not found" in await stale.inner_text() # the card must not link to the wrong-slug detail page; find-it goes # to the title's search instead assert not await stale.get_attribute("href") await stale.locator("button:has-text('find it')").click(force=True) await page.wait_for_function( "location.search.includes('q=')", timeout=15000, polling=POLL) assert not errors, errors await browser.close()