File size: 6,200 Bytes
2ecdf7d 2c1d035 2ecdf7d 8e2dddb 2ecdf7d cd30830 2ecdf7d cd30830 04985c9 2ecdf7d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | from __future__ import annotations
import copy
import asyncio
import json
import unittest
from unittest import mock
IMPORT_ERROR = None
try:
from backend import app
except ModuleNotFoundError as exc: # pragma: no cover
app = None
IMPORT_ERROR = exc
@unittest.skipIf(app is None, f"app dependencies unavailable: {IMPORT_ERROR}")
class ApiContractTests(unittest.TestCase):
def setUp(self):
self.names = (
"records", "word_index", "folder_index", "repo_counts", "repo_list",
"repo_records_map", "sorted_by_name", "sorted_by_size",
"repo_sorted_by_name", "repo_sorted_by_size", "vocab_by_len",
"extension_counts", "extension_list", "repo_extension_counts",
"txt_record_indices", "repo_txt_record_indices", "reader_record_indices", "repo_reader_record_indices", "folder_tree_data",
"folder_browser_data", "api_response_cache", "initial_payload_global",
"initial_payload_by_repo",
)
self.saved = {name: copy.deepcopy(getattr(app, name)) for name in self.names}
app.records = [{"Repo": "VoiceOfML/Test", "File": "readme", "Extension": "txt", "Folder": ["docs"], "Size": 10, "HasTxt": True}]
app.build_indexes()
app.build_initial_payloads()
def tearDown(self):
for name, value in self.saved.items():
setattr(app, name, value)
def test_search_request_defaults_and_response_shape(self):
with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v": 1, "f": {}})):
body = asyncio.run(app.api_search(app.SearchRequest(q="readme", page_size=1)))
self.assertEqual(body.status_code, 200)
payload = body.body.decode()
self.assertIn('"page":1', payload)
self.assertIn('"page_size":1', payload)
self.assertIn('"results"', payload)
def test_encoded_search_cache_reuses_body_and_invalidates_asset_revision(self):
app.api_response_cache.clear()
request = app.SearchRequest(q="readme", page_size=1)
with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v":1,"f":{}})), \
mock.patch.object(app, "reader_assets_revision", "first"):
first = asyncio.run(app.api_search(request))
with mock.patch.object(app, "add_reader_assets", side_effect=AssertionError("repeat enrichment")):
second = asyncio.run(app.api_search(request))
self.assertIs(first.body, second.body)
enriched = {**json.loads(first.body), "updated":True}
with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value={"v":1,"f":{}})), \
mock.patch.object(app, "reader_assets_revision", "second"), \
mock.patch.object(app, "add_reader_assets", return_value=enriched):
updated = asyncio.run(app.api_search(request))
self.assertTrue(json.loads(updated.body)["updated"])
self.assertNotIn("readme", updated.headers["server-timing"])
def test_unknown_repo_endpoints_are_404(self):
self.assertEqual(asyncio.run(app.api_search_repo("Missing", app.SearchRequest())).status_code, 404)
self.assertEqual(app.api_folders("Missing").status_code, 404)
self.assertEqual(app.api_folder_contents("Missing").status_code, 404)
self.assertEqual(app.serve_initial_repo("Missing").status_code, 404)
def test_search_response_adds_converted_reader_asset(self):
key = "VoiceOfML/Test\0docs/readme.txt"
assets = {"v": 1, "f": {key: {"s": 2, "m": "e",
"p": "chapters/ebook/epub/" + "a" * 64 + "/" + "b" * 16 + "/chapter-manifest.json",
"b": "vomebook/reader-assets-v2",
"c": "chapters/ebook/epub/" + "a" * 64 + "/" + "b" * 16 + "/chapter-manifest.json",
"cb": "vomebook/reader-assets-v2"}}}
with mock.patch.object(app, "get_reader_assets", mock.AsyncMock(return_value=assets)):
body = asyncio.run(app.api_search(app.SearchRequest(q="readme", page_size=1)))
payload = body.body.decode()
self.assertIn('"ReaderExtension":"epub-chapters"', payload)
self.assertIn('"ReaderLink":"https://voiceofml-search.hf.space/api/reader-bucket-resource?path=chapters/ebook/epub/', payload)
def test_reader_assets_preserve_native_foliate_extensions(self):
extensions = ("epub", "mobi", "azw", "azw3", "fb2")
results = []
assets = {"v": 1, "f": {}}
for extension in extensions:
record = {
"Repo": "VoiceOfML/Test",
"File": f"book-{extension}",
"Extension": extension,
"Folder": ["docs"],
"Size": 10,
"HasTxt": False,
}
results.append(record)
assets["f"][app.build_reader_asset_key(record)] = {
"s": 2,
"m": "e",
"p": f"objects/aa/{'a' * 64}/document.{extension}",
}
converted = {
"Repo": "VoiceOfML/Test",
"File": "converted",
"Extension": "txt",
"Folder": ["docs"],
"Size": 10,
"HasTxt": True,
}
results.append(converted)
assets["f"][app.build_reader_asset_key(converted)] = {
"s": 2,
"m": "e",
"p": f"objects/bb/{'b' * 64}/document.epub",
}
enriched = app.add_reader_assets({"results": results}, assets)
actual = {item["Extension"]: item["ReaderExtension"] for item in enriched["results"]}
self.assertEqual(
{extension: extension for extension in extensions} | {"txt": "epub"},
actual,
)
self.assertEqual(app.serve_initial_manifest().body.decode(), '{"urls":["/data/initial/global.json","/data/initial/repos/Test.json"]}')
contents = app.api_folder_contents("Test", path="")
self.assertIn(b'"folders"', contents.body)
self.assertIn(b'"files"', contents.body)
if __name__ == "__main__":
unittest.main()
|