453 lines
20 KiB
Python
453 lines
20 KiB
Python
"""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()
|