1199 lines
49 KiB
Python
1199 lines
49 KiB
Python
#!/usr/bin/env python3
|
|
"""Aggregate WWAR v1 Phase 0 conformance gate."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unicodedata
|
|
from typing import Any
|
|
|
|
import tool
|
|
|
|
|
|
ROOT = pathlib.Path(__file__).resolve().parent
|
|
SCHEMA = ROOT / "schema"
|
|
GENERATED = ROOT / "generated" / "wwar_v1_tables.json"
|
|
VECTORS = ROOT / "vectors"
|
|
EXPECTED_FILES = {
|
|
"README.md",
|
|
"check.py",
|
|
"tool.py",
|
|
"schema/digests.json",
|
|
"schema/records.json",
|
|
"schema/wire.json",
|
|
"schema/wwar-v1.sha256",
|
|
"generated/wwar_v1_tables.json",
|
|
"vectors/digests.json",
|
|
"vectors/records.json",
|
|
"vectors/wire-invalid.json",
|
|
"vectors/wire-valid.json",
|
|
}
|
|
ARCHIVED_ASSIGNMENT_SHA256 = "7b8660b451cc5260e9c7ad8d45da8e62313ff1f9b54259e211ff04b49c38671f"
|
|
|
|
|
|
def fail(message: str) -> None:
|
|
raise AssertionError(message)
|
|
|
|
|
|
def exact_keys(value: Any, expected: set[str], where: str) -> None:
|
|
if not isinstance(value, dict) or set(value) != expected:
|
|
actual = set(value) if isinstance(value, dict) else set()
|
|
fail(
|
|
f"{where}: keys differ: missing={sorted(expected-actual)} "
|
|
f"unknown={sorted(actual-expected)}"
|
|
)
|
|
|
|
|
|
def register_case_id(case: Any, seen: set[str], where: str) -> None:
|
|
if not isinstance(case, dict):
|
|
fail(f"{where}: expected object")
|
|
case_id = case.get("id")
|
|
if (
|
|
not isinstance(case_id, str)
|
|
or not case_id
|
|
or len(case_id.encode("utf-8")) > 255
|
|
or case_id in seen
|
|
):
|
|
fail(f"{where}: invalid or duplicate case id")
|
|
seen.add(case_id)
|
|
|
|
|
|
def check_files() -> None:
|
|
actual = {
|
|
path.relative_to(ROOT).as_posix()
|
|
for path in ROOT.rglob("*")
|
|
if path.is_file()
|
|
}
|
|
if actual != EXPECTED_FILES:
|
|
fail(f"protocol file set differs: missing={sorted(EXPECTED_FILES-actual)} extra={sorted(actual-EXPECTED_FILES)}")
|
|
|
|
|
|
def check_no_bytecode() -> None:
|
|
bad = [
|
|
path.relative_to(ROOT).as_posix()
|
|
for path in ROOT.rglob("*")
|
|
if path.is_file() and (path.suffix in {".pyc", ".pyo"} or "__pycache__" in path.parts)
|
|
]
|
|
if bad:
|
|
fail(f"Python bytecode under protocol/: {bad}")
|
|
|
|
|
|
def run_tool(*arguments: str, expect: int = 0) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["PYTHONDONTWRITEBYTECODE"] = "1"
|
|
env["LC_ALL"] = "C"
|
|
completed = subprocess.run(
|
|
[sys.executable, "-B", str(ROOT / "tool.py"), *arguments],
|
|
cwd=ROOT,
|
|
env=env,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.PIPE,
|
|
check=False,
|
|
)
|
|
if completed.returncode != expect:
|
|
fail(
|
|
f"tool command returned {completed.returncode}, wanted {expect}: {arguments}\n"
|
|
f"stdout={completed.stdout}\nstderr={completed.stderr}"
|
|
)
|
|
return completed
|
|
|
|
|
|
def oracle_encode(ast_value: Any, wire: dict[str, Any], depth: int = 1) -> bytes:
|
|
if depth > wire["limits"]["nesting_depth_max"]:
|
|
raise ValueError("depth")
|
|
if not isinstance(ast_value, dict) or len(ast_value) != 1:
|
|
raise ValueError("shape")
|
|
name, value = next(iter(ast_value.items()))
|
|
codes = {row["name"]: row["code"] for row in wire["wire_types"]}
|
|
if name == "bool":
|
|
payload = bytes([int(value)])
|
|
elif name == "bytes":
|
|
payload = bytes.fromhex(value)
|
|
elif name == "string":
|
|
payload = value.encode("utf-8")
|
|
elif name == "uint":
|
|
number = int(value)
|
|
payload = number.to_bytes(max(1, (number.bit_length() + 7) // 8), "big")
|
|
elif name == "list":
|
|
children = [oracle_encode(child, wire, depth + 1) for child in value]
|
|
payload = struct.pack(">I", len(children)) + b"".join(
|
|
struct.pack(">Q", len(child)) + child for child in children
|
|
)
|
|
elif name == "map":
|
|
parts = [struct.pack(">I", len(value))]
|
|
for key, child_value in value:
|
|
key_bytes = key.encode("utf-8")
|
|
child = oracle_encode(child_value, wire, depth + 1)
|
|
parts.extend((struct.pack(">Q", len(key_bytes)), key_bytes, struct.pack(">Q", len(child)), child))
|
|
payload = b"".join(parts)
|
|
elif name == "record":
|
|
parts = [struct.pack(">I", len(value))]
|
|
for tag, child_value in value:
|
|
child = oracle_encode(child_value, wire, depth + 1)
|
|
parts.extend((struct.pack(">I", tag), struct.pack(">Q", len(child)), child))
|
|
payload = b"".join(parts)
|
|
else:
|
|
raise ValueError(name)
|
|
return bytes([codes[name]]) + struct.pack(">Q", len(payload)) + payload
|
|
|
|
|
|
def oracle_envelope(ast_value: Any, wire: dict[str, Any]) -> bytes:
|
|
return bytes.fromhex(wire["magic_hex"]) + struct.pack(">H", wire["schema_version"]) + oracle_encode(ast_value, wire)
|
|
|
|
|
|
def load_vectors(
|
|
name: str, expected_format: str, expected_keys: set[str]
|
|
) -> dict[str, Any]:
|
|
value = tool.load_json(VECTORS / name)
|
|
exact_keys(value, expected_keys, name)
|
|
if value.get("format") != expected_format or value.get("version") != 1:
|
|
fail(f"{name}: bad vector format/version")
|
|
return value
|
|
|
|
|
|
def check_wire_vectors(bundle: tool.SchemaBundle) -> None:
|
|
valid = load_vectors(
|
|
"wire-valid.json", "wwar-v1-wire-valid", {"format", "version", "cases"}
|
|
)
|
|
ids: set[str] = set()
|
|
covered_types: set[str] = set()
|
|
for case in valid["cases"]:
|
|
if set(case) != {"id", "value", "wwar_hex", "sha256"} or case["id"] in ids:
|
|
fail("wire-valid: malformed or duplicate case")
|
|
ids.add(case["id"])
|
|
encoded = tool.encode_envelope(case["value"], bundle.wire)
|
|
oracle = oracle_envelope(case["value"], bundle.wire)
|
|
if encoded != oracle or encoded.hex() != case["wwar_hex"]:
|
|
fail(f"wire-valid {case['id']}: encoder/oracle/golden mismatch")
|
|
if hashlib.sha256(encoded).hexdigest() != case["sha256"]:
|
|
fail(f"wire-valid {case['id']}: SHA-256 mismatch")
|
|
if tool.decode_envelope(encoded, bundle.wire) != case["value"]:
|
|
fail(f"wire-valid {case['id']}: round-trip mismatch")
|
|
covered_types.add(next(iter(case["value"])))
|
|
if covered_types != set(bundle.type_codes):
|
|
fail(f"wire-valid: type coverage differs: {covered_types}")
|
|
|
|
invalid = load_vectors(
|
|
"wire-invalid.json", "wwar-v1-wire-invalid", {"format", "version", "cases"}
|
|
)
|
|
ids.clear()
|
|
required = {
|
|
"bytes-limit-before-truncation",
|
|
"bytes-truncated",
|
|
"bool-truncated-before-length",
|
|
"bool-length-mismatch",
|
|
"list-child-length-mismatch",
|
|
"nesting-limit",
|
|
}
|
|
for case in invalid["cases"]:
|
|
if set(case) != {"id", "input_hex", "error"} or case["id"] in ids:
|
|
fail("wire-invalid: malformed or duplicate case")
|
|
exact_keys(case["error"], {"code", "path"}, f"wire-invalid {case['id']}.error")
|
|
ids.add(case["id"])
|
|
try:
|
|
tool.decode_envelope(bytes.fromhex(case["input_hex"]), bundle.wire)
|
|
except tool.ProtocolError as exc:
|
|
if {"code": exc.code, "path": exc.path} != case["error"]:
|
|
fail(f"wire-invalid {case['id']}: got {exc.code} {exc.path}, wanted {case['error']}")
|
|
else:
|
|
fail(f"wire-invalid {case['id']}: accepted")
|
|
if not required <= ids:
|
|
fail(f"wire-invalid: missing precedence cases {sorted(required-ids)}")
|
|
|
|
|
|
def assignment_catalog(bundle: tool.SchemaBundle) -> dict[str, Any]:
|
|
return {
|
|
"scalar_types": {
|
|
row["name"]: {key: value for key, value in row.items() if key != "name"}
|
|
for row in bundle.records_data["scalar_types"]
|
|
},
|
|
"path_classes": {
|
|
row["name"]: {key: value for key, value in row.items() if key != "name"}
|
|
for row in bundle.records_data["path_classes"]
|
|
},
|
|
"records": {
|
|
record["name"]: {
|
|
"top_level_kind": record["top_level_kind"],
|
|
"fields": {
|
|
str(field["tag"]): {
|
|
key: field[key]
|
|
for key in ("name", "type", "cardinality", "encoded_default", "order", "order_by")
|
|
}
|
|
for field in record["fields"]
|
|
},
|
|
}
|
|
for record in bundle.records_data["records"]
|
|
},
|
|
"enums": {
|
|
enum["name"]: {str(value["value"]): value["name"] for value in enum["values"]}
|
|
for enum in bundle.records_data["enums"]
|
|
},
|
|
"unions": {
|
|
union["name"]: {
|
|
"record": union["record"],
|
|
"discriminator": union["discriminator"],
|
|
"cases": {
|
|
case["value"]: {
|
|
key: sorted(case[key]) for key in ("required", "allowed", "nonempty", "empty")
|
|
}
|
|
for case in union["cases"]
|
|
},
|
|
}
|
|
for union in bundle.records_data["unions"]
|
|
},
|
|
"record_kinds": {
|
|
str(row["kind"]): {"name": row["name"], "schema": row["schema"]}
|
|
for row in bundle.records_data["record_kinds"]
|
|
},
|
|
"digests": {
|
|
row["name"]: {
|
|
key: row[key]
|
|
for key in ("algorithm", "separator_utf8_hex", "input", "entry_encoding", "formula", "result")
|
|
if key in row
|
|
}
|
|
for row in bundle.digests_data["domains"]
|
|
},
|
|
"artifact_digest_mapping": {
|
|
row["artifact_kind"]: {
|
|
key: row[key] for key in ("domain", "record_kind", "record_schema")
|
|
}
|
|
for row in bundle.records_data["artifact_digest_mapping"]
|
|
},
|
|
"wrappers": {
|
|
row["name"]: {key: row[key] for key in ("body_record", "magic_hex", "identity")}
|
|
for row in bundle.records_data["wrappers"]
|
|
},
|
|
}
|
|
|
|
|
|
def expect_protocol_error(operation: Any, expected: dict[str, str], case_id: str) -> None:
|
|
try:
|
|
operation()
|
|
except tool.ProtocolError as exc:
|
|
if {"code": exc.code, "path": exc.path} != expected:
|
|
fail(f"{case_id}: got {exc.code} {exc.path}, wanted {expected}")
|
|
else:
|
|
fail(f"{case_id}: accepted")
|
|
|
|
|
|
def check_record_vectors(bundle: tool.SchemaBundle) -> None:
|
|
vectors = load_vectors(
|
|
"records.json",
|
|
"wwar-v1-record-vectors",
|
|
{
|
|
"format",
|
|
"version",
|
|
"archived_assignment_sha256",
|
|
"assignment_sha256",
|
|
"coverage",
|
|
"cases",
|
|
"union_cases",
|
|
"wrapper_cases",
|
|
"invalid_cases",
|
|
},
|
|
)
|
|
if vectors["archived_assignment_sha256"] != ARCHIVED_ASSIGNMENT_SHA256:
|
|
fail("records: archived assignment fingerprint changed")
|
|
catalog = assignment_catalog(bundle)
|
|
digest = hashlib.sha256(
|
|
json.dumps(catalog, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
).hexdigest()
|
|
if digest != vectors["assignment_sha256"]:
|
|
fail("records: assignment catalog fingerprint changed")
|
|
expected_coverage = {
|
|
"scalar_types": len(bundle.scalars),
|
|
"path_classes": len(bundle.paths),
|
|
"records": len(bundle.records),
|
|
"fields": sum(len(row["fields"]) for row in bundle.records.values()),
|
|
"encoded_defaults": sum(
|
|
field["encoded_default"] is not None
|
|
for row in bundle.records.values()
|
|
for field in row["fields"]
|
|
),
|
|
"enums": len(bundle.enums),
|
|
"enum_values": sum(len(row["values"]) for row in bundle.enums.values()),
|
|
"unions": len(bundle.records_data["unions"]),
|
|
"union_arms": sum(len(row["cases"]) for row in bundle.records_data["unions"]),
|
|
"record_kinds": len(bundle.kinds),
|
|
"artifact_mappings": len(bundle.records_data["artifact_digest_mapping"]),
|
|
"wrappers": len(bundle.wrappers),
|
|
}
|
|
exact_keys(vectors["coverage"], set(expected_coverage), "records.coverage")
|
|
if vectors["coverage"] != expected_coverage:
|
|
fail(f"records: coverage mismatch {vectors['coverage']} != {expected_coverage}")
|
|
for enum in bundle.enums.values():
|
|
for member in enum["values"]:
|
|
type_spec = {"kind": "enum", "name": enum["name"]}
|
|
ast_value = bundle._semantic_ast(type_spec, member["name"], "$.enum")
|
|
if ast_value != {"uint": str(member["value"])}:
|
|
fail(f"enum {enum['name']}.{member['name']}: typed numeric mismatch")
|
|
encoded = tool.encode_envelope(ast_value, bundle.wire)
|
|
decoded_ast = tool.decode_envelope(encoded, bundle.wire)
|
|
if bundle._decode_semantic_ast(type_spec, decoded_ast, "$.enum") != member["name"]:
|
|
fail(f"enum {enum['name']}.{member['name']}: typed round-trip mismatch")
|
|
case_ids: set[str] = set()
|
|
for case in vectors["cases"]:
|
|
allowed = {"id", "record", "value", "wwar_hex", "decoded", "record_id"}
|
|
required = {"id", "record", "value", "wwar_hex"}
|
|
if not isinstance(case, dict) or not required <= set(case) <= allowed:
|
|
fail("records: malformed ordinary case")
|
|
register_case_id(case, case_ids, "records ordinary case")
|
|
encoded = bundle.encode_record(case["record"], case["value"])
|
|
if encoded.hex() != case["wwar_hex"]:
|
|
fail(f"record {case['id']}: golden mismatch")
|
|
decoded = bundle.decode_record(case["record"], encoded)
|
|
if "decoded" in case and decoded != case["decoded"]:
|
|
fail(f"record {case['id']}: decoded mismatch")
|
|
if "record_id" in case:
|
|
kind = bundle.records[case["record"]]["top_level_kind"]
|
|
if bundle.record_id(case["record"], kind, 1, case["value"]).hex() != case["record_id"]:
|
|
fail(f"record {case['id']}: identity mismatch")
|
|
actual_union_cases: set[tuple[str, str]] = set()
|
|
for case in vectors["union_cases"]:
|
|
exact_keys(case, {"id", "record", "value", "wwar_hex"}, "records.union case")
|
|
register_case_id(case, case_ids, "records union case")
|
|
encoded = bundle.encode_record(case["record"], case["value"])
|
|
if encoded.hex() != case["wwar_hex"] or bundle.decode_record(case["record"], encoded) != case["value"]:
|
|
fail(f"union {case['id']}: golden/round-trip mismatch")
|
|
actual_union_cases.add((case["record"], case["value"][bundle.unions[case["record"]]["discriminator"]]))
|
|
expected_union_cases = {
|
|
(union["record"], case["value"])
|
|
for union in bundle.records_data["unions"]
|
|
for case in union["cases"]
|
|
}
|
|
if actual_union_cases != expected_union_cases or len(actual_union_cases) != len(vectors["union_cases"]):
|
|
fail("union vectors do not cover every arm exactly once")
|
|
for case in vectors["wrapper_cases"]:
|
|
exact_keys(case, {"id", "wrapper", "value", "bytes_hex"}, "records.wrapper case")
|
|
register_case_id(case, case_ids, "records wrapper case")
|
|
encoded = bundle.encode_wrapper(case["wrapper"], case["value"])
|
|
if encoded.hex() != case["bytes_hex"]:
|
|
fail(f"wrapper {case['id']}: golden mismatch")
|
|
if bundle.decode_wrapper(case["wrapper"], encoded) != bundle._materialize_record(
|
|
bundle.wrappers[case["wrapper"]]["body_record"], case["value"], "$"
|
|
):
|
|
fail(f"wrapper {case['id']}: round-trip mismatch")
|
|
for case in vectors["invalid_cases"]:
|
|
register_case_id(case, case_ids, "records invalid case")
|
|
if case["operation"] == "record-id":
|
|
exact_keys(
|
|
case,
|
|
{"id", "operation", "record", "kind", "schema", "value", "error"},
|
|
"records.invalid record-id",
|
|
)
|
|
operation = lambda case=case: bundle.record_id(
|
|
case["record"], case["kind"], case["schema"], case["value"]
|
|
)
|
|
elif case["operation"] == "decode-record":
|
|
exact_keys(
|
|
case,
|
|
{"id", "operation", "record", "input_hex", "error"},
|
|
"records.invalid decode-record",
|
|
)
|
|
operation = lambda case=case: bundle.decode_record(
|
|
case["record"], bytes.fromhex(case["input_hex"])
|
|
)
|
|
elif case["operation"] == "decode-wrapper":
|
|
exact_keys(
|
|
case,
|
|
{"id", "operation", "wrapper", "input_hex", "error"},
|
|
"records.invalid decode-wrapper",
|
|
)
|
|
operation = lambda case=case: bundle.decode_wrapper(
|
|
case["wrapper"], bytes.fromhex(case["input_hex"])
|
|
)
|
|
else:
|
|
fail(f"record invalid case {case['id']}: unknown operation")
|
|
exact_keys(case["error"], {"code", "path"}, f"records.invalid {case['id']}.error")
|
|
expect_protocol_error(operation, case["error"], case["id"])
|
|
|
|
|
|
def check_assignment_probes(bundle: tool.SchemaBundle) -> None:
|
|
covered_records: set[str] = set()
|
|
covered_fields: set[tuple[str, int]] = set()
|
|
covered_defaults: set[tuple[str, int]] = set()
|
|
covered_kinds: set[int] = set()
|
|
for name, record in bundle.records.items():
|
|
full = bundle.conformance_sample(name)
|
|
encoded = bundle.encode_record(name, full)
|
|
raw = tool.decode_envelope(encoded, bundle.wire)
|
|
tags = [tag for tag, _ in raw["record"]]
|
|
expected_tags = [field["tag"] for field in record["fields"]]
|
|
if tags != expected_tags:
|
|
fail(f"assignment probe {name}: encoded tags differ")
|
|
decoded = bundle.decode_record(name, encoded)
|
|
if bundle.encode_record(name, decoded) != encoded:
|
|
fail(f"assignment probe {name}: typed codec is not byte-idempotent")
|
|
covered_records.add(name)
|
|
covered_fields.update((name, tag) for tag in tags)
|
|
|
|
minimal = {
|
|
field["name"]: copy.deepcopy(full[field["name"]])
|
|
for field in record["fields"]
|
|
if field["encoded_default"] is None
|
|
}
|
|
union = bundle.unions.get(name)
|
|
if union:
|
|
selected = full[union["discriminator"]]
|
|
case = next(row for row in union["cases"] if row["value"] == selected)
|
|
for field_name in case["nonempty"]:
|
|
minimal[field_name] = copy.deepcopy(full[field_name])
|
|
default_encoded = bundle.encode_record(name, minimal)
|
|
default_raw = tool.decode_envelope(default_encoded, bundle.wire)
|
|
if [tag for tag, _ in default_raw["record"]] != expected_tags:
|
|
fail(f"assignment probe {name}: defaulted tags were omitted")
|
|
default_decoded = bundle.decode_record(name, default_encoded)
|
|
for field in record["fields"]:
|
|
if field["encoded_default"] is None:
|
|
continue
|
|
if field["name"] not in minimal and default_decoded[field["name"]] != field["encoded_default"]:
|
|
fail(f"assignment probe {name}.{field['name']}: encoded default differs")
|
|
|
|
candidates = union["cases"] if union else [None]
|
|
for candidate in candidates:
|
|
case_name = candidate["value"] if candidate else None
|
|
omitted = bundle.conformance_sample(name, case_name)
|
|
omitted.pop(field["name"])
|
|
try:
|
|
omitted_encoded = bundle.encode_record(name, omitted)
|
|
omitted_decoded = bundle.decode_record(name, omitted_encoded)
|
|
except (tool.ProtocolError, tool.SchemaError):
|
|
continue
|
|
omitted_raw = tool.decode_envelope(omitted_encoded, bundle.wire)
|
|
if field["tag"] not in {tag for tag, _ in omitted_raw["record"]}:
|
|
fail(f"assignment probe {name}.{field['name']}: default tag was omitted")
|
|
if omitted_decoded[field["name"]] != field["encoded_default"]:
|
|
fail(f"assignment probe {name}.{field['name']}: omitted default differs")
|
|
covered_defaults.add((name, field["tag"]))
|
|
break
|
|
else:
|
|
fail(f"assignment probe {name}.{field['name']}: no valid omission probe")
|
|
|
|
kind = record["top_level_kind"]
|
|
if kind is not None:
|
|
if len(bundle.record_id(name, kind, 1, full)) != 32:
|
|
fail(f"assignment probe {name}: record identity length differs")
|
|
covered_kinds.add(kind)
|
|
|
|
expected_fields = {
|
|
(name, field["tag"])
|
|
for name, record in bundle.records.items()
|
|
for field in record["fields"]
|
|
}
|
|
expected_defaults = {
|
|
(name, field["tag"])
|
|
for name, record in bundle.records.items()
|
|
for field in record["fields"]
|
|
if field["encoded_default"] is not None
|
|
}
|
|
if covered_records != set(bundle.records):
|
|
fail("assignment probes do not cover every record")
|
|
if covered_fields != expected_fields:
|
|
fail("assignment probes do not cover every field tag")
|
|
if covered_defaults != expected_defaults:
|
|
fail("assignment probes do not cover every encoded default")
|
|
if covered_kinds != set(bundle.kinds):
|
|
fail("assignment probes do not cover every record kind")
|
|
|
|
covered_wrappers = set()
|
|
for name, wrapper in bundle.wrappers.items():
|
|
value = bundle.conformance_sample(wrapper["body_record"])
|
|
encoded = bundle.encode_wrapper(name, value)
|
|
decoded = bundle.decode_wrapper(name, encoded)
|
|
if bundle.encode_wrapper(name, decoded) != encoded:
|
|
fail(f"assignment probe wrapper {name}: codec is not byte-idempotent")
|
|
covered_wrappers.add(name)
|
|
if covered_wrappers != set(bundle.wrappers):
|
|
fail("assignment probes do not cover every wrapper")
|
|
|
|
order_probe = None
|
|
for name, record in bundle.records.items():
|
|
for field in record["fields"]:
|
|
if field["order"] == "sorted":
|
|
order_probe = (name, field)
|
|
break
|
|
if order_probe:
|
|
break
|
|
if order_probe is None:
|
|
fail("assignment probes found no sorted field")
|
|
record_name, field = order_probe
|
|
value = bundle.conformance_sample(record_name)
|
|
item = bundle._sample_type(field["type"]["item"], (record_name,))
|
|
value[field["name"]] = [item, copy.deepcopy(item)]
|
|
expect_protocol_error(
|
|
lambda: bundle.encode_record(record_name, value),
|
|
{"code": "CONSTRAINT_VIOLATION", "path": f"$.{field['name']}[1]"},
|
|
"derived-duplicate-sort-key",
|
|
)
|
|
|
|
|
|
def oracle_source_tree(entries: list[dict[str, Any]]) -> bytes:
|
|
out = bytearray()
|
|
for index, entry in enumerate(entries):
|
|
exact_keys(
|
|
entry,
|
|
{"path", "type", "executable", "content"},
|
|
f"digests.source-tree entry {index}",
|
|
)
|
|
path = entry["path"].encode("utf-8")
|
|
out += struct.pack(">Q", len(path)) + path
|
|
if entry["type"] == "dir":
|
|
out += b"\x01\x00" + struct.pack(">Q", 0)
|
|
else:
|
|
content = bytes.fromhex(entry["content"])
|
|
out += b"\x02" + bytes([int(entry["executable"])]) + struct.pack(">Q", len(content)) + content
|
|
return bytes(out)
|
|
|
|
|
|
def check_digest_vectors(bundle: tool.SchemaBundle) -> None:
|
|
vectors = load_vectors(
|
|
"digests.json", "wwar-v1-digest-vectors", {"format", "version", "cases"}
|
|
)
|
|
if {case["domain"] for case in vectors["cases"]} != set(bundle.domains) or len(vectors["cases"]) != len(bundle.domains):
|
|
fail("digests: every domain must have exactly one vector")
|
|
action_domains = {
|
|
domain["name"]
|
|
for domain in bundle.domains.values()
|
|
if tuple(domain["formula"]) == tool.FORMULA_WWAR
|
|
and set(domain["result"]) == {"kind", "scalar"}
|
|
}
|
|
if len(action_domains) != 1:
|
|
fail("digests: expected one scalar-result WWAR action-key formula")
|
|
action_seen: set[str] = set()
|
|
case_ids: set[str] = set()
|
|
for case in vectors["cases"]:
|
|
if not isinstance(case, dict) or "domain" not in case or case["domain"] not in bundle.domains:
|
|
fail("digests: malformed or unknown domain case")
|
|
register_case_id(case, case_ids, "digests case")
|
|
domain = bundle.domains[case["domain"]]
|
|
formula = tuple(domain["formula"])
|
|
separator = bytes.fromhex(domain["separator_utf8_hex"])
|
|
if formula == tool.FORMULA_BYTES:
|
|
exact_keys(case, {"id", "domain", "input_hex", "digest"}, "digests.bytes case")
|
|
value: Any = bytes.fromhex(case["input_hex"])
|
|
expected = hashlib.sha256(separator + struct.pack(">Q", len(value)) + value).digest()
|
|
actual = bundle.digest(case["domain"], value)
|
|
elif formula == tool.FORMULA_WWAR:
|
|
exact_keys(
|
|
case,
|
|
{"id", "domain", "record", "value", "wwar_hex", "digest"},
|
|
"digests.WWAR case",
|
|
)
|
|
encoded = bundle.encode_record(case["record"], case["value"])
|
|
if encoded.hex() != case["wwar_hex"]:
|
|
fail(f"digest {case['id']}: WWAR golden mismatch")
|
|
expected = hashlib.sha256(separator + struct.pack(">Q", len(encoded)) + encoded).digest()
|
|
actual = bundle.digest(case["domain"], case["value"], record_name=case["record"])
|
|
if case["domain"] in action_domains:
|
|
action_seen.add(case["domain"])
|
|
elif formula == tool.FORMULA_RECORD:
|
|
exact_keys(
|
|
case,
|
|
{"id", "domain", "record", "kind", "schema", "value", "wwar_hex", "digest"},
|
|
"digests.record case",
|
|
)
|
|
encoded = bundle.encode_record(case["record"], case["value"])
|
|
if encoded.hex() != case["wwar_hex"]:
|
|
fail(f"digest {case['id']}: WWAR golden mismatch")
|
|
expected = hashlib.sha256(
|
|
separator
|
|
+ struct.pack(">I", case["kind"])
|
|
+ struct.pack(">I", case["schema"])
|
|
+ struct.pack(">Q", len(encoded))
|
|
+ encoded
|
|
).digest()
|
|
actual = bundle.digest(
|
|
case["domain"],
|
|
case["value"],
|
|
record_name=case["record"],
|
|
kind=case["kind"],
|
|
schema=case["schema"],
|
|
)
|
|
elif formula == tool.FORMULA_SOURCE_TREE:
|
|
exact_keys(case, {"id", "domain", "entries", "digest"}, "digests.source-tree case")
|
|
value = case["entries"]
|
|
expected = hashlib.sha256(separator + oracle_source_tree(value)).digest()
|
|
actual = bundle.digest(case["domain"], value)
|
|
else:
|
|
fail(f"digest {case['id']}: unknown formula")
|
|
if actual != expected or actual.hex() != case["digest"]:
|
|
fail(f"digest {case['id']}: tool/oracle/golden mismatch")
|
|
if action_seen != action_domains:
|
|
fail("digests: missing action-key vector")
|
|
probe = b"domain separation probe"
|
|
separated = []
|
|
for domain in bundle.domains.values():
|
|
separator = bytes.fromhex(domain["separator_utf8_hex"])
|
|
formula = tuple(domain["formula"])
|
|
if formula in {tool.FORMULA_BYTES, tool.FORMULA_WWAR}:
|
|
preimage = separator + struct.pack(">Q", len(probe)) + probe
|
|
elif formula == tool.FORMULA_RECORD:
|
|
preimage = separator + struct.pack(">I", 1) + struct.pack(">I", 1) + struct.pack(">Q", len(probe)) + probe
|
|
else:
|
|
preimage = separator + probe
|
|
separated.append(hashlib.sha256(preimage).digest())
|
|
if len(separated) != len(set(separated)):
|
|
fail("digests: domain separation collision in fixed probe")
|
|
|
|
|
|
def write_schema_set(directory: pathlib.Path, wire: Any, records: Any, digests: Any) -> None:
|
|
directory.mkdir(parents=True)
|
|
for name, value in (("wire.json", wire), ("records.json", records), ("digests.json", digests)):
|
|
(directory / name).write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
tool.write_schema_manifest(directory)
|
|
|
|
|
|
def expect_schema_error(operation: Any, label: str) -> None:
|
|
try:
|
|
operation()
|
|
except tool.SchemaError:
|
|
return
|
|
fail(f"schema mutation accepted: {label}")
|
|
|
|
|
|
def expect_assertion(operation: Any, label: str) -> None:
|
|
try:
|
|
operation()
|
|
except AssertionError:
|
|
return
|
|
fail(f"checker mutation accepted: {label}")
|
|
|
|
|
|
def synthetic_records(
|
|
record_name: str,
|
|
enum_name: str,
|
|
field_name: str,
|
|
tag: int = 7,
|
|
default: str = "red",
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"format": "wwar-v1-record-schema",
|
|
"version": 1,
|
|
"scalar_types": [],
|
|
"path_classes": [],
|
|
"enums": [{"name": enum_name, "values": [{"name": "red", "value": 1}, {"name": "blue", "value": 2}]}],
|
|
"record_kinds": [{"kind": 1, "name": record_name, "schema": 1}],
|
|
"records": [
|
|
{
|
|
"name": record_name,
|
|
"top_level_kind": 1,
|
|
"fields": [
|
|
{
|
|
"tag": tag,
|
|
"name": field_name,
|
|
"type": {"kind": "enum", "name": enum_name},
|
|
"cardinality": "1",
|
|
"encoded_default": default,
|
|
"order": "none",
|
|
"order_by": [],
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"unions": [],
|
|
"artifact_digest_mapping": [],
|
|
"wrappers": [],
|
|
}
|
|
|
|
|
|
def synthetic_digests(record_name: str) -> dict[str, Any]:
|
|
return {
|
|
"format": "wwar-v1-digest-schema",
|
|
"version": 1,
|
|
"domains": [
|
|
{
|
|
"name": "raw",
|
|
"algorithm": "sha256",
|
|
"separator_utf8_hex": "53594e2d52415700",
|
|
"input": {"kind": "bytes"},
|
|
"formula": list(tool.FORMULA_BYTES),
|
|
"result": {"kind": "bytes", "length": 32},
|
|
},
|
|
{
|
|
"name": "shape",
|
|
"algorithm": "sha256",
|
|
"separator_utf8_hex": "53594e2d534841504500",
|
|
"input": {"record": record_name, "encoding": "WWAR"},
|
|
"formula": list(tool.FORMULA_WWAR),
|
|
"result": {"kind": "bytes", "length": 32},
|
|
},
|
|
{
|
|
"name": "identity",
|
|
"algorithm": "sha256",
|
|
"separator_utf8_hex": "53594e2d4944454e5449545900",
|
|
"input": {"record": "top-level record", "encoding": "WWAR"},
|
|
"formula": list(tool.FORMULA_RECORD),
|
|
"result": {"kind": "bytes", "length": 32},
|
|
},
|
|
],
|
|
}
|
|
|
|
|
|
def replace_exact_strings(value: Any, mapping: dict[str, str]) -> Any:
|
|
if isinstance(value, str):
|
|
return mapping.get(value, value)
|
|
if isinstance(value, list):
|
|
return [replace_exact_strings(item, mapping) for item in value]
|
|
if isinstance(value, dict):
|
|
return {
|
|
mapping.get(key, key): replace_exact_strings(child, mapping)
|
|
for key, child in value.items()
|
|
}
|
|
return value
|
|
|
|
|
|
def check_genericity_and_mutations(bundle: tool.SchemaBundle) -> None:
|
|
with tempfile.TemporaryDirectory(prefix="wwar-schema-mutations-") as raw_tmp:
|
|
base = pathlib.Path(raw_tmp)
|
|
wire = copy.deepcopy(bundle.wire)
|
|
records = copy.deepcopy(bundle.records_data)
|
|
digests = copy.deepcopy(bundle.digests_data)
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["predicate_programs"] = []
|
|
directory = base / "unknown-root"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "semantic root")
|
|
|
|
mutation = copy.deepcopy(wire)
|
|
mutation["error_precedence"] = {
|
|
"op": "context-lookup",
|
|
"relation": "provider-table",
|
|
}
|
|
directory = base / "wire-policy-object"
|
|
write_schema_set(directory, mutation, records, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "wire policy object")
|
|
|
|
mutation = copy.deepcopy(wire)
|
|
mutation["root"] = "X" * 1_000_000
|
|
directory = base / "giant-wire-root"
|
|
write_schema_set(directory, mutation, records, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "giant wire root")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["records"][0]["fields"][0]["constraint"] = {"op": "context-lookup"}
|
|
directory = base / "field-expression"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "field expression")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["wrappers"][0]["identity"] = {
|
|
"op": "graph-dfs",
|
|
"policy": "bootstrap",
|
|
}
|
|
directory = base / "wrapper-policy-object"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "wrapper policy object")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["path_classes"][0]["rules"] = [
|
|
{"op": "call", "operation": "provider-expand"}
|
|
]
|
|
directory = base / "path-policy-object"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "path policy object")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["path_classes"][0]["rules"] = ["no-backslash"]
|
|
directory = base / "redundant-path-rules"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "redundant path rules")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
scalar_field = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "string" and field["cardinality"] == "1"
|
|
)
|
|
scalar_field["cardinality"] = "*"
|
|
directory = base / "scalar-list-cardinality"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "scalar list cardinality")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
scalar_field = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "string" and field["order"] == "none"
|
|
)
|
|
scalar_field["order"] = "ordered"
|
|
directory = base / "scalar-ordered"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "scalar ordered field")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
scalar_field = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "string"
|
|
and field["cardinality"] == "1"
|
|
and field["encoded_default"] is None
|
|
and not field["type"].get("scalar")
|
|
)
|
|
scalar_field["encoded_default"] = "x" * 100_000
|
|
directory = base / "giant-string-default"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "giant string default")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
list_field = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "list"
|
|
and field["type"]["item"]["kind"] == "string"
|
|
)
|
|
list_field["encoded_default"] = ["x"] * 2000
|
|
directory = base / "giant-list-default"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "giant list default")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
map_field = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "map"
|
|
)
|
|
map_field["encoded_default"] = {f"k{index}": "" for index in range(1100)}
|
|
directory = base / "giant-map-default"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "giant map default")
|
|
|
|
mutation = copy.deepcopy(digests)
|
|
mutation["domains"][0]["formula"] = {"op": "call"}
|
|
directory = base / "digest-expression"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "digest expression")
|
|
|
|
mutation = copy.deepcopy(digests)
|
|
mutation["domains"][0]["result"] = {"scalar": "context-lookup"}
|
|
directory = base / "digest-result-policy"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "digest result policy")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
incomplete = next(
|
|
field
|
|
for record in mutation["records"]
|
|
for field in record["fields"]
|
|
if field["type"]["kind"] == "record"
|
|
and isinstance(field["encoded_default"], dict)
|
|
and field["encoded_default"]
|
|
)
|
|
incomplete["encoded_default"].pop(next(iter(incomplete["encoded_default"])))
|
|
directory = base / "incomplete-record-default"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "incomplete record default")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["records"][0]["fields"][1]["tag"] = mutation["records"][0]["fields"][0]["tag"]
|
|
directory = base / "duplicate-tag"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "duplicate tag")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
mutation["enums"][0]["values"][1]["value"] = mutation["enums"][0]["values"][0]["value"]
|
|
directory = base / "duplicate-enum"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "duplicate enum value")
|
|
|
|
mutation = copy.deepcopy(digests)
|
|
mutation["domains"][1]["separator_utf8_hex"] = mutation["domains"][0]["separator_utf8_hex"]
|
|
directory = base / "duplicate-domain"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "duplicate digest separator")
|
|
|
|
mutation = copy.deepcopy(digests)
|
|
mutation["domains"][0]["separator_utf8_hex"] = "ff00"
|
|
directory = base / "invalid-utf8-separator"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "invalid UTF-8 separator")
|
|
|
|
mutation = copy.deepcopy(digests)
|
|
mutation["domains"][0]["separator_utf8_hex"] = "78007900"
|
|
directory = base / "interior-nul-separator"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "interior NUL separator")
|
|
|
|
wrong_result_record = next(name for name in bundle.records if name not in bundle.unions)
|
|
for formula, label in (
|
|
(tool.FORMULA_BYTES, "blob"),
|
|
(tool.FORMULA_RECORD, "record"),
|
|
):
|
|
mutation = copy.deepcopy(digests)
|
|
result_domain = next(
|
|
domain
|
|
for domain in mutation["domains"]
|
|
if tuple(domain["formula"]) == formula and "record" in domain["result"]
|
|
)
|
|
result_domain["result"]["record"] = wrong_result_record
|
|
directory = base / f"wrong-{label}-result-record"
|
|
write_schema_set(directory, wire, records, mutation)
|
|
expect_schema_error(
|
|
lambda directory=directory: tool.SchemaBundle.from_dir(directory),
|
|
f"wrong {label} result record",
|
|
)
|
|
|
|
mutation = copy.deepcopy(records)
|
|
union_case = mutation["unions"][0]["cases"][0]
|
|
union_case["allowed"].append(union_case["empty"][0])
|
|
directory = base / "union-allowed-empty-overlap"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "union allowed/empty overlap")
|
|
|
|
mutation = copy.deepcopy(records)
|
|
union_case = next(
|
|
case
|
|
for union in mutation["unions"]
|
|
for case in union["cases"]
|
|
if case["nonempty"]
|
|
)
|
|
union_case["required"].remove(union_case["nonempty"][0])
|
|
directory = base / "union-nonempty-not-required"
|
|
write_schema_set(directory, wire, mutation, digests)
|
|
expect_schema_error(lambda: tool.SchemaBundle.from_dir(directory), "union nonempty not required")
|
|
|
|
duplicate = base / "duplicate.json"
|
|
duplicate.write_text('{"format":"x","format":"y"}\n', encoding="utf-8")
|
|
expect_schema_error(lambda: tool.load_json(duplicate), "duplicate JSON key")
|
|
|
|
first_dir = base / "alpha-a"
|
|
second_dir = base / "alpha-b"
|
|
write_schema_set(first_dir, wire, synthetic_records("Box", "Shade", "hue"), synthetic_digests("Box"))
|
|
write_schema_set(second_dir, wire, synthetic_records("Crate", "Tone", "tint"), synthetic_digests("Crate"))
|
|
first_table = first_dir / "table.json"
|
|
second_table = second_dir / "table.json"
|
|
tool.generate_table(first_dir, first_table)
|
|
tool.generate_table(second_dir, second_table)
|
|
first = tool.load_json(first_table)
|
|
second = tool.load_json(second_table)
|
|
first.pop("schema_digest")
|
|
first.pop("schema_files")
|
|
second.pop("schema_digest")
|
|
second.pop("schema_files")
|
|
second = replace_exact_strings(second, {"Crate": "Box", "Tone": "Shade", "tint": "hue"})
|
|
if first != second:
|
|
fail("alpha-renamed synthetic schemas changed generator behavior")
|
|
|
|
third_dir = base / "different-tag-default"
|
|
write_schema_set(
|
|
third_dir,
|
|
wire,
|
|
synthetic_records("Vessel", "Hue", "shade", tag=23, default="blue"),
|
|
synthetic_digests("Vessel"),
|
|
)
|
|
third = tool.SchemaBundle.from_dir(third_dir)
|
|
encoded = third.encode_record("Vessel", {})
|
|
raw = tool.decode_envelope(encoded, third.wire)
|
|
if raw["record"][0][0] != 23 or third.decode_record("Vessel", encoded) != {"shade": "blue"}:
|
|
fail("synthetic changed tag/default did not drive the generic codec")
|
|
|
|
expect_assertion(
|
|
lambda: exact_keys(
|
|
{"format": "x", "version": 1, "cases": [], "provider_policy": {}},
|
|
{"format", "version", "cases"},
|
|
"synthetic vector",
|
|
),
|
|
"vector policy root",
|
|
)
|
|
|
|
duplicate_ids: set[str] = set()
|
|
register_case_id({"id": "same"}, duplicate_ids, "synthetic vector")
|
|
expect_assertion(
|
|
lambda: register_case_id({"id": "same"}, duplicate_ids, "synthetic vector"),
|
|
"duplicate vector id",
|
|
)
|
|
|
|
stale = tool.load_json(GENERATED)
|
|
first_record = next(iter(stale["records"].values()))
|
|
if not first_record[1]:
|
|
fail("cannot construct stale-output mutation")
|
|
first_record[1][0][0] += 100
|
|
stale_path = base / "stale.json"
|
|
stale_path.write_bytes(tool.canonical_json(stale))
|
|
run_tool("check-output", "--schema-dir", str(SCHEMA), "--output", str(stale_path), expect=2)
|
|
|
|
|
|
def check_tool_independence(bundle: tool.SchemaBundle) -> None:
|
|
source = (ROOT / "tool.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source, filename="tool.py")
|
|
allowed_imports = {
|
|
"argparse",
|
|
"copy",
|
|
"hashlib",
|
|
"json",
|
|
"pathlib",
|
|
"struct",
|
|
"sys",
|
|
"unicodedata",
|
|
"dataclasses",
|
|
"typing",
|
|
"__future__",
|
|
}
|
|
forbidden_calls = {"eval", "exec", "compile", "getattr", "setattr", "delattr", "id"}
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Import):
|
|
for alias in node.names:
|
|
if alias.name.split(".")[0] not in allowed_imports:
|
|
fail(f"tool.py imports disallowed module {alias.name}")
|
|
elif isinstance(node, ast.ImportFrom):
|
|
if (node.module or "").split(".")[0] not in allowed_imports:
|
|
fail(f"tool.py imports from disallowed module {node.module}")
|
|
elif isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in forbidden_calls:
|
|
fail(f"tool.py uses forbidden dynamic call {node.func.id}")
|
|
action_spellings = {
|
|
value["name"]
|
|
for enum in bundle.enums.values()
|
|
if any("." in value["name"] for value in enum["values"])
|
|
for value in enum["values"]
|
|
}
|
|
semantic_identifiers = (
|
|
set(bundle.records)
|
|
| set(bundle.enums)
|
|
| {row["name"] for row in bundle.records_data["unions"]}
|
|
| set(bundle.wrappers)
|
|
| (set(bundle.domains) - {"blob", "record"})
|
|
| action_spellings
|
|
| {
|
|
value["name"]
|
|
for enum in bundle.enums.values()
|
|
for value in enum["values"]
|
|
if "." in value["name"]
|
|
}
|
|
)
|
|
for filename in ("tool.py", "check.py"):
|
|
parsed = ast.parse((ROOT / filename).read_text(encoding="utf-8"), filename=filename)
|
|
for node in ast.walk(parsed):
|
|
if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
|
|
if node.func.id in forbidden_calls:
|
|
fail(f"{filename} uses forbidden dynamic/object-identity call {node.func.id}")
|
|
if (
|
|
isinstance(node, ast.Constant)
|
|
and isinstance(node.value, str)
|
|
and node.value in semantic_identifiers
|
|
):
|
|
fail(f"{filename} hard-codes schema semantic identifier {node.value}")
|
|
|
|
|
|
def check_deterministic_generation(bundle: tool.SchemaBundle) -> None:
|
|
checked = GENERATED.read_bytes()
|
|
expected = tool.expected_table(SCHEMA)
|
|
if checked != expected:
|
|
fail("checked-in generated output is stale")
|
|
generated = tool.load_json(GENERATED)
|
|
exact_keys(
|
|
generated,
|
|
{
|
|
"format",
|
|
"version",
|
|
"schema_digest",
|
|
"schema_files",
|
|
"wire",
|
|
"scalars",
|
|
"paths",
|
|
"enums",
|
|
"records",
|
|
"unions",
|
|
"kinds",
|
|
"artifacts",
|
|
"wrappers",
|
|
"digests",
|
|
},
|
|
"generated table",
|
|
)
|
|
if generated != bundle.generated_table():
|
|
fail("generated table is not the exact compiled codec projection")
|
|
with tempfile.TemporaryDirectory(prefix="wwar-generate-a-") as first_tmp, tempfile.TemporaryDirectory(
|
|
prefix="wwar-generate-b-"
|
|
) as second_tmp:
|
|
first = pathlib.Path(first_tmp) / "generated" / GENERATED.name
|
|
second = pathlib.Path(second_tmp) / "generated" / GENERATED.name
|
|
run_tool("emit-tables", "--schema-dir", str(SCHEMA), "--output", str(first))
|
|
run_tool("emit-tables", "--schema-dir", str(SCHEMA), "--output", str(second))
|
|
run_tool("check-output", "--schema-dir", str(SCHEMA), "--output", str(first))
|
|
run_tool("check-output", "--schema-dir", str(SCHEMA), "--output", str(second))
|
|
if first.read_bytes() != second.read_bytes() or first.read_bytes() != checked:
|
|
fail("two clean generations and checked-in output are not byte-identical")
|
|
|
|
|
|
def check_size_budgets() -> None:
|
|
byte_limits = {
|
|
"README.md": 64 * 1024,
|
|
"check.py": 256 * 1024,
|
|
"tool.py": 256 * 1024,
|
|
"schema/digests.json": 128 * 1024,
|
|
"schema/records.json": 1024 * 1024,
|
|
"schema/wire.json": 64 * 1024,
|
|
"schema/wwar-v1.sha256": 1024,
|
|
"generated/wwar_v1_tables.json": 512 * 1024,
|
|
"vectors/digests.json": 512 * 1024,
|
|
"vectors/records.json": 1024 * 1024,
|
|
"vectors/wire-invalid.json": 256 * 1024,
|
|
"vectors/wire-valid.json": 256 * 1024,
|
|
}
|
|
schema_lines = 0
|
|
human_lines = 0
|
|
human_bytes = 0
|
|
total_bytes = 0
|
|
for path in sorted(EXPECTED_FILES):
|
|
full = ROOT / path
|
|
raw = full.read_bytes()
|
|
lines = len(raw.splitlines())
|
|
total_bytes += len(raw)
|
|
if len(raw) > byte_limits[path]:
|
|
fail(f"Phase 0 file exceeds byte budget: {path} ({len(raw)} bytes)")
|
|
if not path.startswith("generated/"):
|
|
human_lines += lines
|
|
human_bytes += len(raw)
|
|
if path.startswith("schema/") and path.endswith(".json"):
|
|
schema_lines += lines
|
|
if lines > 5000:
|
|
fail(f"human-authored schema file exceeds 5000 lines: {path}")
|
|
if schema_lines >= 5000 or human_lines >= 15000 or human_bytes >= 2 * 1024 * 1024:
|
|
fail(
|
|
"Phase 0 human-authored budget exceeded: "
|
|
f"schema_lines={schema_lines}, human_lines={human_lines}, human_bytes={human_bytes}"
|
|
)
|
|
if total_bytes >= 4 * 1024 * 1024:
|
|
fail(f"Phase 0 total byte budget exceeded: {total_bytes}")
|
|
|
|
|
|
def main() -> int:
|
|
check_no_bytecode()
|
|
check_files()
|
|
bundle = tool.SchemaBundle.from_dir(SCHEMA)
|
|
check_tool_independence(bundle)
|
|
check_deterministic_generation(bundle)
|
|
check_wire_vectors(bundle)
|
|
check_record_vectors(bundle)
|
|
check_assignment_probes(bundle)
|
|
check_digest_vectors(bundle)
|
|
check_genericity_and_mutations(bundle)
|
|
check_size_budgets()
|
|
check_no_bytecode()
|
|
print(
|
|
"WWAR v1 protocol OK: "
|
|
f"{len(bundle.records)} records, "
|
|
f"{sum(len(row['fields']) for row in bundle.records.values())} fields, "
|
|
f"{len(bundle.enums)} enums, {len(bundle.domains)} digest domains"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|