"""Validate actual local HTTP responses against served and checked-in OpenAPI. Requires jsonschema. Usage: python semantic_http_contract.py /path/to/arcadia-server The supplied binary is launched on loopback with a fresh temporary store. """ import copy import json import os import re import subprocess import sys import tempfile import urllib.error import urllib.request from pathlib import Path from jsonschema import Draft202012Validator, FormatChecker ROOT = Path(__file__).resolve().parents[3] local = json.loads((ROOT / "docs/openapi.json").read_text(encoding="utf-8")) for schema in local["components"]["schemas"].values(): Draft202012Validator.check_schema(schema) def validator(spec, name): return Draft202012Validator({"$ref": "#/components/schemas/" + name, "components": spec["components"]}, format_checker=FormatChecker()) with tempfile.TemporaryDirectory(prefix="arcadia-semantic-http-") as directory: env = dict(os.environ) env.pop("ARCADIA_SERVER_TOKEN", None) process = subprocess.Popen([sys.argv[1], "--bind", "127.0.0.1:0", "--data-dir", directory], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env, creationflags=subprocess.CREATE_NO_WINDOW if os.name == "nt" else 0) try: banner = process.stderr.readline() address = re.search(r"listening on (127\.0\.0\.1:\d+)", banner) assert address, banner base = "http://" + address.group(1) def call(method, path, body=None, etag=None, expected=200): headers = {"Content-Type": "application/json"} if etag: headers["If-Match"] = etag request = urllib.request.Request(base + path, data=None if body is None else json.dumps(body).encode(), headers=headers, method=method) try: response = urllib.request.urlopen(request, timeout=15) except urllib.error.HTTPError as error: response = error with response: data = json.load(response) assert response.status == expected, (response.status, data) return data, response.headers.get("ETag") served, _ = call("GET", "/openapi.json") assert served == local, "Server binary embeds a different OpenAPI file; rebuild it" checked = 0 def validate(name, value): global checked for spec in (local, served): validator(spec, name).validate(value) checked += 1 old = json.loads((ROOT / "crates/oa-core/tests/fixtures/v4-reviewed-prose.json").read_text()) imported, etag = call("POST", "/api/v1/projects/import", old, expected=201) path = "/api/v1/projects/" + imported["resource_id"] stored, _ = call("GET", path) validate("StoredProject", stored) validate("Project", stored["model"]) trace = next(iter(stored["model"]["traces"])) oa = stored["model"]["traces"][trace]["oa"] sa = stored["model"]["traces"][trace]["sa"] query = {"traces": [trace], "references": {oa: sa}, "revision": stored["revision"]} validate("SemanticQueryRequest", query) queried, _ = call("POST", path + "/semantic-transitions/query", query) validate("SemanticQueryResult", queried) request = {"transition": {"direction": "OaToSa", "sources": [oa], "scope": "SelectedOnly", "mappings": [{"source": oa, "target": sa, "target_kind": None}], "fields": [], "resolutions": [], "destination_diagram": None}, "traces": {oa: trace}, "references": {}, "reviews": [{"source": oa, "resolutions": [ {"facet": {"Header": "Name"}, "decision": "TakeSource"}, {"facet": {"Data": "Unit"}, "decision": "KeepTarget"}], "reset": []}], "candidates": [], "accepted_impacts": []} validate("SemanticTransitionRequest", request) one_sided = copy.deepcopy(request) one_sided["reviews"][0]["resolutions"] = [{"facet": {"Literal": {"oa": "Only source"}}, "decision": "KeepTarget"}] validate("SemanticTransitionRequest", one_sided) admitted, _ = call("POST", path + "/semantic-transitions/preview", one_sided, etag) validate("SemanticTransitionEnvelope", admitted) assert admitted["plan"]["conflicts"] # Valid typed input, inapplicable to the integer/real fixture. preview, _ = call("POST", path + "/semantic-transitions/preview", request, etag) validate("SemanticTransitionEnvelope", preview) assert not preview["plan"]["conflicts"], preview result, updated_etag = call("POST", path + "/semantic-transitions/apply", preview, etag) validate("MutationResult", result) stored, _ = call("GET", path) validate("StoredProject", stored) validate("Project", stored["model"]) baseline = stored["model"]["traces"][trace]["baseline"] validate("TraceBaseline", baseline) validate("SemanticBaseline", baseline["semantic"]) query.pop("revision") queried, _ = call("POST", path + "/semantic-transitions/query", query) validate("SemanticQueryResult", queried) problem, _ = call("POST", path + "/semantic-transitions/apply", preview, etag, expected=412) validate("Problem", problem) assert updated_etag != etag invalid = [ ("FacetSelector", {"path": ["definition", "minimum"]}), ("FacetSelector", {"Data": "Unsupported"}), ("FacetReview", {"Data": {"IntegerMinimum": {"oa": {"Value": "wrong"}, "sa": {"Value": None}}}}), ("FacetReview", {"Data": {"Unit": {"oa": {"Value": 42}, "sa": "NotApplicable"}}}), ("FacetSelector", {"Literal": {"oa": "literal", "sa": None, "index": 0}}), ("SemanticQueryRequest", {"traces": [trace, trace]}), ("SemanticQueryRequest", {"traces": []}), ("SemanticTransitionEnvelope", {**preview, "unknown": True}), ] obsolete = copy.deepcopy(stored["model"]) obsolete["schema_version"] = 4 invalid.append(("Project", obsolete)) for name, value in invalid: assert not validator(local, name).is_valid(value), (name, value) assert not validator(served, name).is_valid(value), (name, value) # NotApplicable is distinct from Value(null); each leaf fixes the optionality. validate("FacetReview", {"Data": {"IntegerMinimum": {"oa": "NotApplicable", "sa": {"Value": None}}}}) validate("FacetSelector", {"Annotation": {"oa": {"path": [oa], "field": sa}, "sa": None}}) validate("FacetSelector", {"Annotation": {"oa": {"path": [oa], "field": sa}}}) validate("SemanticImpact", {"object": oa, "after": {"Element": stored["model"]["elements"][oa]}}) print(f"{len(local['components']['schemas'])} valid schemas; {checked} actual/typed values validated against served+local contracts; {len(invalid)} invalid shapes rejected") finally: process.terminate() process.wait(timeout=15)