vdm/tests/conformance/check_contract.py

#!/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 "<root>"
        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")

    # Segment ranges are the one place an off-by-one is both easy and expensive, and the
    # schema cannot express a cross-field invariant. So it is checked here instead.
    for path, doc in iter_fixtures():
        rel = path.relative_to(REPO)
        result = (doc.get("response") or {}).get("result") if isinstance(doc.get("response"), dict) else None
        if not isinstance(result, dict):
            continue
        segments = result.get("segmentDetail")
        summary = result.get("summary")
        if not isinstance(segments, list) or not isinstance(summary, dict):
            continue

        if len(segments) != summary.get("segments"):
            fail(f"{rel}: segmentDetail has {len(segments)} entries but summary.segments is "
                 f"{summary.get('segments')}")
        for seg in segments:
            if seg["endByte"] < seg["startByte"]:
                fail(f"{rel}: segment {seg['index']} has endByte < startByte; the range is "
                     "inclusive and a segment always covers at least one byte")
            span = seg["endByte"] - seg["startByte"] + 1
            if seg["downloadedBytes"] > span:
                fail(f"{rel}: segment {seg['index']} has downloadedBytes above its range size "
                     f"({seg['downloadedBytes']} > {span}) — check for an off-by-one from "
                     "treating endByte as exclusive")
        for a, b in zip(segments, segments[1:]):
            if a["endByte"] + 1 != b["startByte"]:
                fail(f"{rel}: segments {a['index']} and {b['index']} are not contiguous: "
                     f"{a['endByte']} + 1 != {b['startByte']}")
        size = summary.get("sizeBytes")
        if segments and isinstance(size, int):
            if segments[0]["startByte"] != 0 or segments[-1]["endByte"] != size - 1:
                fail(f"{rel}: segments must cover exactly [0, {size - 1}] inclusive, got "
                     f"[{segments[0]['startByte']}, {segments[-1]['endByte']}]")

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