150 lines
8.4 KiB
Python
150 lines
8.4 KiB
Python
"""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))
|