"""Tiny real SQLite/text candidates; never open/rebuild production databases.""" import copy import gzip import hashlib import json from pathlib import Path import tempfile import unittest from unittest.mock import patch import app import build_fulltext_db as builder import ocr_release as release from fastapi.testclient import TestClient class OcrReleaseTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) self.bucket = self.root / "bucket" self.bucket.mkdir() self.base = self.root / "base.json.gz" base_record = {"doc_id": "CCRD:1", "source": "CCRD", "source_name": "CCRD", "storage_root": "CCRD", "storage_rel_path": "old.txt", "display_name": "旧文件", "display_dirs": [], "display_rel_path": "旧文件.txt", "size": 7, "extension": "txt"} self.base.write_bytes(gzip.compress(json.dumps({"records": [base_record], "sources": [ {"slug": "CCRD", "name": "CCRD", "count": 1}]}).encode(), mtime=0)) self.base_sha = hashlib.sha256(self.base.read_bytes()).hexdigest() self.content = "手机文化\n\n\n\n末页手。机 ABC" texts = ["手机文化", "", "末页手。机 ABC"] body = self.content.encode() txt_sha = release.digest(body) text_ref = {"bucket": release.BUCKET, "path": f"text/objects/{txt_sha[:2]}/{txt_sha}.txt", "sha256": txt_sha, "bytes": len(body), "role": "runtime"} self.write(text_ref["path"], body) original = "原目录/book.pdf" identity = {"recipe": "ocr-plain-utf8-pages-v1", "origin_repo": "VoiceOfML/Test", "origin_rel_path": original, "origin_revision": "b" * 40, "origin_sha256": "a" * 64, "source_sha256": "a" * 64, "doc_id": "OCR:1", "text_generation": "c" * 64, "txt_sha256": txt_sha} generation = release.digest(release.canonical(identity).rstrip(b"\n")) book_id = release.digest(("VoiceOfML/Test\0" + original).encode()) pages, cursor = [], 0 for number, text in enumerate(texts, 1): pages.append({"page": number, "start": cursor, "end": cursor + len(text), "text_sha256": release.digest(text.encode()), "text_generation": str(number) * 64, "quality": "partially-reviewed" if number == 1 else "unreviewed"}) cursor += len(text) + 2 self.export = {"version": 1, "kind": "ccrd-ocr-text-export", "complete": True, **identity, "generation": generation, "book_id": book_id, "title": "book", "page_count": 3, "source_sha256_role": "reader-primary", "readable_rel_path": "VoiceOfML/Test/原目录/book.txt", "accepted_pages": 1, "quality": "partially-reviewed", "offset_unit": "unicode-codepoint", "page_separator": "\n\n", "pages": pages, "text": text_ref, "retained_reader_resources": [{"bucket": "vomebook/pdf-pages-v2", "path": "objects/aa/reading-manifest.json", "sha256": "a" * 64, "bytes": 1, "role": "provenance"}]} self.export_path = f"text/exports/{book_id}/{generation}/manifest.json" self.export_ref = self.ref(self.export_path, release.canonical(self.export)) core = {"version": 1, "kind": "ccrd-ocr-search-release", "base_metadata_sha256": self.base_sha, "builder_sha256": release.digest(Path(builder.__file__).read_bytes()), "tokenizer_version": release.TOKENIZER, "exports": [self.export_ref]} generation = release.digest(release.canonical(core)) records = [release.export_record(self.bucket, self.export_ref)[0]] db_relative = f"indexes/ocr/{generation}/OCR.sqlite3" directory = self.bucket / Path(db_relative).parent directory.mkdir(parents=True) with patch.object(builder, "BASE_DIR", self.bucket), patch.object(builder, "FULLTEXT_DIR", directory): builder.build_source_db("OCR", records) self.manifest = {**core, "generation": generation, "documents": 1, "database": self.ref(db_relative, (self.bucket / db_relative).read_bytes())} self.manifest_path = f"releases/ocr-{generation}/manifest.json" raw = release.canonical(self.manifest) self.write(self.manifest_path, raw) self.manifest_sha = release.digest(raw) def write(self, path, body): target = self.bucket / path target.parent.mkdir(parents=True, exist_ok=True) target.write_bytes(body) def ref(self, path, body): self.write(path, body) return {"bucket": release.BUCKET, "path": path, "sha256": release.digest(body), "bytes": len(body), "role": "runtime"} def load(self): return release.load_release(self.bucket, self.manifest_path, self.manifest_sha, self.base_sha) def test_whole_release_is_verified_and_origin_hierarchy_is_preserved(self): candidate = self.load() self.assertEqual(candidate["records"][0]["display_rel_path"], "VoiceOfML/Test/原目录/book.txt") self.assertEqual(candidate["text_paths"]["OCR:1"].read_text(), self.content) self.assertEqual(candidate["tree"][0]["count"], 1) self.assertEqual(candidate["browser"]["VoiceOfML/Test/原目录"]["files"][0]["doc_id"], "OCR:1") def test_corrupt_text_or_database_blocks_the_release(self): for ref in (self.export["text"], self.manifest["database"]): target = self.bucket / ref["path"] original = target.read_bytes() try: target.write_bytes(original + b"damage") with self.assertRaisesRegex(ValueError, "checksum"): self.load() finally: target.write_bytes(original) def test_wrong_manifest_or_base_identity_cannot_activate(self): for manifest_sha, base_sha in (("d" * 64, self.base_sha), (self.manifest_sha, "e" * 64)): with self.subTest(manifest_sha=manifest_sha), self.assertRaises(ValueError): release.load_release(self.bucket, self.manifest_path, manifest_sha, base_sha) def test_export_page_offsets_and_missing_provenance_are_rejected(self): for change in ({"pages": []}, {"retained_reader_resources": []}, {"readable_rel_path": "wrong/book.txt"}, {"accepted_pages": 3}, {"origin_repo": "../bad"}): value = {**copy.deepcopy(self.export), **change} ref = self.ref(self.export_path, release.canonical(value)) with self.subTest(change=list(change)), self.assertRaises(ValueError): release.export_record(self.bucket, ref) def test_pin_copies_verified_data_and_is_independent_of_the_mount(self): pinned = release.pin_release(self.bucket, self.manifest_path, self.manifest_sha, self.base_sha, self.root / "cache") self.assertNotIn(self.bucket, pinned["database"].parents) self.assertEqual(pinned["text_paths"]["OCR:1"].read_text(), self.content) self.write(self.export["text"]["path"], b"changed remote mount") self.assertEqual(pinned["text_paths"]["OCR:1"].read_text(), self.content) def test_small_index_uses_existing_in_memory_search_contract(self): candidate = self.load() databases = app.FulltextDatabases(self.root / "empty-db", in_memory=True, extra_paths=[candidate["database"]]) try: self.assertEqual(databases.search_source("OCR", "手机", exact=False), {1}) self.assertEqual(databases.search_source("OCR", "手。机", exact=True), {1}) self.assertEqual(databases.search_source("OCR", "文化 ABC", exact=False), {1}) self.assertEqual(databases.search_source("OCR", "手机 ABC", exact=True), set()) self.assertEqual(databases.connections["OCR"].execute("PRAGMA database_list").fetchone()[2], "") finally: databases.close() def test_startup_loads_one_pinned_release_and_serves_plain_txt(self): tree, browser = self.root / "tree.gz", self.root / "browser.gz" tree.write_bytes(gzip.compress(b'{"CCRD":[]}')); browser.write_bytes(gzip.compress(b'{"CCRD":{}}')) saved = {key: getattr(app, key) for key in ("records", "record_map", "record_map_index", "sources", "source_counts", "folder_tree_data", "folder_browser_data", "active_ocr_release")} try: with patch.dict("os.environ", {"CCRD_INDEX_BUCKET_DIR": str(self.bucket), "CCRD_OCR_RELEASE_PATH": self.manifest_path, "CCRD_OCR_RELEASE_SHA256": self.manifest_sha}), \ patch.object(app, "DATA_PATH", self.base), patch.object(app, "FOLDER_TREE_PATH", tree), \ patch.object(app, "FOLDER_BROWSER_PATH", browser), patch.object(app, "BASE_DIR", self.root): app.load_data() self.assertEqual(app.source_counts, {"CCRD": 1, "OCR": 1}) self.assertEqual(app.record_map["CCRD:1"]["display_name"], "旧文件") client = TestClient(app.app) response = client.get("/txt/OCR/VoiceOfML/Test/原目录/book.txt") self.assertEqual(response.status_code, 200) self.assertEqual(response.text, self.content) self.assertIn("inline", response.headers["content-disposition"]) self.assertEqual(client.get("/txt/OCR/missing.txt").status_code, 404) self.assertEqual(client.get("/api/preview/OCR:1").json()["text"], self.content) self.assertEqual(client.get("/api/preview/OCR:1").json()["text_quality"], "partially-reviewed") self.assertEqual(json.loads(app.sidebar_payload_for_path(""))["sources"][-1]["slug"], "OCR") self.assertEqual(app.zip_entries(["OCR:1"])[0][0], "OCR成果/VoiceOfML/Test/原目录/book.txt") public = app.trim_record(app.record_map["OCR:1"]) self.assertTrue(public["TextUrl"].startswith("/txt/OCR/VoiceOfML/Test/")) self.assertEqual(public["TextQuality"], "partially-reviewed") self.assertEqual(client.get("/data/sidebar/global.json").json()["sources"][-1]["slug"], "OCR") self.assertEqual(client.get("/data/initial/sources/OCR.json").json()["total"], 1) finally: for key, value in saved.items(): setattr(app, key, value) app.build_indexes() if __name__ == "__main__": unittest.main()