from __future__ import annotations import gzip import hashlib import json import unittest from collections import Counter, defaultdict from pathlib import Path ROOT = Path(__file__).resolve().parents[1] DATA = ROOT / "data" FIELDS = ("Repo", "File", "Extension", "Folder", "Size", "HasTxt") WORD_INDEX_TOKENIZER = "jieba-0.42.1-cjk-char-v1" def decode_varint(data, offset): value = 0 shift = 0 while offset < len(data): byte = data[offset] offset += 1 value |= (byte & 0x7F) << shift if not byte & 0x80: return value, offset shift += 7 raise ValueError("truncated varint") def corpus_digest(records): digest = hashlib.sha256() for record in records: values = ( str(record.get("File") or "").lower(), str(record.get("Repo") or "").lower(), "/".join(str(item).lower() for item in (record.get("Folder") or [])), ) for value in values: digest.update(value.encode("utf-8")) digest.update(b"\0") return digest.digest() def validate_posting_entries(data, offset, expected_count=None): count, offset = decode_varint(data, offset) if expected_count is not None: assert count == expected_count for _ in range(count): key_length, offset = decode_varint(data, offset) data[offset:offset + key_length].decode("utf-8") offset += key_length _, offset = decode_varint(data, offset) posting_length, offset = decode_varint(data, offset) offset += posting_length if offset > len(data): raise ValueError("truncated posting data") if offset != len(data): raise ValueError("trailing posting data") return count def decode_corpus_independently(): with gzip.open(DATA / "search_data.json.gz", "rt", encoding="utf-8") as handle: compact = json.load(handle) repos = compact.get("rp", []) folders = compact.get("fd", []) decoded = [] for row in compact.get("rc", []): if not isinstance(row, list) or len(row) < 6: continue repo = repos[row[0]] if isinstance(row[0], int) and 0 <= row[0] < len(repos) else "" folder = folders[row[3]] if isinstance(row[3], int) and 0 <= row[3] < len(folders) else [] decoded.append({"Repo": repo, "File": row[1], "Extension": row[2], "Folder": folder, "Size": row[4], "HasTxt": bool(row[5])}) return decoded class GeneratedDataOracleTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.records = decode_corpus_independently() cls.by_repo = defaultdict(list) for record in cls.records: cls.by_repo[record["Repo"]].append(record) def test_compact_corpus_has_expected_record_count(self): with gzip.open(DATA / "search_data.json.gz", "rt", encoding="utf-8") as handle: rows = json.load(handle)["rc"] self.assertGreater(len(rows), 0) self.assertEqual(len(self.records), len(rows), "decoder must not silently drop malformed rows") def test_sidebar_global_counts_equal_independent_corpus_counts(self): sidebar = json.loads((DATA / "sidebar/global.json").read_text(encoding="utf-8")) expected = Counter(record["Repo"] for record in self.records) self.assertEqual({item["name"]: item["count"] for item in sidebar["repos"]}, dict(expected)) def test_global_initial_total_and_ordered_first_page_equal_corpus(self): payload = json.loads((DATA / "initial/global.json").read_text(encoding="utf-8")) self.assertEqual(payload["total"], len(self.records)) self.assertEqual(payload["results"], self.records[:payload["page_size"]]) def test_each_repo_initial_total_and_ordered_page_equal_corpus(self): for path in sorted((DATA / "initial/repos").glob("*.json")): with self.subTest(repo_file=path.name): payload = json.loads(path.read_text(encoding="utf-8")) records = self.by_repo[payload["repo"]] self.assertEqual(payload["total"], len(records)) self.assertEqual(payload["results"], records[:payload["page_size"]]) def test_initial_manifest_exactly_lists_generated_payload_files(self): manifest = json.loads((DATA / "initial/manifest.json").read_text(encoding="utf-8")) actual = {"/data/initial/global.json"} | {f"/data/initial/repos/{path.name}" for path in (DATA / "initial/repos").glob("*.json")} self.assertEqual(set(manifest["urls"]), actual) def test_sidebar_manifest_exactly_lists_generated_payload_files(self): manifest = json.loads((DATA / "sidebar/manifest.json").read_text(encoding="utf-8")) actual = {"/data/sidebar/global.json"} | {f"/data/sidebar/repos/{path.name}" for path in (DATA / "sidebar/repos").glob("*.json")} self.assertEqual(set(manifest["urls"]), actual) def test_each_repo_sidebar_identity_matches_file_and_corpus_total(self): for path in sorted((DATA / "sidebar/repos").glob("*.json")): with self.subTest(repo_file=path.name): payload = json.loads(path.read_text(encoding="utf-8")) expected_repo = f"VoiceOfML/{path.stem}" self.assertEqual(payload.get("repo"), expected_repo) direct_and_nested = sum(item.get("count", 0) for item in payload.get("folders", payload.get("d", []))) files = payload.get("files", payload.get("f", [])) self.assertLessEqual(len(files), len(self.by_repo[expected_repo])) self.assertGreaterEqual(direct_and_nested + len(files), 0) def test_binary_search_indexes_match_corpus_generation(self): expected_digest = corpus_digest(self.records) words = gzip.decompress((DATA / "search_words.bin.gz").read_bytes()) self.assertTrue(words.startswith(b"VWI2")) offset = 4 tokenizer_length, offset = decode_varint(words, offset) self.assertEqual(words[offset:offset + tokenizer_length].decode("utf-8"), WORD_INDEX_TOKENIZER) offset += tokenizer_length self.assertEqual(words[offset:offset + 32], expected_digest) offset += 32 record_count, offset = decode_varint(words, offset) self.assertEqual(record_count, len(self.records)) self.assertGreater(validate_posting_entries(words, offset), 0) for width in (2, 3): data = gzip.decompress((DATA / f"search_ngrams_{width}.bin.gz").read_bytes()) self.assertEqual(data[:4], b"VNG3") self.assertEqual(data[4], width) self.assertEqual(data[5:37], expected_digest) record_count, offset = decode_varint(data, 37) self.assertEqual(record_count, len(self.records)) self.assertGreater(validate_posting_entries(data, offset), 0) if __name__ == "__main__": unittest.main()