tests: refactor to pytest (conftest session fixture, sync wrappers for async, importorskip for playwright)

requirements.txt

@@ -2,3 +2,7 @@ fastapi
 uvicorn
 curl_cffi
 itsdangerous
+
+# tests
+pytest
+playwright

tests/conftest.py

@@ -0,0 +1,81 @@
+"""Session fixture for the ani-web test suite.
+
+Spins up a throwaway server (127.0.0.1:8399, temp data dir, seeded
+users admin/alice/bob/dave/erin) once per pytest run, tears it down
+afterwards. The real server on :8321 and its data are never touched.
+
+Usage:  .venv/bin/pytest tests/ -q
+"""
+import json
+import os
+import shutil
+import signal
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.request
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).parent.parent
+PORT = 8399
+BASE = f"http://127.0.0.1:{PORT}"
+USERS = (("admin", "adminpw12345"), ("alice", "alicepw12345"), ("bob", "bobpw12345"),
+         ("dave", "davepw12345"), ("erin", "erinpw12345"))
+
+
+def _wait_ready():
+    for _ in range(120):
+        try:
+            urllib.request.urlopen(BASE + "/login", timeout=1)
+            return
+        except Exception:
+            time.sleep(0.5)
+    raise RuntimeError("server did not start")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def server():
+    # fail fast instead of silently testing a leftover server.
+    # SO_REUSEADDR: the suite's own TIME_WAIT sockets (no listener, but
+    # thousands of connections) would otherwise wedge this bind for ~60s.
+    s = socket.socket()
+    s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    try:
+        s.bind(("127.0.0.1", PORT))
+    except OSError:
+        pytest.exit(f"port {PORT} is busy - is a leftover test server running?")
+    finally:
+        s.close()
+
+    data = Path(tempfile.mkdtemp(prefix="aniweb-test-"))
+    os.environ["ANIWEB_DATA"] = str(data)
+    sys.path.insert(0, str(ROOT))
+    from server import users  # noqa: E402
+    for name, pw in USERS:
+        users.add(name, pw)
+    (data / "history").mkdir()
+    (data / "history" / "admin.json").write_text(json.dumps([
+        {"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1, "ep_id": "1",
+         "lang": "eng", "ep": "1", "pos": 10.0, "t": time.time()}]))
+
+    proc = subprocess.Popen(
+        [sys.executable, "-m", "uvicorn", "server.app:app",
+         "--host", "127.0.0.1", "--port", str(PORT)],
+        cwd=ROOT, env=os.environ,
+        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        start_new_session=True)
+    try:
+        _wait_ready()
+        yield
+    finally:
+        os.killpg(proc.pid, signal.SIGTERM)
+        try:
+            proc.wait(timeout=5)
+        except subprocess.TimeoutExpired:
+            os.killpg(proc.pid, signal.SIGKILL)
+            proc.wait()
+        shutil.rmtree(data, ignore_errors=True)

tests/run.py

