"""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 sys.path.insert(0, 'C:\\Users\\User\\Documents\\Git\\ARCADIA-Studio\\.local\\python-deps') from jsonschema import Draft202012Validator, FormatChecker ROOT = Path('C:\\Users\\User\\Documents\\Git\\ARCADIA-Studio') EVIDENCE = Path('C:\\Users\\User\\Documents\\Git\\ARCADIA-Studio\\.local\\packaged-server-865483b-20260911') import contextlib 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 contextlib.nullcontext(str(EVIDENCE / "schema-store")) 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: (EVIDENCE / 'schema-launch.json').write_text(json.dumps({'pid':process.pid,'args':process.args,'cwd':os.getcwd(),'PATH':env['PATH']},indent=2)) banner = process.stderr.readline() (EVIDENCE / 'schema-server.stderr.log').write_text(banner) responses = [] 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: raw = response.read() data = json.loads(raw) responses.append({'method':method,'path':path,'body':body,'status':response.status,'etag':response.headers.get('ETag'),'response':raw.decode()}) (EVIDENCE / 'schema-http.json').write_text(json.dumps(responses,indent=2)) 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) stdout, stderr = process.communicate(timeout=15) (EVIDENCE / 'schema-server.stdout.log').write_text(stdout) with (EVIDENCE / 'schema-server.stderr.log').open('a') as log: log.write(stderr) (EVIDENCE / 'schema-stop.json').write_text(json.dumps({'pid':process.pid,'exit_code':process.returncode,'has_exited':process.poll() is not None,'method':'owned process terminate in original finally'},indent=2))