237 lines
10 KiB
Python
237 lines
10 KiB
Python
"""Browser tests for ani-web (Playwright, headless Chromium).
|
|
|
|
Run via tests/run.py or standalone: python tests/test_browser.py <base_url>
|
|
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")
|
|
|
|
|
|
async def test_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"}, method="POST"))
|
|
op.open(urllib.request.Request(BASE + "/api/searches",
|
|
data=json.dumps({"q": "chainsaw"}).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, "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()
|
|
print("ok test_recent_searches")
|
|
|
|
|
|
async def test_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()
|
|
print("ok test_auto_next_countdown")
|
|
|
|
|
|
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())
|
|
asyncio.run(test_recent_searches())
|
|
asyncio.run(test_auto_next_countdown())
|