@@ -1,91 +0,0 @@
-"""run.py — full ani-web test suite.
-
-Spins up a throwaway server (127.0.0.1:8399, temp data dir, seeded
-users admin/alice/bob), runs the API tests and (if playwright is
-installed) the browser tests, then tears everything down. The real
-server on :8321 and its data are never touched.
-
-Usage:  .venv/bin/python tests/run.py
-"""
-import asyncio
-import json
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
-import time
-import urllib.error
-import urllib.request
-from pathlib import Path
-
-ROOT = Path(__file__).parent.parent
-PORT = 8399
-BASE = f"http://127.0.0.1:{PORT}"
-USERS = (("admin", "adminpw12345"), ("alice", "alicepw12345"), ("bob", "bobpw12345"),
-         ("dave", "davepw12345"), ("erin", "erinpw12345"))
-
-
-def _wait_ready():
-    for _ in range(120):
-        try:
-            urllib.request.urlopen(BASE + "/login", timeout=1)
-            return
-        except Exception:
-            time.sleep(0.5)
-    raise SystemExit("server did not start")
-
-
-def main():
-    data = Path(tempfile.mkdtemp(prefix="aniweb-test-"))
-    os.environ["ANIWEB_DATA"] = str(data)
-    sys.path.insert(0, str(ROOT))
-    from server import users
-    for name, pw in USERS:
-        users.add(name, pw)
-    (data / "history").mkdir()
-    (data / "history" / "admin.json").write_text(json.dumps([
-        {"title": "Seed Anime", "slug_id": "seed-1", "season_id": 1, "ep_id": "1",
-         "lang": "eng", "ep": "1", "pos": 10.0, "t": time.time()}]))
-
-    proc = subprocess.Popen(
-        [sys.executable, "-m", "uvicorn", "server.app:app",
-         "--host", "127.0.0.1", "--port", str(PORT)],
-        cwd=ROOT, env=os.environ,
-        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    try:
-        _wait_ready()
-
-        import test_api  # noqa: E402  (tests/ on sys.path via script dir)
-        # order matters: lockout test locks bob in memory, run it last
-        api_tests = [fn for n, fn in sorted(vars(test_api).items())
-                     if n.startswith("test_") and n != "test_lockout_per_user"]
-        for fn in api_tests:
-            print(f"ok api.{fn.__name__}")
-            fn(BASE)
-        print("ok api.test_lockout_per_user")
-        test_api.test_lockout_per_user(BASE)
-
-        try:
-            import test_browser
-        except ImportError:
-            print("SKIP browser tests (pip install playwright && playwright install chromium)")
-        else:
-            print("ok browser (playwright)")
-            asyncio.run(test_browser.test_login_ui())
-            asyncio.run(test_browser.test_video_plays_behind_auth())
-            asyncio.run(test_browser.test_history_page_and_delete())
-            asyncio.run(test_browser.test_stream_failure_retry())
-            asyncio.run(test_browser.test_recent_searches())
-            asyncio.run(test_browser.test_auto_next_countdown())
-            asyncio.run(test_browser.test_episode_watch_state())
-
-        print("ALL TESTS PASSED")
-    finally:
-        proc.terminate()
-        proc.wait()
-        shutil.rmtree(data, ignore_errors=True)
-
-
-if __name__ == "__main__":
-    main()

tests/test_api.py

@@ -1,8 +1,8 @@
-"""HTTP-level tests for ani-web (stdlib only). Run via tests/run.py or
+"""HTTP-level tests for ani-web (stdlib only), run via pytest.
-standalone against a running server: python tests/test_api.py <base_url>
 
-Uses a throwaway server + temp data dir (see run.py) — never touches
+Uses the throwaway server + temp data dir from conftest.py — never
-the real server/data. Seeded users: admin/alice/bob (pw <name>pw12345).
+touches the real server/data. Seeded users: admin/alice/bob/dave/erin
+(pw <name>pw12345).
 """
 import http.cookiejar as _cj
 import json
@@ -187,12 +187,3 @@ def _json(method, path, jar=None, base=BASE):
 def _users_file():
     import os
     return os.path.join(os.environ["ANIWEB_DATA"], "users.json")
-
-
-if __name__ == "__main__":
-    import sys
-    b = sys.argv[1] if len(sys.argv) > 1 else BASE
-    for name, fn in sorted(globals().items()):
-        if name.startswith("test_"):
-            print("ok", name)
-            fn(b)

tests/test_browser.py

@@ -1,16 +1,17 @@
 """Browser tests for ani-web (Playwright, headless Chromium).
 
-Run via tests/run.py or standalone: python tests/test_browser.py <base_url>
+Run via pytest (see conftest.py for the throwaway server). Covers what
-Covers what HTTP tests can't: the login UI, the user chip, and real
+HTTP tests can't: the login UI, the user chip, and real video playback
-video playback behind auth. Needs: pip install playwright && playwright install chromium
+behind auth. Skipped if playwright is not installed.
 """
 import asyncio
 import http.cookiejar
 import json
-import sys
 import urllib.request
 
-from playwright.async_api import async_playwright
+import pytest
+
+async_playwright = pytest.importorskip("playwright.async_api").async_playwright
 
 BASE = "http://127.0.0.1:8399"
 LAUNCH_ARGS = ["--autoplay-policy=no-user-gesture-required"]
@@ -36,7 +37,11 @@ async def _login(page, user, pw):
         timeout=5000, polling=POLL)
 
 
-async def test_login_ui():
+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()
@@ -65,7 +70,6 @@ async def test_login_ui():
         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):
@@ -74,7 +78,11 @@ async def _login_settled(page):
         "document.getElementById('me').textContent.length > 0", timeout=5000, polling=POLL)
 
 
-async def test_video_plays_behind_auth():
+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()
@@ -94,10 +102,13 @@ async def test_video_plays_behind_auth():
         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():
+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))
@@ -129,10 +140,13 @@ async def test_history_page_and_delete():
         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():
+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()
@@ -157,10 +171,13 @@ async def test_stream_failure_retry():
             "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():
+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()
@@ -192,10 +209,13 @@ async def test_recent_searches():
         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():
+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:
@@ -223,10 +243,13 @@ async def test_auto_next_countdown():
         assert "E1" in await page.inner_text(".ep.active")
         assert not errors, errors
         await browser.close()
-    print("ok test_auto_next_countdown")
 
 
-async def test_episode_watch_state():
+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.
@@ -252,16 +275,3 @@ async def test_episode_watch_state():
         await page.wait_for_selector("#eps .ep.nextup", timeout=10000)
         assert not errors, errors
         await browser.close()
-    print("ok test_episode_watch_state")
-
-
-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())
-    asyncio.run(test_episode_watch_state())