#!/usr/bin/env python3 """Static conformance: the contract, the fixtures and the generated code agree. This is the cheap half of the suite and the half that catches the most. It runs without a daemon, without Node and without a compiler, so it is the first thing CI does on every lane's PR. Checks 1. Every schema file parses and every $ref resolves. 2. Every method in contracts/README.md's surface has a schema, and vice versa. 3. Every method has at least one success fixture. 4. Every fixture's params and result validate against that method's schema. 5. Every error fixture uses a code the ErrorCode type defines, and one the method documents in x-errors (or a universal code). 6. Every event has a fixture, and every event fixture validates. 7. SettingKey and Settings.properties name exactly the same keys. 8. The committed generated code is up to date with the schemas. Run: python3 tests/conformance/check_contract.py """ from __future__ import annotations import json import re import subprocess import sys import tempfile from pathlib import Path REPO = Path(__file__).resolve().parent.parent.parent sys.path.insert(0, str(REPO / "contracts" / "codegen")) import jsonschema # noqa: E402 from referencing import Registry, Resource # noqa: E402 from referencing.jsonschema import DRAFT202012 # noqa: E402 from schema_ir import ID_PREFIX, Loader, load # noqa: E402 SCHEMA = REPO / "contracts" / "schema" FIXTURES = REPO / "contracts" / "fixtures" # Values that cannot be pinned in a golden file. The runners treat them as "any value of # the right shape"; here they are swapped for a concrete one so the schema can be applied. PLACEHOLDERS = { "$uuid": "e6f0a1b2-3c4d-4e5f-8a9b-0c1d2e3f4a5b", "$isoDate": "2026-09-09T10:14:52Z", "$any": "placeholder", # Bound by the runner to a task it creates, so a fixture never depends on a task id # that happens to exist in a seeded mock. "$taskId": "3f7a2b1c-5d6e-4f80-9a1b-2c3d4e5f6071", "$taskId2": "8c1d4e5f-6a7b-4c8d-9e0f-1a2b3c4d5e6f", # Long enough to satisfy a token's minLength: an opaque credential-shaped string. "$opaque": "cGxhY2Vob2xkZXItdG9rZW4tNjQtYnl0ZXMtb2YtZW50cm9weS1nb2VzLWhlcmU", } # Codes any method may return regardless of its x-errors list. UNIVERSAL = {-32700, -32600, -32601, -32602, -32603, -32001, -32002, -32003} failures: list[str] = [] def fail(msg: str) -> None: failures.append(msg) def substitute(node: object) -> object: if isinstance(node, str): return PLACEHOLDERS.get(node, node) if isinstance(node, list): return [substitute(n) for n in node] if isinstance(node, dict): return {k: substitute(v) for k, v in node.items()} return node def build_registry(loader: Loader) -> Registry: resources = [(sid, Resource(contents=doc, specification=DRAFT202012)) for sid, doc in loader.by_id.items()] return Registry().with_resources(resources) def validate(registry: Registry, ref: str, instance: object, where: str) -> None: validator = jsonschema.Draft202012Validator({"$ref": ref}, registry=registry) errors = sorted(validator.iter_errors(instance), key=lambda e: list(e.absolute_path)) for e in errors[:3]: path = "/".join(str(p) for p in e.absolute_path) or "" fail(f"{where}: {path}: {e.message}") def iter_fixtures() -> list[tuple[Path, dict]]: out = [] for p in sorted(FIXTURES.rglob("*.json")): with p.open() as fh: out.append((p, json.load(fh))) return out def main() -> int: loader = Loader() registry = build_registry(loader) contract = load() method_ids = {doc["title"]: sid for sid, doc in loader.by_id.items() if "/methods/" in sid} event_ids = {doc["title"]: sid for sid, doc in loader.by_id.items() if "/events/" in sid} # 2. the schema surface matches the documented surface, both directions readme = (REPO / "contracts" / "README.md").read_text() for name in method_ids: # The README writes runs of related methods as `download.start` | `.pause` | ... short = "." + name.split(".", 1)[1] if name not in readme and short not in readme: fail(f"method {name} has a schema but is not in contracts/README.md") namespaces = {n.split(".", 1)[0] for n in method_ids} for token in set(re.findall(r"\b([a-z]+\.[a-zA-Z][A-Za-z]*)\b", readme)): ns = token.split(".", 1)[0] if ns in namespaces and token not in method_ids: fail(f"contracts/README.md documents {token}, which has no schema") # 7. SettingKey and Settings agree keys = set(loader.by_id[ID_PREFIX + "types/SettingKey.schema.json"]["enum"]) props = set(loader.by_id[ID_PREFIX + "types/Settings.schema.json"]["properties"]) for k in sorted(keys - props): fail(f"SettingKey lists {k} but Settings.schema.json has no such property") for k in sorted(props - keys): fail(f"Settings.schema.json has property {k} but SettingKey does not list it") # 3-6. fixtures covered_methods: set[str] = set() covered_events: set[str] = set() for path, doc in iter_fixtures(): rel = path.relative_to(REPO) is_event = "notification" in doc if is_event: frame = doc["notification"] name = frame.get("method") if name not in event_ids: fail(f"{rel}: unknown event {name!r}") continue covered_events.add(name) validate(registry, event_ids[name] + "#/properties/params", substitute(frame.get("params")), f"{rel} params") continue request = doc.get("request") if not isinstance(request, dict): fail(f"{rel}: no request object") continue name = request.get("method") if name not in method_ids: # method-not-found.json deliberately names a method that does not exist. if doc.get("response", {}).get("error", {}).get("code") == -32601: continue fail(f"{rel}: unknown method {name!r}") continue sid = method_ids[name] expected_code = doc.get("response", {}).get("error", {}).get("code") if doc.get("response") else None if expected_code == -32602: # This fixture exists precisely because its params are invalid. Assert that # they really do fail the schema, or it is testing nothing. v = jsonschema.Draft202012Validator({"$ref": sid + "#/properties/params"}, registry=registry) if not list(v.iter_errors(substitute(request.get("params", {})))): fail(f"{rel}: expects -32602 but its params are schema-valid") else: validate(registry, sid + "#/properties/params", substitute(request.get("params", {})), f"{rel} request.params") response = doc.get("response") if response is None: if doc.get("kind") != "timeout": fail(f"{rel}: null response without \"kind\": \"timeout\"") continue if "result" in response: covered_methods.add(name) validate(registry, sid + "#/properties/result", substitute(response["result"]), f"{rel} response.result") elif "error" in response: validate(registry, ID_PREFIX + "envelope.schema.json#/$defs/Error", substitute(response["error"]), f"{rel} response.error") code = response["error"]["code"] declared = set(loader.by_id[sid].get("x-errors", [])) if code not in declared | UNIVERSAL: fail(f"{rel}: error {code} is not in {name}'s x-errors {sorted(declared)}") else: fail(f"{rel}: response has neither result nor error") # ids must correlate if response.get("id") != request.get("id"): fail(f"{rel}: response id does not match request id") for name in sorted(method_ids): if name not in covered_methods: fail(f"method {name} has no success fixture — a method with no fixture is not done") for name in sorted(event_ids): if name not in covered_events: fail(f"event {name} has no fixture") # 8. generated code is current for gen, out in [("gen_cpp.py", ["core/generated/velox_proto.hpp", "core/generated/velox_proto.cpp"]), ("gen_ts.py", ["extension/src/shared/protocol/types.ts", "extension/src/shared/protocol/methods.ts", "extension/src/shared/protocol/events.ts", "extension/src/shared/protocol/validate.ts", "extension/src/shared/protocol/index.ts"]), ("gen_openrpc.py", ["contracts/openrpc.json"])]: before = {f: (REPO / f).read_bytes() for f in out if (REPO / f).exists()} subprocess.run([sys.executable, str(REPO / "contracts" / "codegen" / gen)], check=True, capture_output=True) for f in out: if (REPO / f).read_bytes() != before.get(f): fail(f"{f} is stale: re-run contracts/codegen/{gen} and commit the result") print(f"contract v{contract.version}: {len(method_ids)} methods, {len(event_ids)} events, " f"{len(list(iter_fixtures()))} fixtures") if failures: print(f"\n{len(failures)} problem(s):\n") for f in failures: print(" FAIL", f) return 1 print("static conformance OK") return 0 if __name__ == "__main__": raise SystemExit(main())