test: add hermetic e2e support services

backend/src/test/e2e/runtime/test_fake_qb_contract.py

@@ -0,0 +1,154 @@
+"""Contract tests for the hermetic qBittorrent HTTP fake."""
+
+import runpy
+import threading
+from pathlib import Path
+
+import httpx
+import pytest
+
+pytestmark = pytest.mark.e2e
+
+
+SERVER_PATH = Path(__file__).resolve().parents[5] / "e2e" / "fake-qb" / "server.py"
+SERVER_MODULE = runpy.run_path(str(SERVER_PATH))
+create_server = SERVER_MODULE["create_server"]
+MATCHING_HASH = SERVER_MODULE["MATCHING_HASH"]
+NEIGHBOUR_HASH = SERVER_MODULE["NEIGHBOUR_HASH"]
+DELETE_FAILURE_HASH = SERVER_MODULE["DELETE_FAILURE_HASH"]
+
+
+@pytest.fixture
+def fake_qb_url():
+    server = create_server("127.0.0.1", 0)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    host, port = server.server_address
+    try:
+        yield f"http://{host}:{port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def login(client: httpx.Client) -> None:
+    response = client.post(
+        "/api/v2/auth/login",
+        data={"username": "admin", "password": "adminadmin"},
+    )
+    assert response.status_code == 200
+    assert response.text == "Ok."
+
+
+def test_qb_endpoints_and_successful_delete_update_state(fake_qb_url):
+    with httpx.Client(base_url=fake_qb_url) as client:
+        login(client)
+
+        assert client.get("/api/v2/app/version").text == "v5.2.3"
+        preferences = client.get("/api/v2/app/preferences").json()
+        assert preferences["save_path"] == "D:\\Downloads\\Bangumi"
+        assert {key: preferences[key] for key in ("dht", "pex", "lsd", "upnp")} == {
+            "dht": False,
+            "pex": False,
+            "lsd": False,
+            "upnp": False,
+        }
+
+        torrents = client.get("/api/v2/torrents/info").json()
+        by_hash = {torrent["hash"]: torrent for torrent in torrents}
+        assert by_hash[MATCHING_HASH]["save_path"].endswith("Season 1\\")
+        assert by_hash[NEIGHBOUR_HASH]["save_path"].endswith("Season 10")
+
+        response = client.post(
+            "/api/v2/torrents/delete",
+            data={"hashes": MATCHING_HASH, "deleteFiles": "true"},
+        )
+        assert response.status_code == 200
+
+        state = client.get("/__admin/state").json()
+        remaining = {torrent["hash"] for torrent in state["torrents"]}
+        assert MATCHING_HASH not in remaining
+        assert NEIGHBOUR_HASH in remaining
+
+        assert client.post("/api/v2/auth/logout").status_code == 200
+        assert client.get("/api/v2/torrents/info").status_code == 403
+
+
+def test_forced_delete_failure_is_atomic(fake_qb_url):
+    with httpx.Client(base_url=fake_qb_url) as client:
+        login(client)
+
+        response = client.post(
+            "/api/v2/torrents/delete",
+            data={
+                "hashes": f"{MATCHING_HASH}|{DELETE_FAILURE_HASH}",
+                "deleteFiles": "false",
+            },
+        )
+        assert response.status_code == 500
+
+        state = client.get("/__admin/state").json()
+        remaining = {torrent["hash"] for torrent in state["torrents"]}
+        assert MATCHING_HASH in remaining
+        assert DELETE_FAILURE_HASH in remaining
+
+
+def test_request_journal_preserves_form_shape_and_redacts_secrets(fake_qb_url):
+    with httpx.Client(base_url=fake_qb_url) as client:
+        login(client)
+
+        response = client.post(
+            "/api/v2/unsupported?token=query-secret&plain=query-value",
+            data={
+                "hashes": [MATCHING_HASH, NEIGHBOUR_HASH],
+                "password": "form-secret",
+                "auth_token": "form-token",
+                "plain": "form-value",
+            },
+            headers={
+                "Authorization": "Bearer header-secret",
+                "X-Api-Token": "header-token",
+            },
+        )
+        assert response.status_code == 501
+
+        journal = client.get("/__admin/requests").json()["requests"]
+        login_request = journal[0]
+        assert login_request["form"] == {
+            "username": ["admin"],
+            "password": ["[REDACTED]"],
+        }
+
+        unsupported = journal[-1]
+        assert unsupported["query"] == {
+            "token": ["[REDACTED]"],
+            "plain": ["query-value"],
+        }
+        assert unsupported["form"] == {
+            "hashes": [MATCHING_HASH, NEIGHBOUR_HASH],
+            "password": ["[REDACTED]"],
+            "auth_token": ["[REDACTED]"],
+            "plain": ["form-value"],
+        }
+        assert unsupported["headers"]["Authorization"] == "[REDACTED]"
+        assert unsupported["headers"]["Cookie"] == "[REDACTED]"
+        assert unsupported["headers"]["X-Api-Token"] == "[REDACTED]"
+
+
+def test_admin_reset_restores_presets_and_clears_journal(fake_qb_url):
+    with httpx.Client(base_url=fake_qb_url) as client:
+        login(client)
+        client.post(
+            "/api/v2/torrents/delete",
+            data={"hashes": MATCHING_HASH, "deleteFiles": "true"},
+        )
+
+        response = client.post("/__admin/reset")
+        assert response.status_code == 200
+        assert response.json() == {"ok": True}
+
+        state = client.get("/__admin/state").json()
+        hashes = {torrent["hash"] for torrent in state["torrents"]}
+        assert {MATCHING_HASH, NEIGHBOUR_HASH, DELETE_FAILURE_HASH} <= hashes
+        assert client.get("/__admin/requests").json() == {"requests": []}

backend/src/test/e2e/runtime/test_mock_upstream_contract.py

@@ -0,0 +1,285 @@
+"""Contract tests for the hermetic mock-upstream HTTP service."""
+
+import importlib.util
+import json
+import shutil
+import threading
+from pathlib import Path
+
+import httpx
+import pytest
+
+pytestmark = pytest.mark.e2e
+
+REPO_ROOT = Path(__file__).resolve().parents[5]
+SERVER_PATH = REPO_ROOT / "e2e" / "mock-upstream" / "server.py"
+
+
+def _load_create_server():
+    spec = importlib.util.spec_from_file_location("e2e_mock_upstream", SERVER_PATH)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Cannot load mock-upstream from {SERVER_PATH}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.create_server
+
+
+@pytest.fixture
+def fixture_root(tmp_path: Path) -> Path:
+    root = tmp_path / "fixtures"
+    shutil.copytree(REPO_ROOT / "e2e" / "fixtures", root)
+    images = root / "images"
+    images.mkdir()
+    (images / "poster.jpg").write_bytes(b"mock-poster-bytes")
+    return root
+
+
+@pytest.fixture
+def upstream(fixture_root: Path):
+    server = _load_create_server()("127.0.0.1", 0, fixture_root)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    host, port = server.server_address[:2]
+    with httpx.Client(base_url=f"http://{host}:{port}", timeout=5.0) as client:
+        yield client, fixture_root
+    server.shutdown()
+    server.server_close()
+    thread.join(timeout=5)
+    assert not thread.is_alive()
+
+
+def _activate(client: httpx.Client, name: str) -> None:
+    response = client.put(f"/__admin/scenario/{name}")
+    assert response.status_code == 200
+    assert response.json()["scenario"] == name
+
+
+def test_health_and_admin_lifecycle(upstream):
+    client, _ = upstream
+
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}
+
+    response = client.post("/__admin/reset")
+    assert response.status_code == 200
+    assert response.json() == {"ok": True}
+    assert client.get("/__admin/requests").json() == {"requests": []}
+
+    _activate(client, "localized-tv-zh")
+    state = client.get("/__admin/state").json()
+    assert state == {
+        "scenario": "localized-tv-zh",
+        "notifications": [],
+        "request_count": 0,
+    }
+
+    missing = client.put("/__admin/scenario/does-not-exist")
+    assert missing.status_code == 404
+    assert client.get("/__admin/state").json()["scenario"] == "localized-tv-zh"
+
+    reset = client.post("/__admin/reset")
+    assert reset.status_code == 200
+    assert client.get("/__admin/state").json() == {
+        "scenario": None,
+        "notifications": [],
+        "request_count": 0,
+    }
+
+
+def test_tmdb_tv_movie_and_retry_scenarios(upstream):
+    client, _ = upstream
+    client.post("/__admin/reset")
+
+    _activate(client, "localized-tv-zh")
+    tv = client.get(
+        "/tmdb/3/search/tv",
+        params={
+            "query": "Localized Show",
+            "language": "zh-CN",
+            "api_key": "tmdb-secret",
+        },
+    )
+    assert tv.status_code == 200
+    assert tv.json()["results"][0]["name"] == "本地化动画"
+
+    wrong_language = client.get(
+        "/tmdb/3/search/tv",
+        params={"query": "Localized Show", "language": "en-US"},
+    )
+    assert wrong_language.status_code == 200
+    assert wrong_language.json() == {"results": []}
+
+    detail = client.get(
+        "/tmdb/3/tv/1001",
+        params={"language": "zh-CN", "api_key": "tmdb-secret"},
+    )
+    assert detail.status_code == 200
+    assert detail.json()["genres"] == [{"id": 16, "name": "Animation"}]
+    assert (
+        client.get("/tmdb/3/tv/1001", params={"language": "en-US"}).status_code == 404
+    )
+
+    season = client.get(
+        "/tmdb/3/tv/1001/season/1",
+        params={"language": "zh-CN", "api_key": "tmdb-secret"},
+    )
+    assert season.status_code == 200
+    assert season.json()["episodes"][0] == {
+        "episode_number": 1,
+        "air_date": "2026-07-01",
+    }
+    assert (
+        client.get("/tmdb/3/tv/1001/season/1", params={"language": "en-US"}).status_code
+        == 404
+    )
+
+    _activate(client, "localized-movie-jp")
+    movie = client.get(
+        "/tmdb/3/search/movie",
+        params={"query": "Localized Movie", "language": "ja-JP"},
+    )
+    assert movie.status_code == 200
+    assert movie.json()["results"][0]["title"] == "ローカライズ映画"
+    assert client.get(
+        "/tmdb/3/search/tv",
+        params={"query": "Localized Movie", "language": "ja-JP"},
+    ).json() == {"results": []}
+
+    _activate(client, "localized-retry-jp")
+    initial = client.get(
+        "/tmdb/3/search/tv",
+        params={"query": "Retry Localized Show", "language": "ja-JP"},
+    )
+    retried = client.get(
+        "/tmdb/3/search/tv",
+        params={"query": "RetryLocalizedShow", "language": "ja-JP"},
+    )
+    assert initial.json() == {"results": []}
+    assert retried.json()["results"][0]["name"] == "再試行アニメ"
+
+
+def test_static_routes_head_and_single_ranges(upstream):
+    client, fixture_root = upstream
+    media = (fixture_root / "files" / "tiny-media.mkv").read_bytes()
+
+    rss = client.get("/rss/tmdb-tv.xml")
+    assert rss.status_code == 200
+    assert rss.headers["content-type"].startswith("application/rss+xml")
+    assert b"Localized Show" in rss.content
+
+    image = client.get("/images/poster.jpg")
+    assert image.status_code == 200
+    assert image.content == b"mock-poster-bytes"
+
+    player = client.get("/player")
+    assert player.status_code == 200
+    assert player.headers["content-type"].startswith("text/html")
+
+    torrent = client.get("/torrents/tiny-media.torrent")
+    assert torrent.status_code == 200
+    assert torrent.headers["content-type"].startswith("application/x-bittorrent")
+
+    head = client.head("/files/tiny-media.mkv")
+    assert head.status_code == 200
+    assert head.content == b""
+    assert int(head.headers["content-length"]) == len(media)
+    assert head.headers["accept-ranges"] == "bytes"
+
+    partial = client.get("/files/tiny-media.mkv", headers={"Range": "bytes=7-23"})
+    assert partial.status_code == 206
+    assert partial.content == media[7:24]
+    assert partial.headers["content-range"] == f"bytes 7-23/{len(media)}"
+    assert partial.headers["content-length"] == "17"
+
+    suffix = client.get("/files/tiny-media.mkv", headers={"Range": "bytes=-11"})
+    assert suffix.status_code == 206
+    assert suffix.content == media[-11:]
+
+    unsatisfiable = client.get(
+        "/files/tiny-media.mkv", headers={"Range": f"bytes={len(media)}-"}
+    )
+    assert unsatisfiable.status_code == 416
+    assert unsatisfiable.headers["content-range"] == f"bytes */{len(media)}"
+    assert unsatisfiable.content == b""
+
+    multiple = client.get("/files/tiny-media.mkv", headers={"Range": "bytes=0-1,4-5"})
+    assert multiple.status_code == 416
+
+
+def test_notification_capture_and_journal_redact_secrets(upstream):
+    client, _ = upstream
+    client.post("/__admin/reset")
+
+    response = client.post(
+        "/notifications/webhook",
+        params={"token": "query-secret", "message": "visible-query"},
+        headers={
+            "Authorization": "Bearer header-secret",
+            "Cookie": "session=cookie-secret",
+            "X-Api-Key": "header-api-secret",
+        },
+        json={
+            "message": "visible-body",
+            "password": "body-password",
+            "nested": {"api_token": "body-token", "value": "kept"},
+        },
+    )
+    assert response.status_code == 204
+
+    form_response = client.post(
+        "/notifications/form",
+        data={"password": "form-password", "message": "visible-form"},
+    )
+    assert form_response.status_code == 204
+
+    requests = client.get("/__admin/requests").json()["requests"]
+    webhook = requests[0]
+    assert webhook["path"] == "/notifications/webhook"
+    assert webhook["query"] == {
+        "token": ["[REDACTED]"],
+        "message": ["visible-query"],
+    }
+    assert webhook["headers"]["authorization"] == "[REDACTED]"
+    assert webhook["headers"]["cookie"] == "[REDACTED]"
+    assert webhook["headers"]["x-api-key"] == "[REDACTED]"
+    assert webhook["body"] == {
+        "message": "visible-body",
+        "password": "[REDACTED]",
+        "nested": {"api_token": "[REDACTED]", "value": "kept"},
+    }
+    assert requests[1]["body"] == {
+        "password": ["[REDACTED]"],
+        "message": ["visible-form"],
+    }
+
+    state = client.get("/__admin/state").json()
+    assert len(state["notifications"]) == 2
+    serialized = json.dumps({"requests": requests, "state": state}, ensure_ascii=False)
+    for secret in (
+        "query-secret",
+        "header-secret",
+        "cookie-secret",
+        "header-api-secret",
+        "body-password",
+        "body-token",
+        "form-password",
+    ):
+        assert secret not in serialized
+    assert "visible-body" in serialized
+    assert "visible-form" in serialized
+
+
+def test_unknown_route_returns_501_and_is_journaled(upstream):
+    client, _ = upstream
+    client.post("/__admin/reset")
+
+    response = client.get("/not-a-contract-route", params={"api_key": "must-not-leak"})
+
+    assert response.status_code == 501
+    journal = client.get("/__admin/requests").json()["requests"]
+    assert len(journal) == 1
+    assert journal[0]["path"] == "/not-a-contract-route"
+    assert journal[0]["query"]["api_key"] == ["[REDACTED]"]
+    assert journal[0]["response_status"] == 501
+    assert "must-not-leak" not in json.dumps(journal)

