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_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_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() self.assertEqual(payload["generation"], app._data_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 = "objects/aa/" + "a" * 64 + "/epub-chapters-v1/chapter-manifest.json" fallback = "objects/bb/" + "b" * 64 + "/epub-pdf-v1/document.pdf" asset = {"s": 2, "m": "e", "p": "objects/aa/" + "a" * 64 + "/book.epub", "c": chapter_manifest, "f": fallback} 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.assertTrue(result["ReaderFallback"].endswith(fallback)) 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 = "objects/aa/" + "a" * 64 + "/epub-chapters-v1/chapter-manifest.json" fallback = "objects/bb/" + "b" * 64 + "/epub-pdf-v1/document.pdf" asset = {"s": 2, "m": "e", "p": "objects/aa/" + "a" * 64 + "/book.epub", "c": chapter_manifest, "f": fallback} 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.assertTrue(result["fallback"].endswith(fallback)) 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_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_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()