| 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() |
|
|
