tests/test_fastapi_endpoints.py

from __future__ import annotations

import copy
import gzip
import json
import tempfile
import time
import unittest
from pathlib import Path
from unittest import mock

IMPORT_ERROR = None
try:
    from backend import app
    from fastapi.testclient import TestClient
except ModuleNotFoundError as exc:  # pragma: no cover
    app = None
    TestClient = None
    IMPORT_ERROR = exc


@unittest.skipIf(app is None or TestClient is None, f"FastAPI dependencies unavailable: {IMPORT_ERROR}")
class FastApiEndpointTests(unittest.TestCase):
    def setUp(self):
        self.saved = {name: copy.deepcopy(getattr(app, name)) for name in ("records", "repo_counts", "repo_list", "repo_records_map", "extension_counts", "extension_list", "repo_extension_counts", "folder_index", "folder_tree_data", "folder_browser_data", "api_response_cache", "initial_payload_global", "initial_payload_by_repo", "reader_record_indices", "repo_reader_record_indices", "reader_record_key_indices", "reader_assets_cache", "reader_assets_cache_at", "reader_assets_revision", "reader_join_cache_key", "reader_join_cache", "READER_ASSET_ROUTING_CONTRACT_VERSION")}
        self.saved["reader_source_id_indices"] = copy.deepcopy(app.reader_source_id_indices)
        app.records = [{"Repo": "VoiceOfML/Test", "File": "readme", "Extension": "txt", "Folder": ["docs"], "Size": 10, "HasTxt": True}]
        app.reader_assets_cache = {"v": 1, "f": {}}
        app.reader_assets_cache_at = time.monotonic()
        app.reader_join_cache_key = None
        app.reader_join_cache = {"all": {}, "repos": {}}
        app.build_indexes()
        app.build_initial_payloads()
        self.client = TestClient(app.app)

    def tearDown(self):
        for name, value in self.saved.items():
            setattr(app, name, value)

    def test_search_endpoint_accepts_valid_json(self):
        response = self.client.post("/api/search", json={"q": "readme", "page_size": 1})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["total"], 1)
        self.assertRegex(response.headers["server-timing"], r"queue;dur=\d+\.\d+, search;dur=\d+\.\d+")
        self.assertNotIn("readme", response.headers["server-timing"])

    def test_reader_ocr_search_returns_exact_page_hits(self):
        root = "derived/test/" + "a" * 32
        manifest_path = root + "/ocr-manifest.json"
        book_path = root + "/ocr/book-text.json.gz"
        manifest = {"version": 1, "kind": "pdf-ocr", "page_count": 2,
                    "pages": [{"p": 1, "o": root + "/ocr/page-000001.json.gz"},
                              {"p": 2, "o": root + "/ocr/page-000002.json.gz"}],
                    "book_text": {"path": book_path}}
        pages = [{"page": 1, "text": "第一章 手机阅读", "layout": {"offset_unit": "unicode-codepoint"}, "text_spans": []},
                 {"page": 2, "text": "第二章 手机搜索", "layout": {"offset_unit": "unicode-codepoint"}, "text_spans": []}]
        book = gzip.compress(json.dumps({"version": 2, "kind": "pdf-book-text", "complete": True,
                                         "offset_unit": "unicode-codepoint", "source_sha256": "a" * 64,
                                         "page_count": 2, "pages": pages}).encode())

        async def resource(path):
            return json.dumps(manifest).encode() if path == manifest_path else book

        resolved = app.JSONResponse({"ocr_manifest": "https://voiceofml-search.hf.space/api/reader-bucket-resource?path=" + manifest_path})
        with mock.patch.object(app, "api_reader_resolve", mock.AsyncMock(return_value=resolved)), \
             mock.patch.object(app, "get_reader_bucket_resource", side_effect=resource):
            response = self.client.get("/api/reader-ocr-search", params={"id": "a" * 13, "q": "手机"})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["total"], 2)
        self.assertEqual([item["page"] for item in response.json()["results"]], [1, 2])

    def test_reader_ocr_search_counts_all_hits_without_retaining_all_results(self):
        pages = [{"page": 1, "text": "x" * 200, "folded": "x" * 200,
                  "spans": [], "layout": {}}]
        total, results = app.search_reader_ocr_pages(pages, "x", 3, 10)
        self.assertEqual(total, 200)
        self.assertEqual(len(results), 10)
        self.assertEqual([item["start"] for item in results], list(range(20, 30)))

    @staticmethod
    def ocr_pages(*texts, spans=None):
        pages = [{"page": index + 1, "text": text, "layout": {"offset_unit": "unicode-codepoint"},
                  "text_spans": (spans or {}).get(index + 1, [])} for index, text in enumerate(texts)]
        raw = gzip.compress(json.dumps({"version": 2, "kind": "pdf-book-text", "complete": True,
                                        "offset_unit": "unicode-codepoint", "source_sha256": "a" * 64,
                                        "page_count": len(pages), "pages": pages}).encode())
        return app.parse_reader_ocr_book(raw)[0]

    def ocr_search(self, pages, query, page=1, page_size=50):
        return app.search_reader_ocr_pages(pages, query.casefold(), page, page_size)

    def test_reader_ocr_offsets_use_original_text_after_expanding_casefold(self):
        # Offsets, lengths, boxes and snippets are original-text code points.
        box_ss, box_ab = [0, 0, .1, .1], [.2, .2, .3, .3]
        pages = self.ocr_pages("ßßab", "file flow", "İx", "Straße STRASSE",
                               spans={1: [{"start": 0, "end": 2, "box": box_ss},
                                          {"start": 2, "end": 4, "box": box_ab}]})
        total, results = self.ocr_search(pages[:1], "ab")
        self.assertEqual(total, 1)
        self.assertEqual({key: results[0][key] for key in ("start", "length", "snippet_start", "boxes")},
                         {"start": 2, "length": 2, "snippet_start": 2, "boxes": [box_ab]})
        hit = results[0]
        self.assertEqual(hit["snippet"][hit["snippet_start"]:hit["snippet_start"] + hit["length"]], "ab")
        total, results = self.ocr_search(pages[1:2], "FILE")
        self.assertEqual((total, results[0]["start"], results[0]["length"]), (1, 0, 3))
        total, results = self.ocr_search(pages[1:2], "flow")
        self.assertEqual((total, results[0]["start"], results[0]["length"]), (1, 4, 3))
        total, results = self.ocr_search(pages[2:3], "x")
        self.assertEqual((total, results[0]["start"], results[0]["length"]), (1, 1, 1))
        total, results = self.ocr_search(pages[3:4], "strasse")
        self.assertEqual(total, 2)
        self.assertEqual([(item["start"], item["length"]) for item in results], [(0, 6), (7, 7)])

    def test_reader_ocr_expanded_character_matches_are_whole_original_characters(self):
        # "ss" matches each ß once; "s" matches both halves of one ß, and each
        # hit reports that single original ß. Matches stay non-overlapping.
        pages = self.ocr_pages("ßß", "ssss")
        total, results = self.ocr_search(pages[:1], "ss")
        self.assertEqual(total, 2)
        self.assertEqual([(item["start"], item["length"]) for item in results], [(0, 1), (1, 1)])
        total, results = self.ocr_search(pages[:1], "s")
        self.assertEqual(total, 4)
        self.assertEqual([item["start"] for item in results], [0, 0, 1, 1])
        self.assertEqual({item["length"] for item in results}, {1})
        total, results = self.ocr_search(pages[1:], "ß")
        self.assertEqual(total, 2)
        self.assertEqual([(item["start"], item["length"]) for item in results], [(0, 2), (2, 2)])
        total, results = self.ocr_search(pages, "ss", page=2, page_size=3)
        self.assertEqual(total, 4)
        self.assertEqual([(item["page"], item["start"]) for item in results], [(2, 2)])

    def test_reader_ocr_search_maps_storage_failures_to_gateway_statuses(self):
        root = "derived/test/" + "b" * 32
        manifest_path = root + "/ocr-manifest.json"
        resolved = app.JSONResponse({"ocr_manifest": "https://voiceofml-search.hf.space/api/reader-bucket-resource?path=" + manifest_path})
        saved_state = dict(app.app.state._state)
        self.addCleanup(lambda: (app.app.state._state.clear(), app.app.state._state.update(saved_state)))
        app.initialize_reader_bucket_state(app.app)
        cases = (
            ({"side_effect": FileNotFoundError(manifest_path)}, 404),
            ({"side_effect": OSError("HF transport failed")}, 502),
            ({"side_effect": RuntimeError("HF HTTP 500")}, 502),
        )
        with mock.patch.object(app, "api_reader_resolve", mock.AsyncMock(return_value=resolved)):
            for options, status in cases:
                with self.subTest(status=status), \
                        mock.patch.object(app, "read_reader_bucket_resource", **options):
                    response = self.client.get("/api/reader-ocr-search", params={"id": "a" * 13, "q": "x"})
                    self.assertEqual(response.status_code, status, response.text)
                    self.assertEqual(response.headers["cache-control"], "no-store")
            with mock.patch.object(app, "READER_BUCKET_MAX_INFLIGHT", 0), \
                    mock.patch.object(app, "read_reader_bucket_resource", return_value=b"{}") as read:
                response = self.client.get("/api/reader-ocr-search", params={"id": "a" * 13, "q": "x"})
            self.assertEqual(response.status_code, 503)
            self.assertEqual(response.headers["retry-after"], "1")
            read.assert_not_called()

    def test_search_endpoint_rejects_malformed_json(self):
        self.assertEqual(self.client.post("/api/search", content="not-json", headers={"content-type": "application/json"}).status_code, 422)

    def test_search_endpoint_rejects_wrong_field_type(self):
        self.assertEqual(self.client.post("/api/search", json={"page_size": "many"}).status_code, 422)

    def test_search_endpoint_rejects_unsupported_enum_values(self):
        sort_response = self.client.post("/api/search", json={"sort": "random"})
        folder_response = self.client.post("/api/search", json={"folder_match_mode": "contains"})
        self.assertEqual(sort_response.status_code, 422)
        self.assertEqual(folder_response.status_code, 422)
        self.assertIn("relevance", sort_response.json()["detail"][0]["msg"])
        self.assertIn("prefix", folder_response.json()["detail"][0]["msg"])

    def test_search_endpoint_bounds_query_and_list_inputs(self):
        self.assertEqual(self.client.post("/api/search", json={"q": "x" * 257}).status_code, 422)
        self.assertEqual(self.client.post("/api/search", json={"repos": ["Repo"] * 101}).status_code, 422)

    def test_repo_search_endpoint_returns_repo_results(self):
        response = self.client.post("/api/search/Test", json={"q": "readme"})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["results"][0]["Repo"], "VoiceOfML/Test")

    def test_exact_search_endpoint_applies_extension_and_folder_filters(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "readme", "Extension": "txt", "Folder": ["docs"], "Size": 10, "HasTxt": True},
            {"Repo": "VoiceOfML/Test", "File": "readme", "Extension": "epub", "Folder": ["docs"], "Size": 20, "HasTxt": False},
            {"Repo": "VoiceOfML/Test", "File": "readme", "Extension": "pdf", "Folder": ["other"], "Size": 30, "HasTxt": False},
        ]
        app.build_indexes()
        # The UI defaults to exact=true; extension filters must still narrow it.
        exact_all = self.client.post("/api/search", json={"q": "readme", "exact": True})
        self.assertEqual(exact_all.status_code, 200)
        self.assertEqual(exact_all.json()["total"], 3)
        epub = self.client.post("/api/search", json={"q": "readme", "exact": True, "extensions": ["EPUB"]})
        self.assertEqual(epub.json()["total"], 1)
        self.assertEqual(epub.json()["results"][0]["Extension"], "epub")
        docs = self.client.post("/api/search", json={"q": "readme", "exact": True, "folders": ["docs"]})
        self.assertEqual(docs.json()["total"], 2)
        mixed = self.client.post("/api/search", json={
            "q": "readme", "exact": True, "folder_match_mode": "mixed",
            "folder_selfs": ["other"], "folder_subtrees": [],
        })
        self.assertEqual(mixed.json()["total"], 1)
        self.assertEqual(mixed.json()["results"][0]["Extension"], "pdf")
        sized = self.client.post("/api/search", json={"q": "readme", "exact": True, "min_size": 25})
        self.assertEqual(sized.json()["total"], 1)

    def test_unknown_repo_search_is_404(self):
        self.assertEqual(self.client.post("/api/search/Missing", json={}).status_code, 404)

    def test_repos_extensions_and_status_endpoints(self):
        repos = self.client.get("/api/repos")
        self.assertEqual(repos.json()[0]["name"], "VoiceOfML/Test")
        self.assertIn("max-age=300", repos.headers["cache-control"])
        self.assertEqual(self.client.get("/api/extensions").json(), [{"name": "txt", "count": 1}])
        self.assertEqual(self.client.get("/api/random-txt/status").json(), {"available": True, "count": 1})
        self.assertEqual(self.client.get("/api/random-reader/status").json(), {"available": True, "count": 1})

    def test_reader_asset_join_cache_uses_generation_revision_and_contract(self):
        app.records.append({"Repo": "VoiceOfML/Test", "File": "book", "Extension": "docx",
                            "Folder": ["docs"], "Size": 10, "HasTxt": False})
        app.build_indexes()
        key = "VoiceOfML/Test\0docs/book.docx"
        asset = {"s": 2, "m": "d", "p": "documents/office/docx/" + "a" * 64 + "/document.docx",
                 "b": "vomebook/reader-assets-v2"}
        assets = {"v": 1, "f": {key: asset}}
        app.reader_assets_revision = "sidecar-revision-a"
        app.reader_join_cache_key = None

        first = app.converted_reader_indices(assets)
        second = app.converted_reader_indices(assets)
        self.assertIs(first, second)
        self.assertEqual(first, {1: asset})

        app.reader_assets_revision = "sidecar-revision-b"
        refreshed = app.converted_reader_indices({"v": 1, "f": {}})
        self.assertIsNot(first, refreshed)
        self.assertEqual(refreshed, {})

        app.READER_ASSET_ROUTING_CONTRACT_VERSION = "test-next-contract"
        contract_refreshed = app.converted_reader_indices(assets)
        self.assertIsNot(refreshed, contract_refreshed)
        self.assertEqual(contract_refreshed, {1: asset})

    def test_bootstrap_combines_global_startup_data(self):
        response = self.client.get("/api/bootstrap")
        self.assertEqual(response.status_code, 200)
        payload = response.json()
        # Same field and value domain as search/initial responses.
        self.assertEqual(payload["generation"], app.search_generation())
        self.assertIsInstance(payload["generation"], str)
        self.assertEqual(self.client.get("/data/initial/global.json").json()["generation"],
                         payload["generation"])
        self.assertEqual(payload["repos"][0]["name"], "VoiceOfML/Test")
        self.assertEqual(payload["extensions"], [{"name": "txt", "count": 1}])
        self.assertEqual(payload["random_txt"], {"available": True, "count": 1})
        self.assertIn("sidebar", payload)
        self.assertIn("max-age=300", response.headers["cache-control"])

    def test_bootstrap_txt_status_excludes_non_txt_reader_records(self):
        app.records.append({"Repo": "VoiceOfML/Test", "File": "book", "Extension": "epub",
                            "Folder": [], "Size": 12, "HasTxt": False})
        app.build_indexes()
        status = self.client.get("/api/random-txt/status").json()
        self.assertEqual(status, {"available": True, "count": 1})
        self.assertEqual(self.client.get("/api/bootstrap").json()["random_txt"], status)
        self.assertEqual(len(app.reader_record_indices), 2)

    def test_search_normalizes_duplicate_repository_filters(self):
        expected = self.client.post("/api/search", json={"repos": ["VoiceOfML/Test"]}).json()
        actual = self.client.post("/api/search", json={"repos": ["VoiceOfML/Test", " VoiceOfML/Test "]}).json()
        self.assertEqual(actual, expected)

    def test_folder_tree_and_contents_endpoints(self):
        self.assertEqual(self.client.get("/api/folders/Test").status_code, 200)
        self.assertEqual(self.client.get("/api/folders/Test/contents", params={"path": "docs"}).json()["files"][0]["name"], "readme")

    def test_folder_and_initial_caches_include_generation(self):
        app.folder_tree_data = {"VoiceOfML/Test": [{"name": "old"}]}
        app.folder_browser_data = {"VoiceOfML/Test": {"": {"folders": [], "files": [{"name": "old"}]}}}
        app.initial_payload_global = {"generation": "old"}
        app.initial_payload_by_repo = {"VoiceOfML/Test": {"generation": "old"}}
        old_generation = app._data_generation
        try:
            self.assertEqual(self.client.get("/api/folders/Test").json()[0]["name"], "old")
            self.assertEqual(self.client.get("/api/folders/Test/contents").json()["files"][0]["name"], "old")
            self.assertEqual(self.client.get("/data/initial/global.json").json()["generation"], "old")
            self.assertEqual(self.client.get("/data/initial/repos/Test.json").json()["generation"], "old")
            self.assertNotIn("generation", self.client.get("/data/initial/manifest.json").json())
            app._data_generation = old_generation + 1
            app.folder_tree_data = {"VoiceOfML/Test": [{"name": "new"}]}
            app.folder_browser_data = {"VoiceOfML/Test": {"": {"folders": [], "files": [{"name": "new"}]}}}
            app.initial_payload_global = {"generation": "new"}
            app.initial_payload_by_repo = {
                "VoiceOfML/Test": {"generation": "new"},
                "VoiceOfML/New": {"generation": "new"},
            }
            self.assertEqual(self.client.get("/api/folders/Test").json()[0]["name"], "new")
            self.assertEqual(self.client.get("/api/folders/Test/contents").json()["files"][0]["name"], "new")
            self.assertEqual(self.client.get("/data/initial/global.json").json()["generation"], "new")
            self.assertEqual(self.client.get("/data/initial/repos/Test.json").json()["generation"], "new")
            self.assertEqual(
                self.client.get("/data/initial/manifest.json").json()["urls"],
                ["/data/initial/global.json", "/data/initial/repos/New.json", "/data/initial/repos/Test.json"],
            )
        finally:
            app._data_generation = old_generation

    def test_random_endpoint_is_record_shaped_without_private_index_fields(self):
        result = self.client.get("/api/random").json()
        self.assertEqual(result["Repo"], "VoiceOfML/Test")
        self.assertIn("Link", result)
        self.assertIn("Path", result)
        self.assertFalse(any(key.startswith("_") for key in result))

    def test_random_txt_endpoint_is_record_shaped_without_private_index_fields(self):
        result = self.client.get("/api/random-txt").json()
        self.assertTrue(result["HasTxt"])
        self.assertFalse(any(key.startswith("_") for key in result))

    def test_random_reader_hides_private_index_fields(self):
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
            result = self.client.get("/api/random-reader").json()
        self.assertNotIn("ReaderLink", result)
        self.assertTrue(result["Link"].endswith("/docs/readme.txt"))
        self.assertFalse(any(key.startswith("_") for key in result))

    def test_reader_source_id_fixed_vector_and_index_rebuild(self):
        app.records = [{"Repo": "VoiceOfML/Test", "File": "book29", "Extension": "txt", "Folder": []}]
        app.build_indexes(build_words=False)
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})), mock.patch.object(app, "reader_short_id", side_effect=AssertionError("request must use index")):
            response = self.client.get("/api/reader-resolve", params={"id": "0ym7e7wbcxau0"})
        self.assertEqual(response.status_code, 200)
        self.assertTrue(response.json()["url"].endswith("/book29.txt"))
        app.records = []
        app.build_indexes(build_words=False)
        response = self.client.get("/api/reader-resolve", params={"id": "0ym7e7wbcxau0"})
        self.assertEqual(response.status_code, 404)

    def test_reader_source_id_normalizes_frontend_and_backend_paths(self):
        folders = ["马列之声ebook小组第九批成果集成(2018.9.1)",
                   "马列之声ebook小组第九批成果集成(2018.9.1)part 1a", "2018.8 part 1"]
        record = {"Repo": "VoiceOfML/VOMEBOOK", "Folder": folders,
                  "File": "20世纪的社会主义 刘俊奇著 2007.09_c__", "Extension": "pdf"}
        canonical = app.build_record_link(record)
        for link in (canonical, canonical.replace("%28", "(").replace("%29", ")")):
            app.records = [{**record, "Link": link}]
            app.build_indexes(build_words=False)
            with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
                response = self.client.get("/api/reader-resolve", params={"id": "398vk0yyy8m29"})
                self.assertEqual(response.status_code, 200)
                self.assertEqual(response.json()["download"], link)
                self.assertEqual(self.client.get("/api/reader-resolve", params={"id": "1np0zrhual4h1"}).status_code, 404)
        # Different spellings of one source are not a hash collision.
        app.records.append({**record, "Link": canonical})
        app.build_indexes(build_words=False)
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
            self.assertEqual(self.client.get("/api/reader-resolve", params={"id": "398vk0yyy8m29"}).status_code, 200)

    def test_reader_resolve_uses_cached_assets_without_waiting_for_refresh(self):
        app.records = [{"Repo": "VoiceOfML/Test", "File": "book29", "Extension": "txt", "Folder": []}]
        app.build_indexes(build_words=False)
        assets = mock.AsyncMock(return_value={"v": 1, "f": {}})
        with mock.patch.object(app, "get_reader_assets", assets), mock.patch.object(
                app, "reader_assets_cache", {"v": 1, "f": {"ready": {"s": 4}}}):
            response = self.client.get("/api/reader-resolve", params={"id": "0ym7e7wbcxau0"})
        self.assertEqual(response.status_code, 200)
        assets.assert_awaited_once_with(wait=False)

    def test_reader_source_id_collision_is_not_arbitrarily_resolved(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": name, "Extension": "txt", "Folder": []}
            for name in ("first", "second")
        ]
        with mock.patch.object(app, "reader_short_id", return_value="0" * 13):
            app.build_indexes(build_words=False)
        response = self.client.get("/api/reader-resolve", params={"id": "0" * 13})
        self.assertEqual(response.status_code, 409)

    def test_reader_resolve_uses_native_asset_file_extension(self):
        for extension in ("epub", "mobi", "azw", "azw3", "fb2"):
            with self.subTest(extension=extension):
                app.records = [{"Repo": "VoiceOfML/Test", "File": "book", "Extension": "txt", "Folder": [], "Size": 10, "HasTxt": False}]
                app.build_indexes()
                asset = {"s": 2, "m": "e", "p": f"objects/aa/{'a' * 64}/document.{extension}"}
                sidecar = {"v": 1, "f": {"VoiceOfML/Test\0book.txt": asset}}
                with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
                    result = self.client.get("/api/reader-resolve", params={"id": app.reader_short_id(app.records[0].get("Link") or app.build_record_link(app.records[0]))})
                self.assertEqual(result.status_code, 200)
                self.assertEqual(result.json()["extension"], extension)

        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "scan", "Extension": "pdf", "Folder": [], "Size": 10, "HasTxt": False},
            {"Repo": "VoiceOfML/Test", "File": "archive", "Extension": "zip", "Folder": [], "Size": 10, "HasTxt": True},
        ]
        app.build_indexes()
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
            response = self.client.get("/api/random-reader")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["File"], "scan")

    def test_random_reader_uses_native_media_extensions(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "speech", "Extension": "mp3", "Folder": [], "Size": 10, "HasTxt": False},
        ]
        app.build_indexes()
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
            response = self.client.get("/api/random-reader")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["Extension"], "mp3")

    def test_random_reader_uses_original_html(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "page", "Extension": "html", "Folder": [], "Size": 10, "HasTxt": False},
        ]
        app.build_indexes()
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
            response = self.client.get("/api/random-reader")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["Extension"], "html")

    def test_random_reader_includes_native_foliate_formats(self):
        for extension in ("mobi", "azw", "azw3", "fb2", "fbz"):
            with self.subTest(extension=extension):
                app.records = [{"Repo": "VoiceOfML/Test", "File": "book", "Extension": extension,
                                "Folder": [], "Size": 10, "HasTxt": False}]
                app.build_indexes()
                with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
                    response = self.client.get("/api/random-reader?repo=Test")
                    status = self.client.get("/api/random-reader/status?repo=Test")
                self.assertEqual(response.status_code, 200)
                self.assertEqual(response.json()["Extension"], extension)
                self.assertEqual(status.json(), {"available": True, "count": 1})

    def test_random_reader_includes_ready_converted_assets_and_original_download(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "source", "Extension": "doc", "Folder": ["docs"], "Size": 10, "HasTxt": False},
        ]
        app.build_indexes()
        asset = {"s": 2, "m": "p", "p": "derived/test/" + "a" * 32 + "/document.pdf", "b": "vomebook/pdf-pages-v2"}
        sidecar = {"v": 1, "f": {"VoiceOfML/Test\0docs/source.doc": asset}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            response = self.client.get("/api/random-reader", params={"repo": "Test"})
            status = self.client.get("/api/random-reader/status", params={"repo": "Test"})
        result = response.json()
        self.assertEqual(response.status_code, 200)
        self.assertEqual(result["ReaderExtension"], "pdf")
        self.assertTrue(result["ReaderLink"].endswith(asset["p"]))
        self.assertEqual(result["DownloadLink"], result["Link"])
        self.assertEqual(status.json(), {"available": True, "count": 1})

    def test_random_reader_pool_matches_status_with_overlap_and_repository_scope(self):
        app.records = [
            {"Repo": f"VoiceOfML/{repo}", "File": name, "Extension": extension,
             "Folder": [], "Size": 10, "HasTxt": False}
            for repo, name, extension in [
                ("Test", "native", "pdf"), ("Test", "converted", "doc"),
                ("Test", "pending", "bin"), ("Other", "other", "pdf"),
            ]
        ]
        app.build_indexes()
        asset = {"s": 2, "m": "p", "p": "derived/test/" + "a" * 32 + "/document.pdf", "b": "vomebook/pdf-pages-v2"}
        sidecar = {"v": 1, "f": {
            "VoiceOfML/Test\0native.pdf": asset,
            "VoiceOfML/Test\0converted.doc": asset,
            "VoiceOfML/Test\0pending.bin": {"s": 1},
            "VoiceOfML/Test\0missing.doc": asset,
            "VoiceOfML/Other\0other.pdf": asset,
        }}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            for repo, expected in [(None, [0, 3, 1]), ("Test", [0, 1]), ("Other", [3]), ("Missing", [])]:
                with self.subTest(repo=repo), mock.patch.object(app.random, "choice", side_effect=lambda pool: pool[-1]) as choice:
                    params = {"repo": repo} if repo else {}
                    response = self.client.get("/api/random-reader", params=params)
                    status = self.client.get("/api/random-reader/status", params=params)
                    self.assertEqual(status.json(), {"available": bool(expected), "count": len(expected)})
                    self.assertEqual(response.status_code, 200 if expected else 404)
                    if expected:
                        choice.assert_called_once_with(expected)
                    else:
                        choice.assert_not_called()

    def test_random_reader_preserves_converted_reader_extension_semantics(self):
        cases = [
            ("pdf-source", "bin", "p", "pdf"),
            ("epub-source", "epub", "e", "epub"),
            ("mobi-source", "mobi", "e", "mobi"),
            ("azw3-source", "azw3", "e", "azw3"),
            ("fb2-source", "fb2", "e", "fb2"),
            ("docx-source", "doc", "d", "docx"),
            ("html-source", "html", "h", "html"),
            ("audio-source", "mp3", "a", "audio"),
            ("video-source", "mp4", "v", "video"),
        ]
        for name, original_extension, asset_mode, reader_extension in cases:
            with self.subTest(original_extension=original_extension, asset_mode=asset_mode):
                app.records = [
                    {"Repo": "VoiceOfML/Test", "File": name, "Extension": original_extension, "Folder": [], "Size": 10, "HasTxt": False},
                ]
                app.build_indexes()
                asset_paths = {
                    "p": "derived/test/" + "a" * 32 + "/document.pdf",
                    "e": f"chapters/ebook/{original_extension}/{('a' * 64)}/{'b' * 16}/chapter-manifest.json",
                    "d": "documents/office/docx/" + "a" * 64 + "/document.docx",
                    "h": "documents/web/html/" + "a" * 64 + "/document.html",
                    "a": "media/audio/mp3/" + "a" * 64 + "/audio.mp3",
                    "v": "media/video/mp4/" + "a" * 64 + "/video.mp4",
                }
                asset_path = asset_paths[asset_mode]
                asset = {"s": 2, "m": asset_mode, "p": asset_path,
                         "b": "vomebook/pdf-pages-v2" if asset_mode == "p" else "vomebook/reader-assets-v2"}
                sidecar = {"v": 1, "f": {f"VoiceOfML/Test\0{name}.{original_extension}": asset}}
                with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
                    response = self.client.get("/api/random-reader", params={"repo": "Test"})
                result = response.json()
                self.assertEqual(response.status_code, 200)
                self.assertEqual(result["ReaderExtension"], reader_extension)
                self.assertTrue(result["ReaderLink"].endswith(asset_path))
                self.assertEqual(result["DownloadLink"], result["Link"])

    def test_reader_resolve_keeps_original_title_and_download_for_converted_asset(self):
        original_link = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/books/original.mobi"
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "原始书名", "Extension": "mobi", "Folder": ["books"], "Size": 10, "HasTxt": False, "Link": original_link},
        ]
        app.build_indexes()
        asset_path = "objects/aa/" + "a" * 64 + "/document.epub"
        asset = {"s": 2, "m": "e", "p": asset_path}
        sidecar = {"v": 1, "f": {"VoiceOfML/Test\0books/原始书名.mobi": asset}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            random_result = self.client.get("/api/random-reader", params={"repo": "Test"}).json()
            reader_url = random_result["ReaderLink"]
            reader_id = app.reader_short_id(original_link)
            resolved = self.client.get("/api/reader-resolve", params={"id": reader_id})
        self.assertEqual(resolved.status_code, 200)
        data = resolved.json()
        self.assertEqual(data["title"], "原始书名")
        self.assertEqual(data["original_extension"], "mobi")
        self.assertEqual(data["extension"], "epub")
        self.assertEqual(data["download"], original_link)

    def test_reader_resolve_upgrades_original_pdf_id_to_page_stream(self):
        original_link = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/books/scan.pdf"
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "scan", "Extension": "pdf", "Folder": ["books"], "Size": 10, "HasTxt": False, "Link": original_link},
        ]
        app.build_indexes()
        asset_path = "objects/aa/" + "a" * 64 + "/page-manifest.json"
        sidecar = {"v": 1, "f": {"VoiceOfML/Test\0books/scan.pdf": {"s": 2, "m": "p", "p": asset_path}}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            response = self.client.get("/api/reader-resolve", params={"id": app.reader_short_id(original_link)})
        self.assertEqual(response.status_code, 200)
        data = response.json()
        self.assertEqual(data["extension"], "pdf-pages")
        self.assertTrue(data["url"].endswith(asset_path))
        self.assertEqual(data["download"], original_link)

    def test_reader_resolve_deduplicates_metadata_for_the_same_converted_file(self):
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "second", "Extension": "mobi", "Folder": ["z"], "Size": 10, "HasTxt": False,
             "Link": "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/z/second.mobi"},
            {"Repo": "VoiceOfML/Test", "File": "first", "Extension": "mobi", "Folder": ["a"], "Size": 10, "HasTxt": False,
             "Link": "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/a/first.mobi"},
        ]
        app.build_indexes()
        asset_path = "objects/aa/" + "a" * 64 + "/document.mobi"
        asset = {"s": 2, "m": "e", "p": asset_path}
        sidecar = {"v": 1, "f": {
            "VoiceOfML/Test\0z/second.mobi": asset,
            "VoiceOfML/Test\0a/first.mobi": asset,
        }}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            response = self.client.get("/api/reader-resolve", params={"id": app.reader_short_id(app.records[0]["Link"])})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.json()["title"], "second")
        self.assertTrue(response.json()["url"].endswith(asset_path))

    def test_reader_resolve_rejects_shared_asset_ids(self):
        app.records = [
            {"Repo": "VoiceOfML/Teachers", "File": "胡适经典全集 套装44册", "Extension": "mobi", "Folder": ["其他", "胡适"], "Size": 10, "HasTxt": False},
            {"Repo": "VoiceOfML/MLMRL-Hub", "File": "1870520043_胡适经典全集", "Extension": "mobi", "Folder": ["001584"], "Size": 10, "HasTxt": False},
        ]
        app.build_indexes()
        asset_path = "objects/aa/" + "a" * 64 + "/foliate-original-v1/document.mobi"
        asset = {"s": 2, "m": "e", "p": asset_path, "c": asset_path.replace("document.mobi", "epub-chapters/chapter-manifest.json")}
        sidecar = {"v": 1, "f": {app.build_reader_asset_key(record): asset for record in app.records}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            self.assertEqual(self.client.get("/api/reader-resolve", params={"id": "a" * 16}).status_code, 422)

    def test_random_reader_exposes_chapter_manifest_and_fallback_for_converted_book(self):
        original_link = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/books/source.epub"
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "source", "Extension": "epub", "Folder": [], "Size": 10, "HasTxt": False, "Link": original_link},
        ]
        app.build_indexes()
        chapter_manifest = "chapters/ebook/epub/" + "a" * 64 + "/1234567890abcdef/chapter-manifest.json"
        asset = {"s": 2, "m": "e", "p": chapter_manifest, "b": "vomebook/reader-assets-v2",
                 "c": chapter_manifest, "cb": "vomebook/reader-assets-v2"}
        sidecar = {"v": 1, "f": {"VoiceOfML/Test\0source.epub": asset}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            response = self.client.get("/api/random-reader", params={"repo": "Test"})
        result = response.json()
        self.assertEqual(response.status_code, 200)
        self.assertEqual(result["ReaderExtension"], "epub-chapters")
        self.assertTrue(result["ReaderChapterManifest"].endswith(chapter_manifest))
        self.assertNotIn("ReaderFallback", result)
        self.assertEqual(result["DownloadLink"], original_link)

    def test_reader_resolve_exposes_chapter_metadata_for_id_reader(self):
        original_link = "https://huggingface.co/datasets/VoiceOfML/Test/resolve/main/books/source.epub"
        app.records = [
            {"Repo": "VoiceOfML/Test", "File": "source", "Extension": "epub", "Folder": [], "Size": 10, "HasTxt": False, "Link": original_link},
        ]
        app.build_indexes()
        chapter_manifest = "chapters/ebook/epub/" + "a" * 64 + "/1234567890abcdef/chapter-manifest.json"
        asset = {"s": 2, "m": "e", "p": chapter_manifest, "b": "vomebook/reader-assets-v2",
                 "c": chapter_manifest, "cb": "vomebook/reader-assets-v2"}
        sidecar = {"v": 1, "f": {"VoiceOfML/Test\0source.epub": asset}}
        with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=sidecar)):
            response = self.client.get("/api/reader-resolve", params={"id": app.reader_short_id(original_link)})
        self.assertEqual(response.status_code, 200)
        result = response.json()
        self.assertEqual(result["extension"], "epub-chapters")
        self.assertTrue(result["chapter_manifest"].endswith(chapter_manifest))
        self.assertNotIn("fallback", result)

    def test_reader_static_assets_are_served(self):
        for path in ("/static/reader.html", "/static/reader.js", "/static/reader.css", "/static/reader/core/reader-contract.js"):
            self.assertEqual(self.client.get(path).status_code, 200)

    def test_unused_reader_metadata_alias_is_retired(self):
        self.assertEqual(self.client.get('/api/reader-metadata', params={'id': '0' * 13}).status_code, 404)

    def test_random_status_empty_repo_is_404(self):
        self.assertEqual(self.client.get("/api/random-txt", params={"repo": "Missing"}).status_code, 404)

    def test_ping_is_empty_and_not_cached(self):
        response = self.client.get("/api/ping")
        self.assertEqual(response.status_code, 204)
        self.assertEqual(response.headers.get("cache-control"), "no-store")

    def test_cors_header_is_present(self):
        response = self.client.get("/api/repos", headers={"Origin": "https://example.test"})
        self.assertEqual(response.headers.get("access-control-allow-origin"), "*")

    def test_gzip_compresses_large_response_when_requested(self):
        response = self.client.get("/static/app.js", headers={"Accept-Encoding": "gzip"})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.headers.get("content-encoding"), "gzip")

    def test_only_content_hashed_static_assets_are_immutable(self):
        from fastapi import FastAPI
        with tempfile.TemporaryDirectory() as directory:
            static_dir = Path(directory)
            (static_dir / "app.js").write_text("plain", encoding="utf-8")
            (static_dir / "app.0123456789ab.js").write_text("hashed", encoding="utf-8")
            static_app = FastAPI()
            static_app.mount("/static", app.HashedStaticFiles(directory=directory), name="static-test")
            with TestClient(static_app) as client:
                plain = client.get("/static/app.js")
                hashed = client.get("/static/app.0123456789ab.js")
            self.assertNotIn("immutable", plain.headers.get("cache-control", ""))
            self.assertEqual(hashed.headers.get("cache-control"), "public, max-age=31536000, immutable")

    def test_unhashed_static_html_shell_always_revalidates(self):
        from fastapi import FastAPI
        with tempfile.TemporaryDirectory() as directory:
            (Path(directory) / "reader.html").write_text("<p>shell</p>", encoding="utf-8")
            static_app = FastAPI()
            static_app.mount("/static", app.HashedStaticFiles(directory=directory), name="static-test")
            with TestClient(static_app) as client:
                first = client.get("/static/reader.html")
                again = client.get("/static/reader.html", headers={"If-None-Match": first.headers["etag"]})
            self.assertEqual(first.headers.get("cache-control"), "no-cache")
            self.assertEqual(again.status_code, 304)
            self.assertEqual(again.headers.get("cache-control"), "no-cache")

    def test_root_serves_html_and_injects_initial_payload(self):
        response = self.client.get("/")
        self.assertEqual(response.status_code, 200)
        self.assertIn("initial-search-data", response.text)
        self.assertIn("/static/app.js", response.text)

    def test_spa_route_serves_html(self):
        response = self.client.get("/Test/some/client/route")
        self.assertEqual(response.status_code, 200)
        self.assertIn("VoiceOfML Search", response.text)

    def test_unknown_repository_spa_route_is_404(self):
        response = self.client.get("/Missing/some/client/route")
        self.assertEqual(response.status_code, 404)
        self.assertEqual(response.text, "Repository not found")

    def test_api_paths_return_json_errors_and_search_get_is_405(self):
        unknown = self.client.get("/api/does-not-exist")
        self.assertEqual(unknown.status_code, 404)
        self.assertEqual(unknown.headers["content-type"].split(";")[0], "application/json")
        self.assertEqual(unknown.json(), {"detail": "Not Found"})
        get_search = self.client.get("/api/search")
        self.assertEqual(get_search.status_code, 405)
        self.assertEqual(get_search.headers["allow"], "POST")
        self.assertEqual(self.client.put("/api/search", json={}).status_code, 405)

    def test_missing_frontend_shell_is_service_unavailable(self):
        with mock.patch.object(app, "injected_html_cache", {}), mock.patch.object(app, "Path") as path_class:
            path_class.return_value.exists.return_value = False
            root = self.client.get("/")
            spa = self.client.get("/Test/some/client/route")
        self.assertEqual(root.status_code, 503)
        self.assertEqual(spa.status_code, 503)
        self.assertEqual(root.text, "Frontend shell unavailable")

    def test_static_manifest_and_service_worker_routes(self):
        self.assertEqual(self.client.get("/manifest.json").status_code, 200)
        self.assertEqual(self.client.get("/sw.js").status_code, 200)

    def test_initial_manifest_contains_global_and_repo_urls(self):
        payload = self.client.get("/data/initial/manifest.json").json()
        self.assertEqual(payload["urls"], ["/data/initial/global.json", "/data/initial/repos/Test.json"])

    def test_sidebar_missing_repo_is_404(self):
        self.assertEqual(self.client.get("/data/sidebar/repos/Missing.json").status_code, 404)


if __name__ == "__main__":
    unittest.main()