backend/src/test/test_e2e_fixture_generation.py

@@ -0,0 +1,99 @@
+"""Contract tests for deterministic hermetic E2E fixture generation."""
+
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import re
+from pathlib import Path
+from types import ModuleType
+from urllib.parse import urlparse
+
+REPO_ROOT = Path(__file__).resolve().parents[3]
+FIXTURE_ROOT = REPO_ROOT / "e2e" / "fixtures"
+GENERATOR_PATH = REPO_ROOT / "e2e" / "scripts" / "generate_fixtures.py"
+
+
+def _load_generator() -> ModuleType:
+    spec = importlib.util.spec_from_file_location(
+        "e2e_generate_fixtures", GENERATOR_PATH
+    )
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def _bdecode(data: bytes, index: int = 0):
+    token = data[index : index + 1]
+    if token == b"i":
+        end = data.index(b"e", index)
+        return int(data[index + 1 : end]), end + 1
+    if token == b"l":
+        result = []
+        index += 1
+        while data[index : index + 1] != b"e":
+            value, index = _bdecode(data, index)
+            result.append(value)
+        return result, index + 1
+    if token == b"d":
+        result = {}
+        index += 1
+        while data[index : index + 1] != b"e":
+            key, index = _bdecode(data, index)
+            value, index = _bdecode(data, index)
+            result[key] = value
+        return result, index + 1
+    colon = data.index(b":", index)
+    length = int(data[index:colon])
+    start = colon + 1
+    end = start + length
+    return data[start:end], end
+
+
+def test_generated_binary_fixtures_match_committed_bytes(tmp_path: Path):
+    generator = _load_generator()
+    generated_root = tmp_path / "fixtures"
+
+    generator.write_fixtures(generated_root)
+
+    for relative_path in generator.GENERATED_PATHS:
+        assert (generated_root / relative_path).read_bytes() == (
+            FIXTURE_ROOT / relative_path
+        ).read_bytes()
+
+    checksums = json.loads((FIXTURE_ROOT / "checksums.json").read_text())
+    assert checksums == {
+        str(path): hashlib.sha256((FIXTURE_ROOT / path).read_bytes()).hexdigest()
+        for path in generator.GENERATED_PATHS
+    }
+
+
+def test_torrent_contains_only_local_web_seed_and_valid_piece_hashes():
+    torrent, end = _bdecode(
+        (FIXTURE_ROOT / "torrents" / "tiny-media.torrent").read_bytes()
+    )
+    media = (FIXTURE_ROOT / "files" / "tiny-media.mkv").read_bytes()
+
+    assert end == len((FIXTURE_ROOT / "torrents" / "tiny-media.torrent").read_bytes())
+    assert torrent[b"url-list"] == (b"http://mock-upstream:18888/files/tiny-media.mkv")
+    info = torrent[b"info"]
+    assert info[b"name"] == b"Tiny Show - 01 [E2E].mkv"
+    assert info[b"length"] == len(media) == 64 * 1024
+    assert info[b"piece length"] == 16 * 1024
+    assert info[b"pieces"] == b"".join(
+        hashlib.sha1(media[offset : offset + info[b"piece length"]]).digest()
+        for offset in range(0, len(media), info[b"piece length"])
+    )
+
+
+def test_text_fixtures_reference_only_hermetic_hosts():
+    allowed_hosts = {"mock-upstream", "127.0.0.1", "localhost"}
+    urls: list[str] = []
+    for pattern in ("*.xml", "*.json", "*.html"):
+        for path in FIXTURE_ROOT.rglob(pattern):
+            urls.extend(re.findall(r"https?://[^\s\"'<>]+", path.read_text()))
+
+    assert urls
+    assert {urlparse(url).hostname for url in urls} <= allowed_hosts

backend/src/test/test_e2e_stack_contract.py

