File size: 6,931 Bytes
2ecdf7d 5492cc8 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 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 | 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()
|