tests/test_api_contract.py

from __future__ import annotations

import copy
import asyncio
import json
import unittest
from unittest import mock

IMPORT_ERROR = None
try:
    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_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": "objects/aa/book.epub"}}}
        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"', payload)
        self.assertIn('"ReaderLink":"https://huggingface.co/datasets/vomebook/Reader-Assets/resolve/main/objects/aa/book.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()