@@ -0,0 +1,365 @@
+"""Pure contract tests for the hermetic E2E stack infrastructure."""
+
+from __future__ import annotations
+
+import importlib.util
+import re
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[3]
+E2E = ROOT / "e2e"
+PYTHON_IMAGE = (
+    "python:3.13.14-alpine3.23@sha256:"
+    "9fdbf2e3e82628351513560b121e2ee6ce31cac212be9e070c5a5e2769fb5e76"
+)
+QB_IMAGE = (
+    "linuxserver/qbittorrent:5.2.3@sha256:"
+    "2e074403c7b72e6d89cee3d0d41a47f7b5708c6a9e5316f3958c90765cbe12ce"
+)
+
+
+def _load_script(name: str):
+    path = E2E / "scripts" / f"{name}.py"
+    spec = importlib.util.spec_from_file_location(f"e2e_{name}", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_compose_files_are_isolated_and_use_random_loopback_ports():
+    browser = (E2E / "compose" / "browser.yml").read_text()
+    downloader = (E2E / "compose" / "downloader.yml").read_text()
+
+    assert "internal: true" in browser
+    assert "container_name" not in browser + downloader
+    assert ":latest" not in browser + downloader
+    assert '"127.0.0.1::7892"' in browser
+    assert '"127.0.0.1::18888"' in browser
+    assert '"127.0.0.1::8080"' in browser
+    assert '"127.0.0.1::8080"' in downloader
+    assert not re.search(r"127\.0\.0\.1:\d+:\d+", browser + downloader)
+    assert QB_IMAGE in downloader
+
+
+def test_support_service_images_are_pinned_and_dependency_free():
+    for relative in (
+        "mock-upstream/Dockerfile",
+        "fake-qb/Dockerfile",
+    ):
+        dockerfile = (E2E / relative).read_text()
+        assert f"FROM {PYTHON_IMAGE}" in dockerfile
+        assert "pip install" not in dockerfile
+        assert "apk add" not in dockerfile
+        assert "server.py" in dockerfile
+
+
+def test_qbittorrent_fixture_has_fixed_credentials_and_disables_peer_discovery():
+    config = (E2E / "fixtures/qbittorrent/qBittorrent.conf").read_text()
+
+    assert "WebUI\\Username=admin" in config
+    assert (
+        'WebUI\\Password_PBKDF2="@ByteArray('
+        "ARQ77eY1NUZaQsuDHbIMCA==:"
+        "0WMRkYTUWVT9wVvdDtHAjU9b3b7uB8NR1Gur2hmQCvCDpm39Q+PsJRJPaCU51dEiz+"
+        'dTzh8qbPsL8WkFljQYFQ==)"'
+    ) in config
+    assert "DHTEnabled=false" in config
+    assert "PeXEnabled=false" in config
+    assert "LSDEnabled=false" in config
+    assert "PortForwardingEnabled=false" in config
+    assert "CheckForUpdates=false" in config
+
+
+def test_build_context_stages_dist_and_version_without_touching_source(tmp_path):
+    build = _load_script("build_test_image")
+    repo = tmp_path / "repo"
+    context = tmp_path / "context"
+    dist = tmp_path / "dist"
+    tracked = (
+        Path(".dockerignore"),
+        Path("Dockerfile"),
+        Path("entrypoint.sh"),
+        Path("backend/pyproject.toml"),
+        Path("backend/uv.lock"),
+        Path("backend/src/main.py"),
+        Path("backend/src/module/example.py"),
+    )
+    for relative in tracked:
+        target = repo / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(f"fixture:{relative}\n")
+    (dist / "assets").mkdir(parents=True)
+    (dist / "index.html").write_text("<main>e2e</main>")
+    (dist / "assets/app.js").write_text("console.log('e2e')")
+
+    build.prepare_context(
+        repo,
+        dist,
+        context,
+        version="3.3.999-e2e.1",
+        tracked_files=tracked,
+    )
+
+    assert (context / "backend/src/dist/index.html").is_file()
+    assert (context / "backend/src/dist/assets/app.js").is_file()
+    assert (context / "backend/src/module/__version__.py").read_text() == (
+        'VERSION = "3.3.999-e2e.1"\n'
+    )
+    assert not (repo / "backend/src/dist").exists()
+    assert not (repo / "backend/src/module/__version__.py").exists()
+
+
+def test_build_command_uses_only_the_temporary_context(tmp_path):
+    build = _load_script("build_test_image")
+
+    command = build.build_command(
+        tmp_path / "context",
+        image="auto-bangumi:e2e",
+        version="3.3.999-e2e.1",
+    )
+
+    assert command[:2] == ["docker", "build"]
+    assert ["--build-arg", "VERSION=3.3.999-e2e.1"] == command[2:4]
+    assert ["--tag", "auto-bangumi:e2e"] == command[4:6]
+    assert command[-1] == str(tmp_path / "context")
+
+
+def test_stack_builds_unique_compose_commands_and_environment(tmp_path):
+    stack_module = _load_script("stack")
+    first_name = stack_module.unique_project_name("browser")
+    second_name = stack_module.unique_project_name("browser")
+    assert first_name != second_name
+    assert re.fullmatch(r"[a-z0-9][a-z0-9-]{0,62}", first_name)
+
+    stack = stack_module.Stack(
+        profile="downloader",
+        repo_root=ROOT,
+        work_dir=tmp_path / "work",
+        project_name="ab-e2e-downloader-test",
+    )
+    command = stack.compose_command("config")
+
+    assert command[:4] == [
+        "docker",
+        "compose",
+        "--project-name",
+        "ab-e2e-downloader-test",
+    ]
+    assert str(E2E / "compose/browser.yml") in command
+    assert str(E2E / "compose/downloader.yml") in command
+    assert command[-1] == "config"
+    assert stack.environment["AB_E2E_CONFIG_DIR"] == str(
+        (tmp_path / "work/config").resolve()
+    )
+    assert stack.environment["AB_E2E_DOWNLOAD_DIR"] == str(
+        (tmp_path / "work/downloads").resolve()
+    )
+    assert stack.environment["AB_E2E_LOG_FILE"] == str(
+        (tmp_path / "work/logs/log.txt").resolve()
+    )
+    assert stack.environment["AB_E2E_USERNAME"] == "testadmin"
+    assert stack.environment["AB_E2E_PASSWORD"] == "testpassword123"
+    stack._publish_urls(
+        {"app": 40001, "mock-upstream": 40002, "fake-qb": 40003, "qbittorrent": 40004}
+    )
+    assert stack.environment["AB_E2E_REAL_QB_URL"] == "http://127.0.0.1:40004"
+    assert stack.environment["AB_E2E_QB_URL"] == "http://127.0.0.1:40004"
+
+
+@pytest.mark.parametrize(
+    ("output", "expected"),
+    [
+        ("127.0.0.1:49152", 49152),
+        ("0.0.0.0:32768", 32768),
+        ("[::1]:60000", 60000),
+    ],
+)
+def test_stack_parses_dynamic_compose_ports(output, expected):
+    stack_module = _load_script("stack")
+    assert stack_module.parse_published_port(output) == expected
+
+
+def test_qb_runtime_contract_checks_credentials_and_deterministic_preferences():
+    stack_module = _load_script("stack")
+    preferences = {
+        "dht": False,
+        "pex": False,
+        "lsd": False,
+        "upnp": False,
+        "save_path": "/downloads/Bangumi",
+        "rss_processing_enabled": False,
+        "rss_auto_downloading_enabled": False,
+    }
+
+    stack_module.validate_qb_preferences(preferences)
+    with pytest.raises(AssertionError, match="unsafe or non-deterministic"):
+        stack_module.validate_qb_preferences({**preferences, "dht": True})
+
+
+def test_qb_runtime_contract_accepts_login_204_and_checks_preferences(monkeypatch):
+    stack_module = _load_script("stack")
+    preferences = {
+        "dht": False,
+        "pex": False,
+        "lsd": False,
+        "upnp": False,
+        "save_path": "/downloads/Bangumi",
+        "rss_processing_enabled": False,
+        "rss_auto_downloading_enabled": False,
+    }
+
+    class Response:
+        def __init__(self, status, payload):
+            self.status = status
+            self.payload = payload
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return False
+
+        def read(self):
+            return self.payload
+
+    class Opener:
+        def __init__(self):
+            self.calls = []
+
+        def open(self, request, timeout):
+            self.calls.append((request, timeout))
+            if not isinstance(request, str):
+                return Response(204, b"")
+            return Response(200, __import__("json").dumps(preferences).encode())
+
+    opener = Opener()
+    monkeypatch.setattr(
+        stack_module.urllib.request,
+        "build_opener",
+        lambda *_handlers: opener,
+    )
+
+    assert stack_module.verify_qb_runtime("http://127.0.0.1:12345") == preferences
+    assert len(opener.calls) == 2
+
+
+def test_run_with_stack_always_stops_after_command_failure():
+    stack_module = _load_script("stack")
+
+    class FakeStack:
+        started = False
+        stopped = False
+        environment = {"AB_E2E_BASE_URL": "http://127.0.0.1:1"}
+
+        def start(self):
+            self.started = True
+
+        def stop(self, *, collect=True):
+            assert collect is True
+            self.stopped = True
+
+    fake = FakeStack()
+
+    def fail(_command, _environment):
+        raise RuntimeError("boom")
+
+    with pytest.raises(RuntimeError, match="boom"):
+        stack_module.run_with_stack(fake, ["pytest"], runner=fail)
+
+    assert fake.started is True
+    assert fake.stopped is True
+
+
+def test_run_with_stack_handles_sigterm_and_restores_handlers(monkeypatch):
+    stack_module = _load_script("stack")
+    installed = {}
+    signal_calls = []
+    monkeypatch.setattr(
+        stack_module.signal,
+        "getsignal",
+        lambda signum: f"previous-{signum}",
+    )
+
+    def install(signum, handler):
+        installed[signum] = handler
+        signal_calls.append((signum, handler))
+
+    monkeypatch.setattr(stack_module.signal, "signal", install)
+
+    class FakeStack:
+        environment = {}
+
+        def __init__(self):
+            self.stop_count = 0
+
+        def start(self):
+            return None
+
+        def stop(self, *, collect=True):
+            assert collect is True
+            self.stop_count += 1
+
+    fake = FakeStack()
+
+    def terminate(_command, _environment):
+        installed[stack_module.signal.SIGTERM](stack_module.signal.SIGTERM, None)
+
+    with pytest.raises(SystemExit):
+        stack_module.run_with_stack(fake, ["pytest"], runner=terminate)
+
+    assert fake.stop_count >= 1
+    assert signal_calls[-2:] == [
+        (
+            stack_module.signal.SIGINT,
+            f"previous-{stack_module.signal.SIGINT}",
+        ),
+        (
+            stack_module.signal.SIGTERM,
+            f"previous-{stack_module.signal.SIGTERM}",
+        ),
+    ]
+
+
+def test_artifact_redaction_removes_credentials_and_tokens():
+    artifacts = _load_script("collect_artifacts")
+    raw = (
+        "password=adminadmin Authorization: Bearer api-secret "
+        "Cookie: SID=session-secret token=raw-token"
+    )
+
+    redacted = artifacts.redact_text(raw, secrets=("adminadmin",))
+
+    for secret in ("adminadmin", "api-secret", "session-secret", "raw-token"):
+        assert secret not in redacted
+    assert redacted.count("[REDACTED]") >= 4
+
+
+def test_static_audit_accepts_the_real_e2e_tree():
+    audit = _load_script("audit_sources")
+    assert audit.audit_repository(ROOT) == []
+
+
+def test_static_audit_reports_fixed_ports_floating_images_and_blind_waits(tmp_path):
+    audit = _load_script("audit_sources")
+    (tmp_path / "e2e/compose").mkdir(parents=True)
+    (tmp_path / "e2e/scripts").mkdir(parents=True)
+    (tmp_path / "e2e/compose/bad.yml").write_text(
+        "services:\n"
+        "  bad:\n"
+        "    container_name: fixed\n"
+        "    image: example/tool:latest\n"
+        '    ports: ["127.0.0.1:8080:8080"]\n'
+    )
+    (tmp_path / "e2e/scripts/bad.py").write_text("time.sleep(10)\n")
+
+    messages = [finding.message for finding in audit.audit_repository(tmp_path)]
+
+    assert any("container_name" in message for message in messages)
+    assert any("floating image" in message for message in messages)
+    assert any("fixed host port" in message for message in messages)
+    assert any("blind sleep" in message for message in messages)

e2e/compose/browser.yml

@@ -0,0 +1,115 @@
+services:
+  app:
+    image: ${AB_E2E_APP_IMAGE:-auto-bangumi:e2e}
+    pull_policy: never
+    environment:
+      TZ: UTC
+      PUID: ${AB_E2E_PUID:-1000}
+      PGID: ${AB_E2E_PGID:-1000}
+      UMASK: "022"
+      AB_WEBUI_PORT: "7892"
+      AB_DOWNLOADER_HOST: fake-qb:8080
+      AB_DOWNLOADER_USERNAME: admin
+      AB_DOWNLOADER_PASSWORD: adminadmin
+      AB_DOWNLOAD_PATH: /downloads/Bangumi
+      AB_INTERVAL_TIME: "3600"
+      AB_RENAME_FREQ: "1"
+      AB_DEBUG_MODE: "true"
+      AB_DEV_NO_AUTH: "0"
+    depends_on:
+      mock-upstream:
+        condition: service_healthy
+      fake-qb:
+        condition: service_healthy
+    ports:
+      - "127.0.0.1::7892"
+    volumes:
+      - type: bind
+        source: ${AB_E2E_CONFIG_DIR:?AB_E2E_CONFIG_DIR must be set}
+        target: /app/config
+      - type: bind
+        source: ${AB_E2E_DATA_DIR:?AB_E2E_DATA_DIR must be set}
+        target: /app/data
+      - type: bind
+        source: ${AB_E2E_DOWNLOAD_DIR:?AB_E2E_DOWNLOAD_DIR must be set}
+        target: /downloads
+      - type: bind
+        source: ${AB_E2E_LOG_FILE:?AB_E2E_LOG_FILE must be set}
+        target: /app/data/log.txt
+    healthcheck:
+      test: ["CMD", "wget", "-qO-", "http://127.0.0.1:7892/health"]
+      interval: 2s
+      timeout: 3s
+      retries: 30
+      start_period: 5s
+    stop_grace_period: 20s
+    networks:
+      - e2e
+
+  mock-upstream:
+    image: ${AB_E2E_MOCK_IMAGE:-auto-bangumi-e2e/mock-upstream:local}
+    pull_policy: never
+    build:
+      context: ../..
+      dockerfile: e2e/mock-upstream/Dockerfile
+    environment:
+      HOST: 0.0.0.0
+      PORT: "18888"
+      FIXTURES_ROOT: /fixtures
+    init: true
+    read_only: true
+    ports:
+      - "127.0.0.1::18888"
+    volumes:
+      - type: bind
+        source: ${AB_E2E_FIXTURES_DIR:?AB_E2E_FIXTURES_DIR must be set}
+        target: /fixtures
+        read_only: true
+    tmpfs:
+      - /tmp
+    healthcheck:
+      test:
+        - CMD
+        - python
+        - -c
+        - import urllib.request; urllib.request.urlopen('http://127.0.0.1:18888/health', timeout=2)
+      interval: 2s
+      timeout: 3s
+      retries: 20
+      start_period: 2s
+    networks:
+      - e2e
+
+  fake-qb:
+    image: ${AB_E2E_FAKE_QB_IMAGE:-auto-bangumi-e2e/fake-qb:local}
+    pull_policy: never
+    build:
+      context: ../..
+      dockerfile: e2e/fake-qb/Dockerfile
+    environment:
+      HOST: 0.0.0.0
+      PORT: "8080"
+      QB_USERNAME: admin
+      QB_PASSWORD: adminadmin
+    init: true
+    read_only: true
+    ports:
+      - "127.0.0.1::8080"
+    tmpfs:
+      - /tmp
+    healthcheck:
+      test:
+        - CMD
+        - python
+        - -c
+        - import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/__admin/state', timeout=2)
+      interval: 2s
+      timeout: 3s
+      retries: 20
+      start_period: 2s
+    networks:
+      - e2e
+
+networks:
+  e2e:
+    internal: true

e2e/compose/downloader.yml

@@ -0,0 +1,36 @@
+services:
+  app:
+    environment:
+      AB_DOWNLOADER_HOST: qbittorrent:8080
+      AB_DOWNLOADER_USERNAME: admin
+      AB_DOWNLOADER_PASSWORD: adminadmin
+    depends_on:
+      qbittorrent:
+        condition: service_healthy
+
+  qbittorrent:
+    image: linuxserver/qbittorrent:5.2.3@sha256:2e074403c7b72e6d89cee3d0d41a47f7b5708c6a9e5316f3958c90765cbe12ce
+    environment:
+      TZ: UTC
+      PUID: ${AB_E2E_PUID:-1000}
+      PGID: ${AB_E2E_PGID:-1000}
+      WEBUI_PORT: "8080"
+      TORRENTING_PORT: "6881"
+    ports:
+      - "127.0.0.1::8080"
+    volumes:
+      - type: bind
+        source: ${AB_E2E_QB_CONFIG_DIR:?AB_E2E_QB_CONFIG_DIR must be set}
+        target: /config
+      - type: bind
+        source: ${AB_E2E_DOWNLOAD_DIR:?AB_E2E_DOWNLOAD_DIR must be set}
+        target: /downloads
+    healthcheck:
+      test: ["CMD", "curl", "-fsS", "http://127.0.0.1:8080/"]
+      interval: 2s
+      timeout: 3s
+      retries: 30
+      start_period: 10s
+    stop_grace_period: 20s
+    networks:
+      - e2e

e2e/fake-qb/Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.13.14-alpine3.23@sha256:9fdbf2e3e82628351513560b121e2ee6ce31cac212be9e070c5a5e2769fb5e76
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    HOST=0.0.0.0 \
+    PORT=8080
+
+WORKDIR /app
+COPY e2e/fake-qb/server.py /app/server.py
+
+EXPOSE 8080
+HEALTHCHECK --interval=2s --timeout=3s --start-period=2s --retries=20 \
+    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8080/__admin/state', timeout=2)"]
+
+CMD ["python", "/app/server.py"]

e2e/fake-qb/server.py

@@ -0,0 +1,441 @@
+#!/usr/bin/env python3
+"""Small hermetic qBittorrent Web API fake used by end-to-end tests."""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import secrets
+import threading
+from http.cookies import SimpleCookie
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+from urllib.parse import parse_qs, urlsplit
+
+MATCHING_HASH = "a" * 40
+NEIGHBOUR_HASH = "b" * 40
+DELETE_FAILURE_HASH = "f" * 40
+REDACTED = "[REDACTED]"
+
+_SENSITIVE_FRAGMENTS = (
+    "password",
+    "passwd",
+    "secret",
+    "token",
+    "auth",
+    "cookie",
+    "sid",
+)
+
+_PRESET_TORRENTS = (
+    {
+        "hash": MATCHING_HASH,
+        "name": "Target Anime - 01",
+        "save_path": "D:\\Downloads\\Bangumi\\Test Anime (2024)\\Season 1\\",
+        "category": "Bangumi",
+        "tags": "ab:1",
+        "state": "pausedUP",
+        "progress": 1.0,
+        "size": 65_536,
+    },
+    {
+        "hash": NEIGHBOUR_HASH,
+        "name": "Nearby Anime - 10",
+        "save_path": "D:\\Downloads\\Bangumi\\Test Anime (2024)\\Season 10",
+        "category": "Bangumi",
+        "tags": "ab:10",
+        "state": "pausedUP",
+        "progress": 1.0,
+        "size": 65_536,
+    },
+    {
+        "hash": DELETE_FAILURE_HASH,
+        "name": "Forced Delete Failure",
+        "save_path": "D:\\Downloads\\Bangumi\\Delete Failure\\Season 1",
+        "category": "Bangumi",
+        "tags": "ab:failure",
+        "state": "pausedUP",
+        "progress": 1.0,
+        "size": 65_536,
+    },
+)
+
+_PRESET_PREFERENCES = {
+    "save_path": "D:\\Downloads\\Bangumi",
+    "temp_path_enabled": False,
+    "web_ui_port": 8080,
+    "dht": False,
+    "pex": False,
+    "lsd": False,
+    "upnp": False,
+}
+
+
+def _is_sensitive(name: str) -> bool:
+    normalized = name.lower().replace("-", "_")
+    return any(fragment in normalized for fragment in _SENSITIVE_FRAGMENTS)
+
+
+def _redact_object(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            key: REDACTED if _is_sensitive(str(key)) else _redact_object(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [_redact_object(item) for item in value]
+    return value
+
+
+def _redact_values(name: str, values: list[str]) -> list[str]:
+    if _is_sensitive(name):
+        return [REDACTED for _ in values]
+    if name.lower() == "json":
+        sanitized: list[str] = []
+        for value in values:
+            try:
+                parsed = json.loads(value)
+            except (TypeError, ValueError):
+                sanitized.append(value)
+            else:
+                sanitized.append(
+                    json.dumps(_redact_object(parsed), separators=(",", ":"))
+                )
+        return sanitized
+    return list(values)
+
+
+def _redact_mapping(values: dict[str, list[str]]) -> dict[str, list[str]]:
+    return {key: _redact_values(key, items) for key, items in values.items()}
+
+
+class FakeQbState:
+    def __init__(self, username: str, password: str) -> None:
+        self.username = username
+        self.password = password
+        self.lock = threading.RLock()
+        self.reset()
+
+    def reset(self) -> None:
+        with self.lock:
+            self.torrents = {
+                torrent["hash"]: copy.deepcopy(torrent) for torrent in _PRESET_TORRENTS
+            }
+            self.preferences = copy.deepcopy(_PRESET_PREFERENCES)
+            self.sessions: set[str] = set()
+            self.requests: list[dict[str, Any]] = []
+
+    def record(
+        self,
+        *,
+        method: str,
+        path: str,
+        query: dict[str, list[str]],
+        form: dict[str, list[str]],
+        headers: dict[str, str],
+    ) -> None:
+        sanitized_headers = {
+            key: REDACTED if _is_sensitive(key) else value
+            for key, value in headers.items()
+        }
+        request = {
+            "method": method,
+            "path": path,
+            "query": _redact_mapping(query),
+            "form": _redact_mapping(form),
+            "headers": sanitized_headers,
+        }
+        with self.lock:
+            self.requests.append(request)
+
+    def snapshot(self) -> dict[str, Any]:
+        with self.lock:
+            return {
+                "torrents": copy.deepcopy(list(self.torrents.values())),
+                "preferences": _redact_object(copy.deepcopy(self.preferences)),
+                "session_count": len(self.sessions),
+            }
+
+
+class FakeQbServer(ThreadingHTTPServer):
+    daemon_threads = True
+    allow_reuse_address = True
+
+    def __init__(
+        self,
+        address: tuple[str, int],
+        username: str,
+        password: str,
+    ) -> None:
+        self.state = FakeQbState(username, password)
+        super().__init__(address, FakeQbHandler)
+
+
+class FakeQbHandler(BaseHTTPRequestHandler):
+    server: FakeQbServer
+    protocol_version = "HTTP/1.1"
+
+    def log_message(self, format: str, *args: Any) -> None:
+        return
+
+    def _send_bytes(
+        self,
+        status: int,
+        payload: bytes,
+        content_type: str,
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(payload)))
+        for key, value in (headers or {}).items():
+            self.send_header(key, value)
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def _send_json(self, status: int, body: Any) -> None:
+        payload = json.dumps(body, separators=(",", ":")).encode("utf-8")
+        self._send_bytes(status, payload, "application/json; charset=utf-8")
+
+    def _send_text(
+        self,
+        status: int,
+        body: str = "",
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        self._send_bytes(
+            status,
+            body.encode("utf-8"),
+            "text/plain; charset=utf-8",
+            headers,
+        )
+
+    def _read_form(self) -> dict[str, list[str]]:
+        length = int(self.headers.get("Content-Length", "0") or 0)
+        payload = self.rfile.read(length) if length else b""
+        content_type = self.headers.get("Content-Type", "")
+        if not content_type.startswith("application/x-www-form-urlencoded"):
+            return {}
+        return parse_qs(payload.decode("utf-8"), keep_blank_values=True)
+
+    def _record(
+        self,
+        method: str,
+        path: str,
+        query: dict[str, list[str]],
+        form: dict[str, list[str]],
+    ) -> None:
+        self.server.state.record(
+            method=method,
+            path=path,
+            query=query,
+            form=form,
+            headers=dict(self.headers.items()),
+        )
+
+    def _session_id(self) -> str | None:
+        cookie_header = self.headers.get("Cookie")
+        if not cookie_header:
+            return None
+        cookies = SimpleCookie()
+        try:
+            cookies.load(cookie_header)
+        except Exception:
+            return None
+        sid = cookies.get("SID")
+        return sid.value if sid else None
+
+    def _authenticated(self) -> bool:
+        session_id = self._session_id()
+        with self.server.state.lock:
+            return session_id is not None and session_id in self.server.state.sessions
+
+    def _require_auth(self) -> bool:
+        if self._authenticated():
+            return True
+        self._send_text(403, "Forbidden")
+        return False
+
+    @staticmethod
+    def _one(form: dict[str, list[str]], key: str, default: str = "") -> str:
+        values = form.get(key)
+        return values[0] if values else default
+
+    def do_GET(self) -> None:
+        target = urlsplit(self.path)
+        query = parse_qs(target.query, keep_blank_values=True)
+
+        if target.path == "/__admin/state":
+            self._send_json(200, self.server.state.snapshot())
+            return
+        if target.path == "/__admin/requests":
+            with self.server.state.lock:
+                requests = copy.deepcopy(self.server.state.requests)
+            self._send_json(200, {"requests": requests})
+            return
+
+        self._record("GET", target.path, query, {})
+        if target.path.startswith("/api/v2/") and not self._require_auth():
+            return
+
+        if target.path == "/api/v2/app/version":
+            self._send_text(200, "v5.2.3")
+            return
+        if target.path == "/api/v2/app/preferences":
+            with self.server.state.lock:
+                preferences = copy.deepcopy(self.server.state.preferences)
+            self._send_json(200, preferences)
+            return
+        if target.path == "/api/v2/torrents/info":
+            self._send_json(200, self._filtered_torrents(query))
+            return
+
+        self._send_json(
+            501,
+            {"error": "unsupported endpoint", "method": "GET", "path": target.path},
+        )
+
+    def _filtered_torrents(self, query: dict[str, list[str]]) -> list[dict[str, Any]]:
+        with self.server.state.lock:
+            torrents = copy.deepcopy(list(self.server.state.torrents.values()))
+
+        hashes = query.get("hashes", [""])[0]
+        if hashes:
+            selected_hashes = set(hashes.split("|"))
+            torrents = [item for item in torrents if item["hash"] in selected_hashes]
+        category = query.get("category", [""])[0]
+        if category:
+            torrents = [item for item in torrents if item["category"] == category]
+        tag = query.get("tag", [""])[0]
+        if tag:
+            torrents = [
+                item
+                for item in torrents
+                if tag in {value.strip() for value in item["tags"].split(",")}
+            ]
+        status_filter = query.get("filter", [""])[0]
+        if status_filter == "completed":
+            torrents = [item for item in torrents if item["progress"] >= 1]
+        return torrents
+
+    def do_POST(self) -> None:
+        target = urlsplit(self.path)
+        query = parse_qs(target.query, keep_blank_values=True)
+        form = self._read_form()
+
+        if target.path == "/__admin/reset":
+            self.server.state.reset()
+            self._send_json(200, {"ok": True})
+            return
+
+        self._record("POST", target.path, query, form)
+
+        if target.path == "/api/v2/auth/login":
+            self._login(form)
+            return
+        if target.path.startswith("/api/v2/") and not self._require_auth():
+            return
+        if target.path == "/api/v2/auth/logout":
+            self._logout()
+            return
+        if target.path == "/api/v2/app/setPreferences":
+            self._set_preferences(form)
+            return
+        if target.path == "/api/v2/torrents/delete":
+            self._delete_torrents(form)
+            return
+
+        self._send_json(
+            501,
+            {"error": "unsupported endpoint", "method": "POST", "path": target.path},
+        )
+
+    def _login(self, form: dict[str, list[str]]) -> None:
+        username = self._one(form, "username")
+        password = self._one(form, "password")
+        if (
+            username != self.server.state.username
+            or password != self.server.state.password
+        ):
+            self._send_text(200, "Fails.")
+            return
+        session_id = secrets.token_hex(16)
+        with self.server.state.lock:
+            self.server.state.sessions.add(session_id)
+        self._send_text(
+            200,
+            "Ok.",
+            {"Set-Cookie": f"SID={session_id}; HttpOnly; SameSite=Strict; Path=/"},
+        )
+
+    def _logout(self) -> None:
+        session_id = self._session_id()
+        if session_id:
+            with self.server.state.lock:
+                self.server.state.sessions.discard(session_id)
+        self._send_text(
+            200,
+            "Ok.",
+            {"Set-Cookie": "SID=; Max-Age=0; HttpOnly; SameSite=Strict; Path=/"},
+        )
+
+    def _set_preferences(self, form: dict[str, list[str]]) -> None:
+        try:
+            preferences = json.loads(self._one(form, "json"))
+        except (TypeError, ValueError):
+            self._send_text(400, "Invalid preferences")
+            return
+        if not isinstance(preferences, dict):
+            self._send_text(400, "Invalid preferences")
+            return
+        with self.server.state.lock:
+            self.server.state.preferences.update(preferences)
+        self._send_text(200)
+
+    def _delete_torrents(self, form: dict[str, list[str]]) -> None:
+        hashes_value = self._one(form, "hashes")
+        if not hashes_value:
+            self._send_text(400, "Missing hashes")
+            return
+        with self.server.state.lock:
+            hashes = (
+                set(self.server.state.torrents)
+                if hashes_value == "all"
+                else set(hashes_value.split("|"))
+            )
+            if DELETE_FAILURE_HASH in hashes:
+                self._send_text(500, "Forced delete failure")
+                return
+            for torrent_hash in hashes:
+                self.server.state.torrents.pop(torrent_hash, None)
+        self._send_text(200)
+
+
+def create_server(
+    host: str = "127.0.0.1",
+    port: int = 0,
+    *,
+    username: str = "admin",
+    password: str = "adminadmin",
+) -> FakeQbServer:
+    """Create an unstarted server; callers own ``serve_forever`` and shutdown."""
+    return FakeQbServer((host, port), username, password)
+
+
+def main() -> None:
+    host = os.environ.get("HOST", "0.0.0.0")
+    port = int(os.environ.get("PORT", "8080"))
+    username = os.environ.get("QB_USERNAME", "admin")
+    password = os.environ.get("QB_PASSWORD", "adminadmin")
+    server = create_server(host, port, username=username, password=password)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()

e2e/fixtures/checksums.json

@@ -0,0 +1,4 @@
+{
+  "files/tiny-media.mkv": "7daca2095d0438260fa849183dfc67faa459fdf4936e1bc91eec6b281b27e4c2",
+  "torrents/tiny-media.torrent": "3ef510e2e5f87bd3fcef4a9ffdf16330d87e163a820866600384eb0863f86615"
+}

e2e/fixtures/player/index.html

@@ -0,0 +1,10 @@
+<!doctype html>
+<html lang="en">
+  <head>
+    <meta charset="utf-8" />
+    <title>AutoBangumi E2E Player</title>
+  </head>
+  <body>
+    <main id="e2e-player-marker">Hermetic player fixture loaded</main>
+  </body>
+</html>

e2e/fixtures/qbittorrent/qBittorrent.conf

@@ -0,0 +1,43 @@
+[Application]
+FileLogger\Age=1
+FileLogger\AgeType=1
+FileLogger\Backup=true
+FileLogger\DeleteOld=true
+FileLogger\Enabled=true
+FileLogger\MaxSizeBytes=1048576
+FileLogger\Path=/config/qBittorrent/logs
+
+[BitTorrent]
+Session\AddTorrentStopped=false
+Session\DHTEnabled=false
+Session\DefaultSavePath=/downloads/Bangumi
+Session\LSDEnabled=false
+Session\PeXEnabled=false
+Session\Port=6881
+Session\PortForwardingEnabled=false
+Session\QueueingSystemEnabled=false
+Session\TempPath=/downloads/.incomplete
+Session\TempPathEnabled=true
+
+[LegalNotice]
+Accepted=true
+
+[Preferences]
+Connection\PortRangeMin=6881
+Connection\UPnP=false
+Downloads\SavePath=/downloads/Bangumi
+Downloads\TempPath=/downloads/.incomplete
+Downloads\TempPathEnabled=true
+General\CheckForUpdates=false
+General\Locale=en
+RSS\AutoDownloaderEnabled=false
+RSS\Enabled=false
+WebUI\Address=*
+WebUI\AlternativeUIEnabled=false
+WebUI\CSRFProtection=false
+WebUI\HostHeaderValidation=false
+WebUI\LocalHostAuth=false
+WebUI\Password_PBKDF2="@ByteArray(ARQ77eY1NUZaQsuDHbIMCA==:0WMRkYTUWVT9wVvdDtHAjU9b3b7uB8NR1Gur2hmQCvCDpm39Q+PsJRJPaCU51dEiz+dTzh8qbPsL8WkFljQYFQ==)"
+WebUI\Port=8080
+WebUI\ServerDomains=*
+WebUI\Username=admin

e2e/fixtures/rss/offset-review.xml

@@ -0,0 +1,20 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Offset Review Controls</title>
+    <link>http://mock-upstream:18888/rss/offset-review.xml</link>
+    <description>False-positive and true-positive offset fixtures</description>
+    <item>
+      <title>[SubsPlease] Haibara-kun no Tsuyokute Seishun New Game - 06v2 (1080p) [E931DD98].mkv</title>
+      <link>http://mock-upstream:18888/items/haibara-06v2</link>
+      <pubDate>Wed, 01 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+    <item>
+      <title>[SubsPlease] Haibara-kun no Tsuyokute Seishun New Game - 13 (1080p) [ABCDEF12].mkv</title>
+      <link>http://mock-upstream:18888/items/haibara-13</link>
+      <pubDate>Wed, 08 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+  </channel>
+</rss>

e2e/fixtures/rss/tiny-download.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rss version="2.0">
+  <channel>
+    <title>AutoBangumi Tiny Download</title>
+    <link>http://mock-upstream:18888/rss/tiny-download.xml</link>
+    <description>Hermetic qBittorrent web-seed fixture</description>
+    <item>
+      <title>[E2E] Tiny Show - 01 [1080p][CHS].mkv</title>
+      <link>http://mock-upstream:18888/items/tiny-show-01</link>
+      <pubDate>Wed, 01 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+  </channel>
+</rss>

e2e/fixtures/rss/tmdb-movie.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Localized Movie Search</title>
+    <link>http://mock-upstream:18888/rss/tmdb-movie.xml</link>
+    <description>TMDB movie language fixture</description>
+    <item>
+      <title>[E2E] Localized Movie Movie [1080p][CHS].mkv</title>
+      <link>http://mock-upstream:18888/items/localized-movie</link>
+      <pubDate>Wed, 01 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+  </channel>
+</rss>

e2e/fixtures/rss/tmdb-retry.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Localized Retry Search</title>
+    <link>http://mock-upstream:18888/rss/tmdb-retry.xml</link>
+    <description>TMDB whitespace-removal retry fixture</description>
+    <item>
+      <title>[E2E] Retry Localized Show - 01 [1080p][CHS].mkv</title>
+      <link>http://mock-upstream:18888/items/retry-localized-show-01</link>
+      <pubDate>Wed, 01 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+  </channel>
+</rss>

e2e/fixtures/rss/tmdb-tv.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="utf-8"?>
+<rss version="2.0">
+  <channel>
+    <title>Localized TV Search</title>
+    <link>http://mock-upstream:18888/rss/tmdb-tv.xml</link>
+    <description>TMDB television language fixture</description>
+    <item>
+      <title>[E2E] Localized Show - 01 [1080p][CHS].mkv</title>
+      <link>http://mock-upstream:18888/items/localized-show-01</link>
+      <pubDate>Wed, 01 Jul 2026 12:00:00 GMT</pubDate>
+      <enclosure url="http://mock-upstream:18888/torrents/tiny-media.torrent" type="application/x-bittorrent" length="0" />
+    </item>
+  </channel>
+</rss>

e2e/fixtures/tmdb/scenarios.json

@@ -0,0 +1,105 @@
+{
+  "scenarios": {
+    "localized-tv-zh": {
+      "search": {
+        "kind": "tv",
+        "query": "Localized Show",
+        "language": "zh-CN",
+        "results": [{ "id": 1001, "name": "本地化动画" }]
+      },
+      "detail": {
+        "id": 1001,
+        "name": "本地化动画",
+        "original_name": "Localized Show",
+        "first_air_date": "2026-07-01",
+        "poster_path": null,
+        "status": "Returning Series",
+        "genres": [{ "id": 16, "name": "Animation" }],
+        "seasons": [
+          {
+            "season_number": 1,
+            "name": "第 1 季",
+            "air_date": "2026-07-01",
+            "episode_count": 12,
+            "poster_path": null
+          }
+        ]
+      }
+    },
+    "localized-tv-jp": {
+      "search": {
+        "kind": "tv",
+        "query": "Localized Show",
+        "language": "ja-JP",
+        "results": [{ "id": 1001, "name": "ローカライズアニメ" }]
+      },
+      "detail": {
+        "id": 1001,
+        "name": "ローカライズアニメ",
+        "original_name": "Localized Show",
+        "first_air_date": "2026-07-01",
+        "poster_path": null,
+        "status": "Returning Series",
+        "genres": [{ "id": 16, "name": "Animation" }],
+        "seasons": [
+          {
+            "season_number": 1,
+            "name": "第 1 季",
+            "air_date": "2026-07-01",
+            "episode_count": 12,
+            "poster_path": null
+          }
+        ]
+      }
+    },
+    "localized-movie-jp": {
+      "search": {
+        "kind": "movie",
+        "query": "Localized Movie",
+        "language": "ja-JP",
+        "results": [
+          {
+            "id": 2001,
+            "title": "ローカライズ映画",
+            "original_title": "Localized Movie",
+            "release_date": "2026-07-01",
+            "poster_path": null
+          }
+        ]
+      }
+    },
+    "localized-retry-jp": {
+      "search": {
+        "kind": "tv",
+        "query": "RetryLocalizedShow",
+        "initial_query": "Retry Localized Show",
+        "language": "ja-JP",
+        "results": [{ "id": 3001, "name": "再試行アニメ" }]
+      },
+      "detail": {
+        "id": 3001,
+        "name": "再試行アニメ",
+        "original_name": "Retry Localized Show",
+        "first_air_date": "2026-07-01",
+        "poster_path": null,
+        "status": "Returning Series",
+        "genres": [{ "id": 16, "name": "Animation" }],
+        "seasons": [
+          {
+            "season_number": 1,
+            "name": "第 1 季",
+            "air_date": "2026-07-01",
+            "episode_count": 12,
+            "poster_path": null
+          }
+        ]
+      }
+    }
+  },
+  "season": {
+    "episodes": [
+      { "episode_number": 1, "air_date": "2026-07-01" },
+      { "episode_number": 2, "air_date": "2026-07-08" }
+    ]
+  }
+}

e2e/fixtures/torrents/tiny-media.torrent

@@ -0,0 +1 @@
+d10:created by24:AutoBangumi hermetic E2E13:creation datei0e4:infod6:lengthi65536e4:name24:Tiny Show - 01 [E2E].mkv12:piece lengthi16384e6:pieces80:€ËœC

e2e/mock-upstream/Dockerfile

@@ -0,0 +1,18 @@
+FROM python:3.13.14-alpine3.23@sha256:9fdbf2e3e82628351513560b121e2ee6ce31cac212be9e070c5a5e2769fb5e76
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    HOST=0.0.0.0 \
+    PORT=18888 \
+    FIXTURES_ROOT=/fixtures
+
+WORKDIR /app
+COPY e2e/mock-upstream/server.py /app/server.py
+COPY e2e/fixtures /fixtures
+
+EXPOSE 18888
+HEALTHCHECK --interval=2s --timeout=3s --start-period=2s --retries=20 \
+    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:18888/health', timeout=2)"]
+
+ENTRYPOINT ["python", "/app/server.py"]
+CMD ["--host", "0.0.0.0", "--port", "18888", "--fixture-root", "/fixtures"]

e2e/mock-upstream/server.py

@@ -0,0 +1,616 @@
+"""Hermetic HTTP upstream for AutoBangumi end-to-end tests.
+
+The implementation intentionally uses only the Python standard library so the
+same server can run directly from a checkout or in a tiny test container.
+"""
+
+from __future__ import annotations
+
+import argparse
+import copy
+import json
+import mimetypes
+import os
+import re
+import threading
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+from urllib.parse import parse_qs, unquote, urlsplit
+
+REDACTED = "[REDACTED]"
+_SENSITIVE_NAMES = (
+    "authorization",
+    "cookie",
+    "password",
+    "passwd",
+    "secret",
+    "api_key",
+    "api-key",
+    "apikey",
+    "access_token",
+    "refresh_token",
+    "token",
+)
+_TEXT_SECRET_RE = re.compile(
+    r"(?i)(authorization|cookie|password|passwd|secret|api[_-]?key|"
+    r"access[_-]?token|refresh[_-]?token|token)(\s*[=:]\s*)"
+    r'([^&\s,;}"]+)'
+)
+
+
+def _is_sensitive(name: str) -> bool:
+    lowered = name.lower()
+    return any(part in lowered for part in _SENSITIVE_NAMES)
+
+
+def _redact(value: Any) -> Any:
+    if isinstance(value, dict):
+        return {
+            str(key): REDACTED if _is_sensitive(str(key)) else _redact(item)
+            for key, item in value.items()
+        }
+    if isinstance(value, list):
+        return [_redact(item) for item in value]
+    return value
+
+
+def _redact_named_values(values: dict[str, Any]) -> dict[str, Any]:
+    def redact_secret(value: Any) -> Any:
+        if isinstance(value, list):
+            return [REDACTED for _ in value]
+        return REDACTED
+
+    return {
+        key: redact_secret(value) if _is_sensitive(key) else _redact(value)
+        for key, value in values.items()
+    }
+
+
+def _content_type(path: Path, route_prefix: str) -> str:
+    if route_prefix == "/rss/":
+        return "application/rss+xml; charset=utf-8"
+    if route_prefix == "/torrents/":
+        return "application/x-bittorrent"
+    if path.suffix.lower() == ".mkv":
+        return "video/x-matroska"
+    guessed, _ = mimetypes.guess_type(path.name)
+    return guessed or "application/octet-stream"
+
+
+def _parse_range(value: str, size: int) -> tuple[int, int]:
+    if size <= 0 or not value.startswith("bytes=") or "," in value:
+        raise ValueError("unsupported byte range")
+    spec = value[6:].strip()
+    if "-" not in spec:
+        raise ValueError("malformed byte range")
+    start_text, end_text = spec.split("-", 1)
+    try:
+        if not start_text:
+            suffix_length = int(end_text)
+            if suffix_length <= 0:
+                raise ValueError("empty suffix range")
+            start = max(size - suffix_length, 0)
+            end = size - 1
+        else:
+            start = int(start_text)
+            if start < 0 or start >= size:
+                raise ValueError("range starts outside resource")
+            end = size - 1 if not end_text else min(int(end_text), size - 1)
+            if end < start:
+                raise ValueError("range end precedes start")
+    except ValueError as exc:
+        raise ValueError("malformed byte range") from exc
+    return start, end
+
+
+class _MockState:
+    def __init__(self, fixture_root: Path):
+        scenario_file = fixture_root / "tmdb" / "scenarios.json"
+        scenario_data = json.loads(scenario_file.read_text(encoding="utf-8"))
+        self.scenarios: dict[str, dict[str, Any]] = scenario_data["scenarios"]
+        self.season: dict[str, Any] = scenario_data["season"]
+        self.lock = threading.RLock()
+        self.scenario_name: str | None = None
+        self.requests: list[dict[str, Any]] = []
+        self.notifications: list[dict[str, Any]] = []
+
+    def reset(self) -> None:
+        with self.lock:
+            self.scenario_name = None
+            self.requests.clear()
+            self.notifications.clear()
+
+    def activate(self, name: str) -> bool:
+        with self.lock:
+            if name not in self.scenarios:
+                return False
+            self.scenario_name = name
+            return True
+
+    def scenario(self) -> dict[str, Any] | None:
+        with self.lock:
+            if self.scenario_name is None:
+                return None
+            return copy.deepcopy(self.scenarios[self.scenario_name])
+
+    def append_request(self, request: dict[str, Any]) -> dict[str, Any]:
+        with self.lock:
+            self.requests.append(request)
+        return request
+
+    def set_response_status(self, request: dict[str, Any], status: int) -> None:
+        with self.lock:
+            request["response_status"] = status
+
+    def append_notification(self, request: dict[str, Any]) -> None:
+        notification = {
+            key: copy.deepcopy(value)
+            for key, value in request.items()
+            if key != "response_status"
+        }
+        with self.lock:
+            self.notifications.append(notification)
+
+    def request_snapshot(self) -> list[dict[str, Any]]:
+        with self.lock:
+            return copy.deepcopy(self.requests)
+
+    def state_snapshot(self) -> dict[str, Any]:
+        with self.lock:
+            return {
+                "scenario": self.scenario_name,
+                "notifications": copy.deepcopy(self.notifications),
+                "request_count": len(self.requests),
+            }
+
+
+class _MockUpstreamServer(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(self, address: tuple[str, int], fixture_root: Path):
+        super().__init__(address, _MockUpstreamHandler)
+        self.fixture_root = fixture_root
+        self.mock_state = _MockState(fixture_root)
+
+
+class _MockUpstreamHandler(BaseHTTPRequestHandler):
+    protocol_version = "HTTP/1.1"
+    server: _MockUpstreamServer
+
+    def log_message(self, format: str, *args: Any) -> None:
+        del format, args
+
+    def do_GET(self) -> None:
+        self._handle_request(head_only=False)
+
+    def do_HEAD(self) -> None:
+        self._handle_request(head_only=True)
+
+    def do_POST(self) -> None:
+        self._handle_request(head_only=False)
+
+    def do_PUT(self) -> None:
+        self._handle_request(head_only=False)
+
+    def _read_body(self) -> Any:
+        try:
+            length = int(self.headers.get("Content-Length", "0"))
+        except ValueError:
+            length = 0
+        raw = self.rfile.read(length) if length > 0 else b""
+        if not raw:
+            return None
+
+        content_type = self.headers.get("Content-Type", "").split(";", 1)[0].lower()
+        if content_type == "application/json":
+            try:
+                return _redact(json.loads(raw.decode("utf-8")))
+            except (UnicodeDecodeError, json.JSONDecodeError):
+                return "<invalid JSON>"
+        if content_type == "application/x-www-form-urlencoded":
+            try:
+                form = parse_qs(raw.decode("utf-8"), keep_blank_values=True)
+            except UnicodeDecodeError:
+                return f"<{len(raw)} bytes>"
+            return _redact_named_values(form)
+        if content_type.startswith("text/"):
+            try:
+                text = raw.decode("utf-8")
+            except UnicodeDecodeError:
+                return f"<{len(raw)} bytes>"
+            return _TEXT_SECRET_RE.sub(rf"\1\2{REDACTED}", text)
+        return f"<{len(raw)} bytes>"
+
+    def _journal(self, parsed, body: Any) -> dict[str, Any]:
+        query = parse_qs(parsed.query, keep_blank_values=True)
+        headers = {key.lower(): value for key, value in self.headers.items()}
+        request = {
+            "method": self.command,
+            "path": unquote(parsed.path),
+            "query": _redact_named_values(query),
+            "headers": _redact_named_values(headers),
+            "body": body,
+            "response_status": None,
+        }
+        return self.server.mock_state.append_request(request)
+
+    def _handle_request(self, *, head_only: bool) -> None:
+        parsed = urlsplit(self.path)
+        body = self._read_body()
+        if self._handle_admin(parsed, body, head_only=head_only):
+            return
+
+        journal = self._journal(parsed, body)
+        path = unquote(parsed.path)
+        if path == "/health":
+            if self.command not in {"GET", "HEAD"}:
+                self._send_method_not_allowed(journal, "GET, HEAD", head_only)
+            else:
+                self._send_json(journal, 200, {"status": "ok"}, head_only)
+            return
+
+        if path.startswith("/tmdb/"):
+            if self.command not in {"GET", "HEAD"}:
+                self._send_method_not_allowed(journal, "GET, HEAD", head_only)
+            else:
+                self._handle_tmdb(parsed, journal, head_only=head_only)
+            return
+
+        static_route = self._static_route(path)
+        if static_route is not None:
+            if self.command not in {"GET", "HEAD"}:
+                self._send_method_not_allowed(journal, "GET, HEAD", head_only)
+            else:
+                route_prefix, directory, relative = static_route
+                self._send_static(
+                    journal,
+                    route_prefix,
+                    directory,
+                    relative,
+                    head_only=head_only,
+                )
+            return
+
+        if path.startswith("/notifications/"):
+            if self.command != "POST":
+                self._send_method_not_allowed(journal, "POST", head_only)
+            else:
+                self.server.mock_state.append_notification(journal)
+                self._send_empty(journal, 204)
+            return
+
+        self._send_json(
+            journal,
+            501,
+            {"detail": f"Unexpected mock-upstream request: {self.command} {path}"},
+            head_only,
+        )
+
+    def _handle_admin(self, parsed, body: Any, *, head_only: bool) -> bool:
+        path = unquote(parsed.path)
+        state = self.server.mock_state
+        if path == "/__admin/reset":
+            if self.command != "POST":
+                self._send_unjournaled_json(405, {"detail": "Method not allowed"})
+            else:
+                state.reset()
+                self._send_unjournaled_json(200, {"ok": True})
+            return True
+        if path == "/__admin/requests":
+            if self.command not in {"GET", "HEAD"}:
+                self._send_unjournaled_json(405, {"detail": "Method not allowed"})
+            else:
+                self._send_unjournaled_json(
+                    200, {"requests": state.request_snapshot()}, head_only=head_only
+                )
+            return True
+        if path == "/__admin/state":
+            if self.command not in {"GET", "HEAD"}:
+                self._send_unjournaled_json(405, {"detail": "Method not allowed"})
+            else:
+                self._send_unjournaled_json(
+                    200, state.state_snapshot(), head_only=head_only
+                )
+            return True
+
+        scenario_match = re.fullmatch(r"/__admin/scenario/([^/]+)", path)
+        if scenario_match:
+            if self.command not in {"PUT", "POST"}:
+                self._send_unjournaled_json(405, {"detail": "Method not allowed"})
+            else:
+                name = scenario_match.group(1)
+                if state.activate(name):
+                    self._send_unjournaled_json(200, {"scenario": name})
+                else:
+                    self._send_unjournaled_json(
+                        404, {"detail": f"Unknown scenario: {name}"}
+                    )
+            return True
+        if path == "/__admin/scenario" and self.command in {"PUT", "POST"}:
+            name = body.get("name") if isinstance(body, dict) else None
+            if isinstance(name, str) and state.activate(name):
+                self._send_unjournaled_json(200, {"scenario": name})
+            else:
+                self._send_unjournaled_json(
+                    404, {"detail": f"Unknown scenario: {name}"}
+                )
+            return True
+        return False
+
+    def _handle_tmdb(self, parsed, journal: dict[str, Any], *, head_only: bool) -> None:
+        scenario = self.server.mock_state.scenario()
+        if scenario is None:
+            self._send_json(
+                journal, 409, {"detail": "No active TMDB scenario"}, head_only
+            )
+            return
+
+        path = unquote(parsed.path)
+        query = parse_qs(parsed.query, keep_blank_values=True)
+        search_match = re.fullmatch(r"/tmdb/3/search/(tv|movie)", path)
+        if search_match:
+            expected = scenario.get("search", {})
+            kind = search_match.group(1)
+            supplied_query = (query.get("query") or [""])[0]
+            supplied_language = (query.get("language") or [""])[0]
+            expected_query = expected.get("query")
+            initial_query = expected.get("initial_query")
+            matches = (
+                expected.get("kind") == kind
+                and supplied_language == expected.get("language")
+                and supplied_query == expected_query
+            )
+            if supplied_query == initial_query:
+                matches = False
+            results = expected.get("results", []) if matches else []
+            self._send_json(journal, 200, {"results": results}, head_only)
+            return
+
+        season_match = re.fullmatch(r"/tmdb/3/tv/(\d+)/season/(\d+)", path)
+        if season_match:
+            detail = scenario.get("detail")
+            if not self._detail_matches(
+                detail,
+                query,
+                int(season_match.group(1)),
+                scenario.get("search", {}).get("language"),
+            ):
+                self._send_json(
+                    journal, 404, {"detail": "TV season not found"}, head_only
+                )
+                return
+            season_number = int(season_match.group(2))
+            known_seasons = {
+                item.get("season_number") for item in detail.get("seasons", [])
+            }
+            if season_number not in known_seasons:
+                self._send_json(
+                    journal, 404, {"detail": "TV season not found"}, head_only
+                )
+                return
+            self._send_json(journal, 200, self.server.mock_state.season, head_only)
+            return
+
+        detail_match = re.fullmatch(r"/tmdb/3/tv/(\d+)", path)
+        if detail_match:
+            detail = scenario.get("detail")
+            if not self._detail_matches(
+                detail,
+                query,
+                int(detail_match.group(1)),
+                scenario.get("search", {}).get("language"),
+            ):
+                self._send_json(
+                    journal, 404, {"detail": "TV show not found"}, head_only
+                )
+                return
+            self._send_json(journal, 200, detail, head_only)
+            return
+
+        self._send_json(
+            journal,
+            501,
+            {"detail": f"Unexpected TMDB request: {path}"},
+            head_only,
+        )
+
+    @staticmethod
+    def _detail_matches(
+        detail: dict[str, Any] | None,
+        query: dict[str, list[str]],
+        requested_id: int,
+        expected_language: str | None,
+    ) -> bool:
+        if detail is None or detail.get("id") != requested_id:
+            return False
+        supplied_language = (query.get("language") or [""])[0]
+        return bool(expected_language) and supplied_language == expected_language
+
+    def _static_route(self, path: str) -> tuple[str, Path, str] | None:
+        root = self.server.fixture_root
+        for route_prefix, directory_name in (
+            ("/rss/", "rss"),
+            ("/images/", "images"),
+            ("/files/", "files"),
+            ("/torrents/", "torrents"),
+        ):
+            if path.startswith(route_prefix):
+                return route_prefix, root / directory_name, path[len(route_prefix) :]
+        if path in {"/player", "/player/"}:
+            return "/player/", root / "player", "index.html"
+        if path.startswith("/player/"):
+            return "/player/", root / "player", path[len("/player/") :]
+        return None
+
+    def _send_static(
+        self,
+        journal: dict[str, Any],
+        route_prefix: str,
+        directory: Path,
+        relative: str,
+        *,
+        head_only: bool,
+    ) -> None:
+        base = directory.resolve()
+        candidate = (base / relative).resolve()
+        try:
+            candidate.relative_to(base)
+        except ValueError:
+            self._send_json(journal, 404, {"detail": "File not found"}, head_only)
+            return
+        if not candidate.is_file():
+            self._send_json(journal, 404, {"detail": "File not found"}, head_only)
+            return
+
+        size = candidate.stat().st_size
+        range_header = self.headers.get("Range")
+        headers = {"Accept-Ranges": "bytes"}
+        status = 200
+        start = 0
+        end = max(size - 1, 0)
+        if range_header:
+            try:
+                start, end = _parse_range(range_header, size)
+            except ValueError:
+                headers["Content-Range"] = f"bytes */{size}"
+                self._send_bytes(
+                    journal,
+                    416,
+                    b"",
+                    "application/octet-stream",
+                    head_only=head_only,
+                    headers=headers,
+                )
+                return
+            status = 206
+            headers["Content-Range"] = f"bytes {start}-{end}/{size}"
+
+        length = end - start + 1 if size else 0
+        if head_only:
+            data = b""
+        else:
+            with candidate.open("rb") as file:
+                file.seek(start)
+                data = file.read(length)
+        self._send_bytes(
+            journal,
+            status,
+            data,
+            _content_type(candidate, route_prefix),
+            head_only=head_only,
+            headers=headers,
+            content_length=length,
+        )
+
+    def _send_method_not_allowed(
+        self, journal: dict[str, Any], allow: str, head_only: bool
+    ) -> None:
+        self._send_json(
+            journal,
+            405,
+            {"detail": "Method not allowed"},
+            head_only,
+            headers={"Allow": allow},
+        )
+
+    def _send_json(
+        self,
+        journal: dict[str, Any],
+        status: int,
+        payload: Any,
+        head_only: bool,
+        headers: dict[str, str] | None = None,
+    ) -> None:
+        data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode(
+            "utf-8"
+        )
+        self._send_bytes(
+            journal,
+            status,
+            data,
+            "application/json; charset=utf-8",
+            head_only=head_only,
+            headers=headers,
+        )
+
+    def _send_empty(self, journal: dict[str, Any], status: int) -> None:
+        self._send_bytes(
+            journal,
+            status,
+            b"",
+            "application/octet-stream",
+            head_only=True,
+        )
+
+    def _send_bytes(
+        self,
+        journal: dict[str, Any],
+        status: int,
+        data: bytes,
+        content_type: str,
+        *,
+        head_only: bool,
+        headers: dict[str, str] | None = None,
+        content_length: int | None = None,
+    ) -> None:
+        self.server.mock_state.set_response_status(journal, status)
+        self.send_response(status)
+        self.send_header("Content-Type", content_type)
+        self.send_header(
+            "Content-Length",
+            str(len(data) if content_length is None else content_length),
+        )
+        for key, value in (headers or {}).items():
+            self.send_header(key, value)
+        self.end_headers()
+        if not head_only and data:
+            self.wfile.write(data)
+
+    def _send_unjournaled_json(
+        self, status: int, payload: Any, *, head_only: bool = False
+    ) -> None:
+        data = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode(
+            "utf-8"
+        )
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(data)))
+        self.end_headers()
+        if not head_only and data:
+            self.wfile.write(data)
+
+
+def create_server(
+    host: str, port: int, fixture_root: str | os.PathLike[str]
+) -> ThreadingHTTPServer:
+    """Create a configured server without starting its serve loop."""
+
+    root = Path(fixture_root).resolve()
+    if not root.is_dir():
+        raise FileNotFoundError(f"Fixture root does not exist: {root}")
+    return _MockUpstreamServer((host, port), root)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--host", default=os.environ.get("HOST", "0.0.0.0"))
+    parser.add_argument(
+        "--port", type=int, default=int(os.environ.get("PORT", "18888"))
+    )
+    parser.add_argument(
+        "--fixture-root",
+        default=os.environ.get("FIXTURES_ROOT", "/fixtures"),
+    )
+    args = parser.parse_args()
+    server = create_server(args.host, args.port, args.fixture_root)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()

e2e/scripts/audit_sources.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Statically reject non-hermetic or collision-prone E2E source patterns."""
+
+from __future__ import annotations
+
+import argparse
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Iterable, Sequence
+from urllib.parse import urlsplit
+
+_URL_RE = re.compile(r"https?://[^\s\"'<>]+")
+_FIXED_PORT_RE = re.compile(r"127\.0\.0\.1:\d+:\d+")
+_BLIND_SLEEP_RE = re.compile(r"\b(?:time\.)?sleep\(\s*\d+(?:\.\d+)?\s*\)")
+_ALLOWED_HOSTS = {
+    "127.0.0.1",
+    "0.0.0.0",
+    "::1",
+    "localhost",
+    "mock-upstream",
+    "fake-qb",
+    "qbittorrent",
+}
+
+
+@dataclass(frozen=True)
+class AuditFinding:
+    path: Path
+    line: int
+    message: str
+
+
+def _source_files(root: Path) -> Iterable[Path]:
+    e2e_root = root / "e2e"
+    if e2e_root.exists():
+        for path in e2e_root.rglob("*"):
+            if path.is_file() and (
+                path.name == "Dockerfile"
+                or path.suffix in {".py", ".yml", ".yaml", ".ts"}
+            ):
+                if path.name == "audit_sources.py":
+                    continue
+                yield path
+    playwright = root / "webui/playwright.config.ts"
+    if playwright.is_file():
+        yield playwright
+
+
+def _add(
+    findings: list[AuditFinding],
+    path: Path,
+    line: int,
+    message: str,
+) -> None:
+    findings.append(AuditFinding(path=path, line=line, message=message))
+
+
+def _audit_line(
+    root: Path,
+    path: Path,
+    line_number: int,
+    line: str,
+    findings: list[AuditFinding],
+) -> None:
+    relative = path.relative_to(root)
+    if "container_name" in line:
+        _add(findings, relative, line_number, "container_name is forbidden")
+    if ":latest" in line:
+        _add(findings, relative, line_number, "floating image tag is forbidden")
+    if _FIXED_PORT_RE.search(line):
+        _add(findings, relative, line_number, "fixed host port is forbidden")
+    if path.name not in {"stack.py", "poll.py"} and _BLIND_SLEEP_RE.search(line):
+        _add(findings, relative, line_number, "blind sleep is forbidden")
+    if path.name == "playwright.config.ts":
+        retry = re.search(r"\bretries\s*:\s*(\d+)", line)
+        if retry and int(retry.group(1)) != 0:
+            _add(findings, relative, line_number, "Playwright retries must be zero")
+    for raw_url in _URL_RE.findall(line):
+        try:
+            hostname = urlsplit(raw_url.rstrip("),];")).hostname
+        except ValueError:
+            if "{" in raw_url:  # A local f-string URL is checked after expansion.
+                continue
+            _add(findings, relative, line_number, "malformed URL in E2E source")
+            continue
+        if hostname and hostname not in _ALLOWED_HOSTS:
+            _add(
+                findings,
+                relative,
+                line_number,
+                f"public upstream host is forbidden: {hostname}",
+            )
+
+
+def _audit_image_references(
+    root: Path,
+    path: Path,
+    lines: list[str],
+    findings: list[AuditFinding],
+) -> None:
+    relative = path.relative_to(root)
+    if path.name == "Dockerfile":
+        for line_number, line in enumerate(lines, 1):
+            match = re.match(r"\s*FROM\s+(\S+)", line, re.I)
+            if match and "@sha256:" not in match.group(1):
+                _add(
+                    findings,
+                    relative,
+                    line_number,
+                    "external Dockerfile base image must use an immutable digest",
+                )
+    if path.suffix in {".yml", ".yaml"}:
+        for line_number, line in enumerate(lines, 1):
+            match = re.match(r"\s*image:\s*([^\s#]+)", line)
+            if not match:
+                continue
+            image = match.group(1).strip("\"'")
+            if image.startswith("${"):
+                continue
+            if "/" in image and "@sha256:" not in image:
+                _add(
+                    findings,
+                    relative,
+                    line_number,
+                    "external Compose image must use an immutable digest",
+                )
+
+
+def audit_repository(root: Path) -> list[AuditFinding]:
+    root = root.resolve()
+    findings: list[AuditFinding] = []
+    for path in sorted(set(_source_files(root))):
+        try:
+            text = path.read_text(encoding="utf-8")
+        except UnicodeDecodeError:
+            continue
+        lines = text.splitlines()
+        for line_number, line in enumerate(lines, 1):
+            _audit_line(root, path, line_number, line, findings)
+        _audit_image_references(root, path, lines, findings)
+    return findings
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "root",
+        nargs="?",
+        type=Path,
+        default=Path(__file__).resolve().parents[2],
+    )
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    findings = audit_repository(args.root)
+    for finding in findings:
+        print(f"{finding.path}:{finding.line}: {finding.message}")
+    return 1 if findings else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

e2e/scripts/build_test_image.py

@@ -0,0 +1,176 @@
+#!/usr/bin/env python3
+"""Build the native E2E image from an isolated temporary context."""
+
+from __future__ import annotations
+
+import argparse
+import re
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Iterable, Sequence
+
+DEFAULT_VERSION = "3.3.999-e2e.1"
+_VERSION_RE = re.compile(r"^\d+\.\d+\.\d+(?:-[0-9A-Za-z]+(?:[.-][0-9A-Za-z]+)*)?$")
+
+
+def repository_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def validate_version(version: str) -> str:
+    if not _VERSION_RE.fullmatch(version):
+        raise ValueError(f"E2E image version must be SemVer-compatible: {version!r}")
+    return version
+
+
+def tracked_backend_files(repo_root: Path) -> tuple[Path, ...]:
+    """Return tracked build inputs while excluding unrelated untracked files."""
+    result = subprocess.run(
+        [
+            "git",
+            "ls-files",
+            "-z",
+            "--",
+            ".dockerignore",
+            "Dockerfile",
+            "entrypoint.sh",
+            "backend",
+        ],
+        cwd=repo_root,
+        check=True,
+        capture_output=True,
+    )
+    return tuple(
+        Path(value.decode("utf-8")) for value in result.stdout.split(b"\0") if value
+    )
+
+
+def _safe_relative(path: Path) -> Path:
+    if path.is_absolute() or ".." in path.parts:
+        raise ValueError(f"Tracked path must stay inside the repository: {path}")
+    return path
+
+
+def prepare_context(
+    repo_root: Path,
+    dist_dir: Path,
+    destination: Path,
+    *,
+    version: str = DEFAULT_VERSION,
+    tracked_files: Iterable[Path] | None = None,
+) -> Path:
+    """Copy tracked build inputs, WebUI dist, and version into *destination*."""
+    version = validate_version(version)
+    repo_root = repo_root.resolve()
+    dist_dir = dist_dir.resolve()
+    destination = destination.resolve()
+    if not (dist_dir / "index.html").is_file():
+        raise FileNotFoundError(f"Built WebUI is missing index.html: {dist_dir}")
+    destination.mkdir(parents=True, exist_ok=True)
+
+    selected = tracked_files or tracked_backend_files(repo_root)
+    for relative in selected:
+        relative = _safe_relative(Path(relative))
+        source = repo_root / relative
+        if not source.is_file():
+            raise FileNotFoundError(f"Tracked build input is missing: {source}")
+        target = destination / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(source, target)
+
+    staged_dist = destination / "backend/src/dist"
+    if staged_dist.exists():
+        shutil.rmtree(staged_dist)
+    shutil.copytree(dist_dir, staged_dist)
+
+    version_module = destination / "backend/src/module/__version__.py"
+    version_module.parent.mkdir(parents=True, exist_ok=True)
+    version_module.write_text(f'VERSION = "{version}"\n', encoding="utf-8")
+    return destination
+
+
+def build_command(context: Path, *, image: str, version: str) -> list[str]:
+    validate_version(version)
+    if not image or any(character.isspace() for character in image):
+        raise ValueError(f"Invalid Docker image name: {image!r}")
+    return [
+        "docker",
+        "build",
+        "--build-arg",
+        f"VERSION={version}",
+        "--tag",
+        image,
+        "--pull=false",
+        str(context.resolve()),
+    ]
+
+
+def verification_command(*, image: str, version: str) -> list[str]:
+    validate_version(version)
+    script = (
+        "import sys; "
+        "from pathlib import Path; "
+        "from module.__version__ import VERSION; "
+        "expected=sys.argv[1]; "
+        "image_version=Path('/app/IMAGE_VERSION').read_text().strip(); "
+        "assert VERSION == expected == image_version, "
+        "(VERSION, expected, image_version)"
+    )
+    return [
+        "docker",
+        "run",
+        "--rm",
+        "--entrypoint",
+        "python",
+        image,
+        "-c",
+        script,
+        version,
+    ]
+
+
+def build_image(
+    *,
+    repo_root: Path,
+    dist_dir: Path,
+    image: str,
+    version: str,
+    runner=subprocess.run,
+) -> None:
+    with tempfile.TemporaryDirectory(prefix="ab-e2e-build-") as temporary:
+        context = prepare_context(
+            repo_root,
+            dist_dir,
+            Path(temporary) / "context",
+            version=version,
+        )
+        runner(build_command(context, image=image, version=version), check=True)
+        runner(verification_command(image=image, version=version), check=True)
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--version", default=DEFAULT_VERSION)
+    parser.add_argument("--image", default="auto-bangumi:e2e")
+    parser.add_argument("--dist", type=Path)
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    root = repository_root()
+    dist = args.dist or root / "webui/dist"
+    build_image(
+        repo_root=root,
+        dist_dir=dist,
+        image=args.image,
+        version=args.version,
+    )
+    print(f"Built and verified {args.image} at version {args.version}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

e2e/scripts/collect_artifacts.py

@@ -0,0 +1,190 @@
+#!/usr/bin/env python3
+"""Collect and redact diagnostics from a running hermetic E2E stack."""
+
+from __future__ import annotations
+
+import argparse
+import http.cookiejar
+import json
+import re
+import shutil
+import subprocess
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Iterable, Mapping, Sequence
+
+REDACTED = "[REDACTED]"
+_AUTH_RE = re.compile(r"(?i)\bAuthorization\b\s*[:=]\s*(?:Bearer\s+)?[^\s,;]+")
+_SENSITIVE_RE = re.compile(
+    r"(?i)\b(password|passwd|secret|token|cookie|sid)\b"
+    r"(\s*[:=]\s*)(?:\"[^\"]*\"|'[^']*'|[^\s,;]+)"
+)
+
+
+def redact_text(text: str, secrets: Iterable[str] = ()) -> str:
+    redacted = text
+    for secret in sorted({value for value in secrets if value}, key=len, reverse=True):
+        redacted = redacted.replace(secret, REDACTED)
+    redacted = _AUTH_RE.sub(f"Authorization: {REDACTED}", redacted)
+    redacted = _SENSITIVE_RE.sub(
+        lambda match: f"{match.group(1)}{match.group(2)}{REDACTED}", redacted
+    )
+    return redacted
+
+
+def _write_redacted(path: Path, text: str, secrets: Iterable[str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(redact_text(text, secrets), encoding="utf-8")
+
+
+def _capture(
+    command: Sequence[str],
+    path: Path,
+    *,
+    environment: Mapping[str, str] | None,
+    secrets: Iterable[str],
+) -> None:
+    try:
+        result = subprocess.run(
+            list(command),
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=30,
+            env=dict(environment) if environment is not None else None,
+        )
+        output = (
+            f"$ {' '.join(command)}\nexit={result.returncode}\n"
+            f"{result.stdout}\n{result.stderr}"
+        )
+    except Exception as exc:  # Diagnostics must never hide the test failure.
+        output = f"$ {' '.join(command)}\ncollection failed: {exc!r}\n"
+    _write_redacted(path, output, secrets)
+
+
+def _fetch_json(url: str, *, opener=None) -> str:
+    client = opener or urllib.request
+    with client.urlopen(url, timeout=5) as response:
+        payload = response.read().decode("utf-8", errors="replace")
+    try:
+        return json.dumps(json.loads(payload), indent=2, ensure_ascii=False)
+    except json.JSONDecodeError:
+        return payload
+
+
+def _collect_admin(
+    name: str,
+    base_url: str,
+    artifact_dir: Path,
+    secrets: Iterable[str],
+) -> None:
+    for endpoint in ("requests", "state"):
+        try:
+            payload = _fetch_json(f"{base_url.rstrip('/')}/__admin/{endpoint}")
+        except Exception as exc:
+            payload = f"request failed: {exc!r}\n"
+        _write_redacted(
+            artifact_dir / f"{name}-{endpoint}.json",
+            payload,
+            secrets,
+        )
+
+
+def _collect_qbittorrent(
+    base_url: str,
+    artifact_dir: Path,
+    secrets: Iterable[str],
+    *,
+    username: str = "admin",
+    password: str = "adminadmin",
+) -> None:
+    cookies = http.cookiejar.CookieJar()
+    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookies))
+    login = urllib.request.Request(
+        f"{base_url.rstrip('/')}/api/v2/auth/login",
+        data=urllib.parse.urlencode(
+            {"username": username, "password": password}
+        ).encode("utf-8"),
+    )
+    try:
+        with opener.open(login, timeout=5) as response:
+            response.read()
+        payload = _fetch_json(
+            f"{base_url.rstrip('/')}/api/v2/torrents/info", opener=opener
+        )
+    except Exception as exc:
+        payload = f"qBittorrent state request failed: {exc!r}\n"
+    _write_redacted(artifact_dir / "qbittorrent-state.json", payload, secrets)
+
+
+def collect_artifacts(
+    *,
+    compose_command: Sequence[str],
+    artifact_dir: Path,
+    data_dir: Path,
+    log_file: Path,
+    admin_urls: Mapping[str, str] | None = None,
+    qb_url: str | None = None,
+    environment: Mapping[str, str] | None = None,
+    secrets: Iterable[str] = (),
+) -> Path:
+    """Best-effort collection; failures are written into the bundle."""
+    artifact_dir.mkdir(parents=True, exist_ok=True)
+    all_secrets = tuple(secrets) + ("adminadmin",)
+    _capture(
+        [*compose_command, "ps", "--all"],
+        artifact_dir / "compose-ps.txt",
+        environment=environment,
+        secrets=all_secrets,
+    )
+    _capture(
+        [*compose_command, "logs", "--no-color", "--timestamps"],
+        artifact_dir / "compose-logs.txt",
+        environment=environment,
+        secrets=all_secrets,
+    )
+
+    for name, base_url in (admin_urls or {}).items():
+        _collect_admin(name, base_url, artifact_dir, all_secrets)
+    if qb_url:
+        _collect_qbittorrent(qb_url, artifact_dir, all_secrets)
+
+    if log_file.is_file():
+        _write_redacted(
+            artifact_dir / "backend-log.txt",
+            log_file.read_text(encoding="utf-8", errors="replace"),
+            all_secrets,
+        )
+    for name in ("data.db", "data.db-wal", "data.db-shm"):
+        source = data_dir / name
+        if source.is_file():
+            shutil.copy2(source, artifact_dir / name)
+    return artifact_dir
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--artifact-dir", required=True, type=Path)
+    parser.add_argument("--data-dir", required=True, type=Path)
+    parser.add_argument("--log-file", required=True, type=Path)
+    parser.add_argument("compose", nargs=argparse.REMAINDER)
+    return parser
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    if not args.compose:
+        raise SystemExit("compose command prefix is required after --")
+    command = args.compose[1:] if args.compose[0] == "--" else args.compose
+    collect_artifacts(
+        compose_command=command,
+        artifact_dir=args.artifact_dir,
+        data_dir=args.data_dir,
+        log_file=args.log_file,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

e2e/scripts/generate_fixtures.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Generate deterministic binary fixtures for the hermetic E2E stack."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+
+FIXTURE_ROOT = Path(__file__).resolve().parents[1] / "fixtures"
+MEDIA_PATH = Path("files/tiny-media.mkv")
+TORRENT_PATH = Path("torrents/tiny-media.torrent")
+CHECKSUM_PATH = Path("checksums.json")
+GENERATED_PATHS = (MEDIA_PATH, TORRENT_PATH)
+
+MEDIA_NAME = "Tiny Show - 01 [E2E].mkv"
+WEB_SEED = "http://mock-upstream:18888/files/tiny-media.mkv"
+PIECE_LENGTH = 16 * 1024
+
+
+def bencode(value: bytes | str | int | list | tuple | dict) -> bytes:
+    """Encode the small bencode subset used by the single-file torrent."""
+    if isinstance(value, bytes):
+        return str(len(value)).encode() + b":" + value
+    if isinstance(value, str):
+        return bencode(value.encode())
+    if isinstance(value, int):
+        return b"i" + str(value).encode() + b"e"
+    if isinstance(value, (list, tuple)):
+        return b"l" + b"".join(bencode(item) for item in value) + b"e"
+    if isinstance(value, dict):
+        items: list[tuple[bytes, object]] = []
+        for key, item in value.items():
+            encoded_key = key if isinstance(key, bytes) else str(key).encode()
+            items.append((encoded_key, item))
+        return (
+            b"d"
+            + b"".join(bencode(key) + bencode(item) for key, item in sorted(items))
+            + b"e"
+        )
+    raise TypeError(f"Unsupported bencode value: {type(value)!r}")
+
+
+def build_media() -> bytes:
+    """Return a stable 64 KiB payload without random or timestamp input."""
+    return bytes(range(256)) * 256
+
+
+def build_torrent(media: bytes) -> bytes:
+    """Return trackerless single-file metainfo backed by the local web seed."""
+    pieces = b"".join(
+        hashlib.sha1(media[offset : offset + PIECE_LENGTH]).digest()
+        for offset in range(0, len(media), PIECE_LENGTH)
+    )
+    return bencode(
+        {
+            b"created by": b"AutoBangumi hermetic E2E",
+            b"creation date": 0,
+            b"info": {
+                b"length": len(media),
+                b"name": MEDIA_NAME,
+                b"piece length": PIECE_LENGTH,
+                b"pieces": pieces,
+            },
+            b"url-list": WEB_SEED,
+        }
+    )
+
+
+def generated_content() -> dict[Path, bytes]:
+    media = build_media()
+    return {MEDIA_PATH: media, TORRENT_PATH: build_torrent(media)}
+
+
+def _checksum_content(content: dict[Path, bytes]) -> bytes:
+    checksums = {
+        str(path): hashlib.sha256(data).hexdigest() for path, data in content.items()
+    }
+    return (json.dumps(checksums, indent=2, sort_keys=True) + "\n").encode()
+
+
+def write_fixtures(root: Path = FIXTURE_ROOT) -> None:
+    content = generated_content()
+    for relative_path, data in content.items():
+        destination = root / relative_path
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        destination.write_bytes(data)
+    (root / CHECKSUM_PATH).write_bytes(_checksum_content(content))
+
+
+def check_fixtures(root: Path = FIXTURE_ROOT) -> list[str]:
+    content = generated_content()
+    expected = {**content, CHECKSUM_PATH: _checksum_content(content)}
+    mismatches: list[str] = []
+    for relative_path, data in expected.items():
+        destination = root / relative_path
+        if not destination.exists():
+            mismatches.append(f"missing: {relative_path}")
+        elif destination.read_bytes() != data:
+            mismatches.append(f"out of date: {relative_path}")
+    return mismatches
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--check", action="store_true")
+    parser.add_argument("--output", type=Path, default=FIXTURE_ROOT)
+    args = parser.parse_args()
+
+    if args.check:
+        mismatches = check_fixtures(args.output)
+        if mismatches:
+            print("\n".join(mismatches))
+            return 1
+        print("E2E binary fixtures are deterministic and up to date.")
+        return 0
+
+    write_fixtures(args.output)
+    print(f"Wrote deterministic E2E fixtures to {args.output}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

e2e/scripts/stack.py

@@ -0,0 +1,509 @@
+#!/usr/bin/env python3
+"""Lifecycle controller for isolated AutoBangumi E2E Compose stacks."""
+
+from __future__ import annotations
+
+import argparse
+import http.cookiejar
+import importlib.util
+import json
+import os
+import re
+import secrets
+import shutil
+import signal
+import subprocess
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Callable, Mapping, Sequence
+
+PROFILES = {"browser", "downloader"}
+DEFAULT_WAIT_SECONDS = 120
+QB_EXPECTED_PREFERENCES = {
+    "dht": False,
+    "pex": False,
+    "lsd": False,
+    "upnp": False,
+    "save_path": "/downloads/Bangumi",
+    "rss_processing_enabled": False,
+    "rss_auto_downloading_enabled": False,
+}
+
+
+def repository_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def unique_project_name(profile: str) -> str:
+    if profile not in PROFILES:
+        raise ValueError(f"Unknown E2E profile: {profile}")
+    raw = f"ab-e2e-{profile}-{os.getpid()}-{secrets.token_hex(4)}".lower()
+    normalized = re.sub(r"[^a-z0-9-]+", "-", raw).strip("-")
+    return normalized[:63].rstrip("-")
+
+
+def parse_published_port(output: str) -> int:
+    match = re.search(r":(\d+)\s*$", output.strip())
+    if not match:
+        raise ValueError(f"Cannot parse Docker Compose published port: {output!r}")
+    port = int(match.group(1))
+    if not 1 <= port <= 65535:
+        raise ValueError(f"Published port is out of range: {port}")
+    return port
+
+
+def validate_qb_preferences(preferences: Mapping[str, object]) -> None:
+    mismatches = {
+        key: {"expected": expected, "actual": preferences.get(key)}
+        for key, expected in QB_EXPECTED_PREFERENCES.items()
+        if preferences.get(key) != expected
+    }
+    if mismatches:
+        raise AssertionError(
+            "qBittorrent E2E preferences are unsafe or non-deterministic: "
+            f"{json.dumps(mismatches, sort_keys=True)}"
+        )
+
+
+def verify_qb_runtime(
+    base_url: str,
+    *,
+    username: str = "admin",
+    password: str = "adminadmin",
+) -> dict[str, object]:
+    cookies = http.cookiejar.CookieJar()
+    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(cookies))
+    request = urllib.request.Request(
+        f"{base_url.rstrip('/')}/api/v2/auth/login",
+        data=urllib.parse.urlencode(
+            {"username": username, "password": password}
+        ).encode("utf-8"),
+    )
+    with opener.open(request, timeout=5) as response:
+        body = response.read().decode("utf-8", errors="replace").strip()
+        if response.status not in (200, 204) or (
+            response.status == 200 and body.lower() != "ok."
+        ):
+            raise AssertionError(
+                f"qBittorrent rejected fixed E2E credentials: "
+                f"HTTP {response.status} {body!r}"
+            )
+    with opener.open(
+        f"{base_url.rstrip('/')}/api/v2/app/preferences", timeout=5
+    ) as response:
+        preferences = json.loads(response.read().decode("utf-8"))
+    if not isinstance(preferences, dict):
+        raise AssertionError("qBittorrent preferences response is not an object")
+    validate_qb_preferences(preferences)
+    return preferences
+
+
+def _read_env_file(path: Path) -> dict[str, str]:
+    values: dict[str, str] = {}
+    if not path.is_file():
+        return values
+    for raw_line in path.read_text(encoding="utf-8").splitlines():
+        line = raw_line.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, raw_value = line.split("=", 1)
+        try:
+            value = json.loads(raw_value)
+        except json.JSONDecodeError:
+            value = raw_value
+        values[key] = str(value)
+    return values
+
+
+def _write_env_file(path: Path, values: Mapping[str, str]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    lines = []
+    for key, value in sorted(values.items()):
+        if "\n" in value or "\r" in value:
+            raise ValueError(f"Environment value for {key} contains a newline")
+        lines.append(f"{key}={json.dumps(value)}")
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def _load_artifact_module(repo_root: Path):
+    path = repo_root / "e2e/scripts/collect_artifacts.py"
+    spec = importlib.util.spec_from_file_location("ab_e2e_collect_artifacts", path)
+    if spec is None or spec.loader is None:
+        raise RuntimeError(f"Cannot load artifact collector: {path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+class Stack:
+    """One Compose project with isolated bind mounts and dynamic host ports."""
+
+    def __init__(
+        self,
+        *,
+        profile: str,
+        repo_root: Path | None = None,
+        work_dir: Path | None = None,
+        project_name: str | None = None,
+        runner=subprocess.run,
+    ) -> None:
+        if profile not in PROFILES:
+            raise ValueError(f"Unknown E2E profile: {profile}")
+        self.profile = profile
+        self.repo_root = (repo_root or repository_root()).resolve()
+        self.project_name = project_name or unique_project_name(profile)
+        self.work_dir = (
+            work_dir.resolve()
+            if work_dir is not None
+            else Path(tempfile.mkdtemp(prefix=f"{self.project_name}-")).resolve()
+        )
+        self.config_dir = self.work_dir / "config"
+        self.data_dir = self.work_dir / "data"
+        self.download_dir = self.work_dir / "downloads"
+        self.log_dir = self.work_dir / "logs"
+        self.log_file = self.log_dir / "log.txt"
+        self.qb_config_dir = self.work_dir / "qb-config"
+        self.artifact_dir = self.work_dir / "artifacts"
+        self.env_file = self.work_dir / "stack.env"
+        self._runner = runner
+        self._started = False
+
+        for directory in (
+            self.config_dir,
+            self.data_dir,
+            self.download_dir,
+            self.log_dir,
+            self.qb_config_dir,
+            self.artifact_dir,
+        ):
+            directory.mkdir(parents=True, exist_ok=True)
+        self.log_file.touch(exist_ok=True)
+        self._install_qb_config()
+
+        existing = _read_env_file(self.env_file)
+        self.environment: dict[str, str] = {
+            "AB_E2E_PROFILE": self.profile,
+            "AB_E2E_PROJECT": self.project_name,
+            "AB_E2E_CONFIG_DIR": str(self.config_dir),
+            "AB_E2E_DATA_DIR": str(self.data_dir),
+            "AB_E2E_DOWNLOAD_DIR": str(self.download_dir),
+            "AB_E2E_LOG_FILE": str(self.log_file),
+            "AB_E2E_QB_CONFIG_DIR": str(self.qb_config_dir),
+            "AB_E2E_FIXTURES_DIR": str(self.repo_root / "e2e/fixtures"),
+            "AB_E2E_ARTIFACT_DIR": str(self.artifact_dir),
+            "AB_E2E_USERNAME": "testadmin",
+            "AB_E2E_PASSWORD": "testpassword123",
+            "AB_E2E_APP_IMAGE": os.environ.get("AB_E2E_APP_IMAGE", "auto-bangumi:e2e"),
+            "AB_E2E_MOCK_IMAGE": os.environ.get(
+                "AB_E2E_MOCK_IMAGE", f"{self.project_name}/mock-upstream:local"
+            ),
+            "AB_E2E_FAKE_QB_IMAGE": os.environ.get(
+                "AB_E2E_FAKE_QB_IMAGE", f"{self.project_name}/fake-qb:local"
+            ),
+            "AB_E2E_PUID": str(getattr(os, "getuid", lambda: 1000)()),
+            "AB_E2E_PGID": str(getattr(os, "getgid", lambda: 1000)()),
+            "AB_E2E_MOCK_INTERNAL_URL": "http://mock-upstream:18888",
+            "AB_E2E_FAKE_QB_INTERNAL_URL": "http://fake-qb:8080",
+            "AB_E2E_QB_INTERNAL_URL": "http://qbittorrent:8080",
+            **existing,
+        }
+        self.write_environment()
+
+    @property
+    def compose_files(self) -> tuple[Path, ...]:
+        files = (self.repo_root / "e2e/compose/browser.yml",)
+        if self.profile == "downloader":
+            files += (self.repo_root / "e2e/compose/downloader.yml",)
+        return files
+
+    @property
+    def process_environment(self) -> dict[str, str]:
+        return {**os.environ, **self.environment}
+
+    def _install_qb_config(self) -> None:
+        source = self.repo_root / "e2e/fixtures/qbittorrent/qBittorrent.conf"
+        destination = self.qb_config_dir / "qBittorrent/qBittorrent.conf"
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        if not destination.exists():
+            shutil.copy2(source, destination)
+
+    def write_environment(self) -> Path:
+        _write_env_file(self.env_file, self.environment)
+        return self.env_file
+
+    def compose_command(self, *arguments: str) -> list[str]:
+        command = [
+            "docker",
+            "compose",
+            "--project-name",
+            self.project_name,
+            "--env-file",
+            str(self.env_file),
+        ]
+        for compose_file in self.compose_files:
+            command.extend(("--file", str(compose_file)))
+        command.extend(arguments)
+        return command
+
+    def _run_compose(
+        self,
+        *arguments: str,
+        check: bool = True,
+        capture_output: bool = False,
+    ):
+        return self._runner(
+            self.compose_command(*arguments),
+            cwd=self.repo_root,
+            env=self.process_environment,
+            check=check,
+            capture_output=capture_output,
+            text=capture_output,
+        )
+
+    def validate(self) -> None:
+        self._run_compose("config", capture_output=True)
+
+    def _published_port(self, service: str, container_port: int) -> int:
+        result = self._run_compose(
+            "port",
+            "--index",
+            "1",
+            service,
+            str(container_port),
+            capture_output=True,
+        )
+        return parse_published_port(result.stdout)
+
+    def ports(self) -> dict[str, int]:
+        ports = {
+            "app": self._published_port("app", 7892),
+            "mock-upstream": self._published_port("mock-upstream", 18888),
+            "fake-qb": self._published_port("fake-qb", 8080),
+        }
+        if self.profile == "downloader":
+            ports["qbittorrent"] = self._published_port("qbittorrent", 8080)
+        return ports
+
+    def _publish_urls(self, ports: Mapping[str, int]) -> None:
+        self.environment.update(
+            {
+                "AB_E2E_BASE_URL": f"http://127.0.0.1:{ports['app']}",
+                "AB_E2E_MOCK_URL": (f"http://127.0.0.1:{ports['mock-upstream']}"),
+                "AB_E2E_FAKE_QB_URL": f"http://127.0.0.1:{ports['fake-qb']}",
+            }
+        )
+        if "qbittorrent" in ports:
+            real_qb_url = f"http://127.0.0.1:{ports['qbittorrent']}"
+            self.environment["AB_E2E_REAL_QB_URL"] = real_qb_url
+            self.environment["AB_E2E_QB_URL"] = real_qb_url
+        self.write_environment()
+
+    def start(self) -> dict[str, int]:
+        if self.profile == "downloader":
+            self._run_compose("pull", "qbittorrent")
+        self._run_compose("build", "mock-upstream", "fake-qb")
+        self._run_compose(
+            "up",
+            "--detach",
+            "--wait",
+            "--wait-timeout",
+            str(DEFAULT_WAIT_SECONDS),
+        )
+        self._started = True
+        published = self.ports()
+        self._publish_urls(published)
+        if self.profile == "downloader":
+            verify_qb_runtime(self.environment["AB_E2E_REAL_QB_URL"])
+        return published
+
+    def _wait_for_health(self, timeout: float = 30) -> None:
+        url = f"{self.environment['AB_E2E_BASE_URL'].rstrip('/')}/health"
+        deadline = time.monotonic() + timeout
+        last_error = "no request attempted"
+        while time.monotonic() < deadline:
+            try:
+                with urllib.request.urlopen(url, timeout=2) as response:
+                    if response.status == 200:
+                        return
+                    last_error = f"HTTP {response.status}"
+            except (OSError, urllib.error.URLError) as exc:
+                last_error = repr(exc)
+            time.sleep(0.2)
+        raise TimeoutError(f"App did not become healthy at {url}: {last_error}")
+
+    def restart(self, service: str = "app") -> None:
+        self._run_compose("restart", service)
+        if service == "app" and "AB_E2E_BASE_URL" in self.environment:
+            self._wait_for_health()
+        else:
+            self._run_compose(
+                "up",
+                "--detach",
+                "--wait",
+                "--wait-timeout",
+                str(DEFAULT_WAIT_SECONDS),
+                service,
+            )
+
+    def collect(self) -> None:
+        collector = _load_artifact_module(self.repo_root)
+        admin_urls = {}
+        if "AB_E2E_MOCK_URL" in self.environment:
+            admin_urls["mock-upstream"] = self.environment["AB_E2E_MOCK_URL"]
+        if "AB_E2E_FAKE_QB_URL" in self.environment:
+            admin_urls["fake-qb"] = self.environment["AB_E2E_FAKE_QB_URL"]
+        collector.collect_artifacts(
+            compose_command=self.compose_command(),
+            artifact_dir=self.artifact_dir,
+            data_dir=self.data_dir,
+            log_file=self.log_file,
+            admin_urls=admin_urls,
+            qb_url=self.environment.get("AB_E2E_REAL_QB_URL"),
+            environment=self.process_environment,
+            secrets=(
+                self.environment.get("AB_E2E_SESSION_TOKEN", ""),
+                self.environment.get("AB_E2E_API_TOKEN", ""),
+            ),
+        )
+
+    def stop(self, *, collect: bool = True) -> None:
+        try:
+            if collect and self._started:
+                self.collect()
+        finally:
+            self._run_compose(
+                "down",
+                "--volumes",
+                "--remove-orphans",
+                check=False,
+            )
+            self._started = False
+
+
+CommandRunner = Callable[[Sequence[str], Mapping[str, str]], object]
+
+
+def _default_command_runner(
+    command: Sequence[str], environment: Mapping[str, str]
+) -> int:
+    return subprocess.run(list(command), env=dict(environment), check=False).returncode
+
+
+@contextmanager
+def scoped_signal_cleanup(stack):
+    """Ensure SIGINT/SIGTERM tear down this stack, then restore prior handlers."""
+    previous: dict[int, object] = {}
+    cleaning = False
+
+    def handle(signum, _frame):
+        nonlocal cleaning
+        if not cleaning:
+            cleaning = True
+            stack.stop(collect=True)
+        raise SystemExit(128 + signum)
+
+    try:
+        for signum in (signal.SIGINT, signal.SIGTERM):
+            previous[signum] = signal.getsignal(signum)
+            signal.signal(signum, handle)
+        yield
+    finally:
+        for signum, handler in previous.items():
+            signal.signal(signum, handler)
+
+
+def run_with_stack(
+    stack,
+    command: Sequence[str],
+    *,
+    runner: CommandRunner = _default_command_runner,
+):
+    with scoped_signal_cleanup(stack):
+        stack.start()
+        try:
+            return runner(list(command), {**os.environ, **stack.environment})
+        finally:
+            stack.stop(collect=True)
+
+
+def _add_stack_arguments(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument("--profile", choices=sorted(PROFILES), default="browser")
+    parser.add_argument("--repo-root", type=Path, default=repository_root())
+    parser.add_argument("--work-dir", type=Path)
+    parser.add_argument("--project-name")
+
+
+def _parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description=__doc__)
+    subparsers = parser.add_subparsers(dest="action", required=True)
+    for action in ("validate", "start", "stop", "ports", "env", "smoke"):
+        subparser = subparsers.add_parser(action)
+        _add_stack_arguments(subparser)
+    restart = subparsers.add_parser("restart")
+    _add_stack_arguments(restart)
+    restart.add_argument("service", nargs="?", default="app")
+    run = subparsers.add_parser("run")
+    _add_stack_arguments(run)
+    run.add_argument("command", nargs=argparse.REMAINDER)
+    return parser
+
+
+def _stack_from_args(args) -> Stack:
+    return Stack(
+        profile=args.profile,
+        repo_root=args.repo_root,
+        work_dir=args.work_dir,
+        project_name=args.project_name,
+    )
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    args = _parser().parse_args(argv)
+    stack = _stack_from_args(args)
+    if args.action == "validate":
+        stack.validate()
+        return 0
+    if args.action == "start":
+        print(json.dumps(stack.start(), sort_keys=True))
+        print(stack.env_file)
+        return 0
+    if args.action == "stop":
+        stack.stop(collect=True)
+        return 0
+    if args.action == "ports":
+        ports = stack.ports()
+        stack._publish_urls(ports)
+        print(json.dumps(ports, sort_keys=True))
+        return 0
+    if args.action == "env":
+        print(stack.env_file.read_text(encoding="utf-8"), end="")
+        return 0
+    if args.action == "restart":
+        stack.restart(args.service)
+        return 0
+    if args.action == "smoke":
+        with scoped_signal_cleanup(stack):
+            stack.start()
+            try:
+                stack._wait_for_health()
+            finally:
+                stack.stop(collect=True)
+        return 0
+    if args.action == "run":
+        command = args.command[1:] if args.command[:1] == ["--"] else args.command
+        if not command:
+            raise SystemExit("A command is required after --")
+        result = run_with_stack(stack, command)
+        return int(result) if isinstance(result, int) else 0
+    raise AssertionError(f"Unhandled action: {args.action}")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())