367 lines
24 KiB
Python
367 lines
24 KiB
Python
"""Refresh the closed semantic OpenAPI catalog from the frozen Rust leaf table.
|
|
|
|
Run from any directory with Python 3: semantic_schema.py [--check] [--validate].
|
|
--validate additionally runs source-7 structural fixtures with jsonschema.
|
|
The desktop stored-history editor reads the checked-in descriptor for field
|
|
shapes. Core types and validation remain authoritative for model mutations.
|
|
"""
|
|
import copy
|
|
import json
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[3]
|
|
PATH = ROOT / "docs/openapi.json"
|
|
spec = json.loads(PATH.read_text(encoding="utf-8"))
|
|
schemas = spec["components"]["schemas"]
|
|
types = (ROOT / "crates/oa-core/src/semantic/types.rs").read_text(encoding="utf-8")
|
|
model = (ROOT / "crates/oa-core/src/model.rs").read_text(encoding="utf-8")
|
|
views = (ROOT / "crates/oa-core/src/view_options.rs").read_text(encoding="utf-8")
|
|
|
|
|
|
def ref(name):
|
|
return {"$ref": f"#/components/schemas/{name}"}
|
|
|
|
|
|
def obj(properties, required=None):
|
|
return {"type": "object", "properties": properties,
|
|
"required": list(properties) if required is None else required,
|
|
"additionalProperties": False}
|
|
|
|
|
|
def array(item, maximum=None):
|
|
result = {"type": "array", "items": item}
|
|
if maximum is not None:
|
|
result["maxItems"] = maximum
|
|
return result
|
|
|
|
|
|
def optional(item):
|
|
return {"anyOf": [item, {"type": "null"}]}
|
|
|
|
|
|
def enum(names):
|
|
return {"type": "string", "enum": names}
|
|
|
|
|
|
def variant(name, value):
|
|
return obj({name: value})
|
|
|
|
|
|
def typed(name):
|
|
name = name.replace("crate::view_options::", "")
|
|
if name.startswith("Option<"):
|
|
return optional(typed(name[7:-1]))
|
|
if name.startswith("Vec<"):
|
|
return array(typed(name[4:-1]))
|
|
primitives = {"String": {"type": "string"}, "bool": {"type": "boolean"},
|
|
"u32": {"type": "integer", "minimum": 0, "maximum": 4294967295},
|
|
"i64": {"type": "integer", "minimum": -9223372036854775808,
|
|
"maximum": 9223372036854775807}, "f64": {"type": "number"}}
|
|
return primitives[name] if name in primitives else ref(name)
|
|
|
|
|
|
def reviewed(name):
|
|
name = name.replace("crate::view_options::", "")
|
|
key = "Reviewed" + re.sub(r"[^A-Za-z0-9]", "", name)
|
|
observation = {"oneOf": [enum(["NotApplicable"]), variant("Value", typed(name))]}
|
|
schemas[key] = obj({"oa": observation, "sa": copy.deepcopy(observation)})
|
|
return ref(key)
|
|
|
|
|
|
def keypair(name):
|
|
key = "KeyPair" + name
|
|
schemas[key] = obj({"oa": optional(typed(name)), "sa": optional(typed(name))}, [])
|
|
return ref(key)
|
|
|
|
|
|
for name in ["DataConstructor", "PortConstructor", "OccurrenceConstructor", "EventConstructor",
|
|
"MessageConstructor", "ReferenceCollection", "OrderedCollection", "IncidentEnd",
|
|
"ArtifactFacet", "ComponentPortKind", "ComponentPortOrientation", "PortDirection",
|
|
"FragmentOperator", "StateActionKind", "ArtifactRelation", "ExchangeFamily",
|
|
"Coverage", "ExchangeLabels", "Kind", "RelationKind"]:
|
|
source = next(s for s in [types, model, views] if f"pub enum {name}" in s)
|
|
body = re.search(r"pub enum " + name + r"\s*\{([^}]+)\}", source).group(1)
|
|
body = re.sub(r"//[^\n]*|#\[[^\]]*\]", "", body)
|
|
names = [value.strip() for value in body.split(",") if value.strip()]
|
|
assert all(re.fullmatch(r"[A-Za-z]+", value) for value in names), (name, names)
|
|
schemas[name] = enum(names)
|
|
|
|
catalog = types.split("macro_rules! scalar_families", 1)[1].split("pub(super) use", 1)[0]
|
|
families = re.findall(r"(\w+), (\w+), (\w+) \{([^}]+)\}", catalog)
|
|
assert len(families) == 14
|
|
selectors, reviews = [], []
|
|
leaf_count = 0
|
|
for family, selector, review, body in families:
|
|
leaves = re.findall(r"(\w+): ([\w:<>]+)", body)
|
|
leaf_count += len(leaves)
|
|
schemas[selector] = enum([leaf for leaf, _ in leaves])
|
|
schemas[review] = {"oneOf": [variant(leaf, reviewed(value)) for leaf, value in leaves]}
|
|
selectors.append(variant(family, ref(selector)))
|
|
reviews.append(variant(family, ref(review)))
|
|
assert leaf_count == 90
|
|
schemas["AnnotationKey"] = obj({"path": array(ref("Id")), "field": ref("Id")})
|
|
id_keys, text_keys, annotation_keys = keypair("Id"), keypair("String"), keypair("AnnotationKey")
|
|
artifact_types = {"Presence": "bool", "Name": "String", "Description": "String", "Owner": "Id",
|
|
"Relation": "ArtifactRelation", "Uri": "String", "Revision": "Option<String>",
|
|
"Fragment": "Option<String>"}
|
|
schemas["ArtifactReview"] = {"oneOf": [variant(k, reviewed(v)) for k, v in artifact_types.items()]}
|
|
selectors += [variant("Member", obj({"collection": ref("ReferenceCollection"), "keys": id_keys})),
|
|
variant("Order", ref("OrderedCollection")), variant("Literal", text_keys),
|
|
variant("Annotation", annotation_keys), variant("Binding", id_keys), variant("Child", id_keys),
|
|
variant("Incident", obj({"keys": id_keys, "end": ref("IncidentEnd")})),
|
|
variant("Artifact", obj({"keys": id_keys, "facet": ref("ArtifactFacet")}))]
|
|
reviews += [variant("Member", obj({"collection": ref("ReferenceCollection"), "keys": id_keys, "review": reviewed("bool")})),
|
|
variant("Order", obj({"collection": ref("OrderedCollection"), "review": reviewed("Vec<Id>")})),
|
|
variant("Literal", obj({"keys": text_keys, "review": reviewed("bool")})),
|
|
variant("Annotation", obj({"keys": annotation_keys, "review": reviewed("Option<MessageValue>")})),
|
|
variant("Binding", obj({"keys": id_keys, "review": reviewed("Option<LifelineBinding>")})),
|
|
variant("Child", obj({"keys": id_keys, "review": reviewed("bool")})),
|
|
variant("Incident", obj({"keys": id_keys, "end": ref("IncidentEnd"), "review": reviewed("bool")})),
|
|
variant("Artifact", obj({"keys": id_keys, "review": ref("ArtifactReview")}))]
|
|
schemas["FacetSelector"] = {"oneOf": selectors}
|
|
schemas["FacetReview"] = {"oneOf": reviews}
|
|
schemas["SemanticBaseline"] = obj({"facets": array(ref("FacetReview"))})
|
|
schemas["TraceBaseline"]["properties"]["semantic"] = {**optional(ref("SemanticBaseline")), "default": None,
|
|
"description": "Absent/null means Unreviewed. Historical typed facts never become live graph references."}
|
|
schemas["Project"]["properties"]["schema_version"]["const"] = 7
|
|
schemas["MigrationReport"]["properties"]["from_schema"]["enum"] = [1, 2, 3, 4, 5, 6]
|
|
schemas["MigrationReport"]["properties"]["to_schema"]["const"] = 7
|
|
schemas["MigrationReport"]["description"] = re.sub(r"V1(?:/V[2-6])+ to V[4-7]", "V1/V2/V3/V4/V5/V6 to V7", schemas["MigrationReport"]["description"])
|
|
spec["info"]["description"] = re.sub(r"schema-[4-6]", "schema-7", spec["info"]["description"])
|
|
spec["paths"]["/api/v1/projects/import"]["post"]["description"] = "Schema-7 native JSON is described below. Schema-1/2/3/4/5/6 JSON and source are accepted through explicit frozen readers and migration, also on replacement and candidate validation. Earlier versions reject later constructors, fields and semantic facets, including explicit null. Prior prose and typed reviews remain intact; absent semantic review stays Unreviewed. Export emits schema 7 / arcadia 7. See docs/language.rst."
|
|
schemas["SourceBundle"]["description"] = re.sub(r"source[4-6] workspace \(source[^)]+\)", "source7 workspace (source3/4/5/6 migrate explicitly)", schemas["SourceBundle"]["description"])
|
|
|
|
# Source-6 additions are closed serde records; selectors with an empty payload
|
|
# remain explicit selections rather than reverting to their omitted defaults.
|
|
schemas["EdgeKind"] = {"oneOf": [enum(["Trace"]), variant("Relation", ref("RelationKind"))]}
|
|
schemas["RelationKinds"] = {"oneOf": [enum(["All"]), variant("Only", array(ref("EdgeKind")))]}
|
|
for selection in ["CategorySelection", "ContextSelection"]:
|
|
schemas[selection] = {"oneOf": [enum(["All"]), variant("Any", array(ref("Id")))]}
|
|
schemas["ViewOptions"] = obj({
|
|
"hidden_edges": {**array(ref("Id")), "default": []},
|
|
"relation_kinds": {**ref("RelationKinds"), "default": "All"},
|
|
"work_categories": {**ref("CategorySelection"), "default": "All"},
|
|
"participant_categories": {**ref("CategorySelection"), "default": "All"},
|
|
"coverage": {**ref("Coverage"), "default": "All"},
|
|
"work_contexts": {**ref("ContextSelection"), "default": "All"},
|
|
"exchange_labels": {**ref("ExchangeLabels"), "default": "Name"}}, [])
|
|
schemas["ViewOptions"]["description"] = "Presentation only. Independent predicates intersect; category members union within a family. Only([]) hides all relations and traces. Any([]) remains an explicit empty selection. Traceability permits both-layer hidden/category/context references; ordinary projections require the diagram layer. Filtering preserves unfiltered automatic port positions."
|
|
schemas["Diagram"]["properties"]["view_options"] = {**ref("ViewOptions"), "default": {}}
|
|
|
|
def upsert_variant(variants, name, value):
|
|
variants[:] = [v for v in variants if name not in v.get("properties", {})]
|
|
variants.append(variant(name, value))
|
|
|
|
upsert_variant(schemas["RelationBody"]["oneOf"], "CommunicationMean", obj({
|
|
"source": ref("Id"), "target": ref("Id"), "oriented": typed("bool"), "items": array(ref("Id"))}))
|
|
|
|
# Source-7 adds exact caller-instance addresses and explicit operation intent.
|
|
# Structural schemas do not replace core owner/path/port/signature validation.
|
|
schemas["OccurrenceEndpoint"] = obj({"references": array(ref("Id"), 64), "occurrence": ref("Id")})
|
|
schemas["OccurrenceEndpoint"]["description"] = "Ordered reference-occurrence IDs in an explicit caller owner, followed by the leaf contribution ID. Empty references means a direct contribution; sequence definition IDs are not path components."
|
|
schemas["ContextualLinkPurpose"] = {"oneOf": [
|
|
variant("Precedence", obj({"guard": typed("String")})),
|
|
variant("Exchange", obj({"exchange": ref("Id")})),
|
|
variant("CommonWork", obj({"work": ref("Id")}))]}
|
|
upsert_variant(schemas["RelationBody"]["oneOf"], "ContextualLink", obj({
|
|
"owner": ref("Id"), "source": ref("OccurrenceEndpoint"),
|
|
"target": ref("OccurrenceEndpoint"), "purpose": ref("ContextualLinkPurpose")}))
|
|
schemas["OperationUse"] = obj({"allocation": optional(ref("Id"))}, [])
|
|
schemas["OperationUse"]["description"] = "Present with absent/null allocation retains unresolved operation intent. A selected UUID identifies exactly an InterfaceItemAllocation, never an item, port, or name. Reply derives operation from its exact correlated Call."
|
|
exchange_body = next(v["properties"]["Exchange"] for v in schemas["OccurrenceBody"]["oneOf"]
|
|
if "Exchange" in v.get("properties", {}))
|
|
exchange_body["properties"]["operation"] = {**optional(ref("OperationUse")), "default": None}
|
|
exchange_body["description"] = "Absent/null operation retains ordinary channel semantics and is omitted on output. Present operation is allowed only on SA BoundaryInterface Call/Asynchronous messages. Reply stores no operation and derives its exact Call's selection. Existing message defaults remain unchanged."
|
|
schemas["OperationOmission"] = obj({"source": ref("Id"), "rationale": typed("String")})
|
|
schemas["TransitionRequest"]["properties"]["operation_omissions"] = {
|
|
**array(ref("OperationOmission"), 200), "default": [],
|
|
"description": "Explicit SA-to-OA abstraction per selected invocation with operation intent. Nonempty rationale required by core; duplicates, irrelevant sources, and other directions are refused. Replies derive the Call abstraction. Source operation identities and values are preserved."}
|
|
upsert_variant(schemas["SharedBody"]["oneOf"][1]["oneOf"], "ExchangeCategory", obj({
|
|
"family": ref("ExchangeFamily"), "exchanges": array(ref("Id"))}))
|
|
schemas["CommunicationPortChoice"] = {"oneOf": [
|
|
variant("Reuse", obj({"port": ref("Id")})),
|
|
variant("Create", obj({"id": ref("Id"), "alias": typed("String"), "name": typed("String"),
|
|
"description": typed("String"), "kind": ref("ComponentPortKind"),
|
|
"orientation": ref("ComponentPortOrientation")}))]}
|
|
schemas["CommunicationEndpointDecision"] = obj({"source": ref("Id"),
|
|
"source_port": ref("CommunicationPortChoice"), "target_port": ref("CommunicationPortChoice")})
|
|
schemas["TransitionRequest"]["properties"]["communication_endpoints"] = {
|
|
**array(ref("CommunicationEndpointDecision"), 200), "default": [],
|
|
"description": "Explicit SA component-port pair per selected OA CommunicationMean. Reuse or Create only. Owners follow reviewed participant mappings; no participant-to-port substitution or automatic interface allocation."}
|
|
schemas["Drift"] = enum(["Unreviewed", "Unchanged", "OaChanged", "SaChanged", "BothChanged"])
|
|
schemas["FacetDrift"] = obj({"selector": ref("FacetSelector"), "current": ref("FacetReview"), "drift": ref("Drift")})
|
|
schemas["SemanticDrift"] = obj({"own": ref("Drift"), "linked": ref("Drift"),
|
|
"own_unreviewed": typed("bool"), "linked_unreviewed": typed("bool")})
|
|
schemas["LinkedSemanticImpact"] = obj({"objects": id_keys, "traces": array(ref("Id")),
|
|
"drift": ref("Drift"), "unreviewed": typed("bool")})
|
|
for name, members in [("SemanticRecord", ["Element", "Relation", "Artifact"]),
|
|
("ImpactRecord", ["Element", "Relation", "Artifact", "Diagram", "Trace"])]:
|
|
schemas[name] = {"oneOf": [variant(n, ref("ArtifactLink" if n == "Artifact" else n)) for n in members]}
|
|
schemas["SemanticImpact"] = obj({"object": ref("Id"), "before": optional(ref("ImpactRecord")), "after": optional(ref("ImpactRecord"))}, ["object"])
|
|
schemas["SemanticDecision"] = enum(["KeepTarget", "TakeSource", "Candidate"])
|
|
schemas["FacetResolution"] = obj({"facet": ref("FacetSelector"), "decision": ref("SemanticDecision")})
|
|
schemas["SemanticReviewRequest"] = obj({"source": ref("Id"), "resolutions": array(ref("FacetResolution"), 5000),
|
|
"reset": {**array(ref("FacetSelector"), 5000), "default": []}}, ["source", "resolutions"])
|
|
|
|
|
|
def mappings(maximum):
|
|
return {"type": "object", "propertyNames": {"format": "uuid"},
|
|
"additionalProperties": ref("Id"), "maxProperties": maximum, "default": {}}
|
|
|
|
|
|
schemas["SemanticTransitionRequest"] = obj({"transition": ref("TransitionRequest"), "traces": mappings(100),
|
|
"references": mappings(5000), "reviews": {**array(ref("SemanticReviewRequest"), 100), "default": []},
|
|
"candidates": {**array(ref("SemanticRecord"), 200), "default": []},
|
|
"accepted_impacts": {**array(ref("SemanticImpact"), 200), "default": []}}, ["transition"])
|
|
schemas["SemanticTransitionRequest"]["description"] = "Canonical OA-to-SA references include historical UUIDs. Traces map selected source IDs to trace IDs. Combined resolutions/reset limit: 5,000. Candidates must equal exactly the result of their selected facets; accepted_impacts explicitly acknowledge exact induced changes."
|
|
schemas["SemanticReviewPreview"] = obj({"source": ref("Id"), "target": ref("Id"), "facets": array(ref("FacetDrift")),
|
|
"acknowledged": array(ref("FacetSelector")), "retired": array(ref("FacetSelector")), "linked_impacts": array(ref("LinkedSemanticImpact"))})
|
|
schemas["SemanticTransitionPlan"] = obj({"base_fingerprint": typed("String"), "request": ref("SemanticTransitionRequest"),
|
|
"transition": ref("TransitionPlan"), "reviews": array(ref("SemanticReviewPreview")), "impacts": array(ref("SemanticImpact")),
|
|
"conflicts": array(typed("String")), "commands": array(ref("Command"), 200), "prepared_fingerprint": optional(typed("String"))})
|
|
schemas["SemanticTransitionPlan"]["required"].remove("prepared_fingerprint")
|
|
schemas["SemanticTransitionEnvelope"] = obj({"base_revision": ref("Revision"), "plan": ref("SemanticTransitionPlan")})
|
|
schemas["SemanticQueryRequest"] = obj({"traces": {**array(ref("Id"), 100), "minItems": 1, "uniqueItems": True},
|
|
"references": mappings(5000), "revision": optional(ref("Revision"))}, ["traces"])
|
|
schemas["SemanticQueryItem"] = obj({"trace": ref("Id"), "oa": ref("Id"), "sa": ref("Id"),
|
|
"facets": array(ref("FacetDrift")), "drift": ref("SemanticDrift"), "linked_impacts": array(ref("LinkedSemanticImpact"))})
|
|
schemas["SemanticQueryResult"] = obj({"revision": ref("Revision"), "items": array(ref("SemanticQueryItem"), 100)})
|
|
for operation, request, result in [("query", "SemanticQueryRequest", "SemanticQueryResult"),
|
|
("preview", "SemanticTransitionRequest", "SemanticTransitionEnvelope"),
|
|
("apply", "SemanticTransitionEnvelope", "MutationResult")]:
|
|
endpoint = copy.deepcopy(spec["paths"]["/api/v1/projects/{id}/transitions/" + ("apply" if operation == "apply" else "preview")])
|
|
post = endpoint["post"]
|
|
post["operationId"] = operation + "SemanticTransition"
|
|
post["summary"] = {"query": "Query typed trace drift and linked impacts without mutation",
|
|
"preview": "Preview and source-validate a typed semantic transition without writing",
|
|
"apply": "Conditionally apply a prepared semantic candidate in one model/source revision"}[operation]
|
|
post["requestBody"]["content"]["application/json"]["schema"] = ref(request)
|
|
post["responses"]["200"]["content"]["application/json"]["schema"] = ref(result)
|
|
post["description"] = "Uses the frozen core typed semantic contract. A supplied plan is a client-owned candidate, not a signed receipt; fingerprints check consistency, not original review authorship. Preview validates source writeback when conflict-free. Apply preserves preview-generated IDs and does not replan random identities."
|
|
if operation == "query":
|
|
post["parameters"] = [p for p in post["parameters"] if p["in"] != "header"]
|
|
post["responses"].pop("412", None)
|
|
post["responses"].pop("428", None)
|
|
post["responses"]["409"] = {"$ref": "#/components/responses/Problem"}
|
|
post["description"] = "Read one immutable snapshot. Optional revision mismatch returns409; missing trace returns404; duplicate/empty trace selection returns400. References are canonical OA-to-SA correspondence. No write, baseline refresh or external artifact fetch."
|
|
spec["paths"]["/api/v1/projects/{id}/semantic-transitions/" + operation] = endpoint
|
|
|
|
|
|
def check_refs(value):
|
|
if isinstance(value, dict):
|
|
if "$ref" in value and value["$ref"].startswith("#/components/schemas/"):
|
|
assert value["$ref"].split("/")[-1] in schemas, value
|
|
for child in value.values():
|
|
check_refs(child)
|
|
elif isinstance(value, list):
|
|
for child in value:
|
|
check_refs(child)
|
|
|
|
|
|
check_refs(spec)
|
|
|
|
if "--validate" in sys.argv:
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
|
|
for schema in schemas.values():
|
|
Draft202012Validator.check_schema(schema)
|
|
|
|
fixture_count = 0
|
|
|
|
def accepts(name, value, expected=True):
|
|
global fixture_count
|
|
validator = Draft202012Validator({**ref(name), "components": spec["components"]},
|
|
format_checker=FormatChecker())
|
|
errors = list(validator.iter_errors(value))
|
|
assert (not errors) == expected, (name, value, [e.message for e in errors])
|
|
fixture_count += 1
|
|
|
|
a, b = "00000000-0000-4000-8000-000000000001", "00000000-0000-4000-8000-000000000002"
|
|
mean = {"source": a, "target": b, "oriented": False, "items": []}
|
|
accepts("RelationBody", {"CommunicationMean": mean})
|
|
accepts("RelationBody", {"CommunicationMean": {**mean, "typo": True}}, False)
|
|
for field in mean:
|
|
accepts("RelationBody", {"CommunicationMean": {k: v for k, v in mean.items() if k != field}}, False)
|
|
category = {"family": "Participant", "exchanges": [a]}
|
|
accepts("SharedBody", {"ExchangeCategory": category})
|
|
accepts("SharedBody", {"ExchangeCategory": {**category, "family": "Future"}}, False)
|
|
accepts("SharedBody", {"ExchangeCategory": {**category, "typo": []}}, False)
|
|
accepts("ViewOptions", {})
|
|
options = {"hidden_edges": [a], "relation_kinds": {"Only": ["Trace", {"Relation": "CommunicationMean"}]},
|
|
"work_categories": {"Any": []}, "participant_categories": {"Any": [b]},
|
|
"coverage": "OutsideBehavior", "work_contexts": {"Any": []}, "exchange_labels": "NameAndItems"}
|
|
accepts("ViewOptions", options)
|
|
accepts("ViewOptions", {**options, "relation_kinds": {"Only": []}})
|
|
for field, value in [("typo", []), ("coverage", "Unknown"), ("hidden_edges", ["not-an-id"]),
|
|
("relation_kinds", {"Only": [], "typo": []}),
|
|
("relation_kinds", {"Only": [{"Relation": "Future"}]}),
|
|
("work_contexts", {"Any": [], "typo": []}),
|
|
("exchange_labels", "Names")]:
|
|
accepts("ViewOptions", {**options, field: value}, False)
|
|
create = {"id": a, "alias": "radio", "name": "Radio", "description": "",
|
|
"kind": "Standard", "orientation": "Unset"}
|
|
accepts("CommunicationPortChoice", {"Create": create})
|
|
accepts("CommunicationPortChoice", {"Reuse": {"port": a}})
|
|
accepts("CommunicationPortChoice", "Auto", False)
|
|
accepts("CommunicationPortChoice", {"Reuse": {"port": a, "owner": b}}, False)
|
|
for field in create:
|
|
accepts("CommunicationPortChoice", {"Create": {k: v for k, v in create.items() if k != field}}, False)
|
|
pair = {"source": a, "source_port": {"Create": create}, "target_port": {"Reuse": {"port": b}}}
|
|
accepts("CommunicationEndpointDecision", pair)
|
|
accepts("CommunicationEndpointDecision", {k: v for k, v in pair.items() if k != "target_port"}, False)
|
|
request = {"direction": "OaToSa", "sources": [a], "scope": "SelectedOnly", "mappings": [],
|
|
"fields": [], "resolutions": [], "destination_diagram": None}
|
|
accepts("TransitionRequest", request)
|
|
accepts("TransitionRequest", {**request, "communication_endpoints": [pair]})
|
|
accepts("TransitionRequest", {**request, "communication_endpoints": [pair] * 200})
|
|
accepts("TransitionRequest", {**request, "communication_endpoints": [pair] * 201}, False)
|
|
accepts("FacetSelector", {"Category": "Family"})
|
|
accepts("FacetSelector", {"Category": "Unknown"}, False)
|
|
accepts("FacetReview", {"Category": {"Family": {"oa": {"Value": "Work"}, "sa": {"Value": "Participant"}}}})
|
|
accepts("FacetSelector", {"Member": {"collection": "CategoryExchanges", "keys": {"oa": a, "sa": b}}})
|
|
endpoint = {"references": [a], "occurrence": b}
|
|
accepts("OccurrenceEndpoint", endpoint)
|
|
accepts("OccurrenceEndpoint", {**endpoint, "references": []})
|
|
accepts("OccurrenceEndpoint", {**endpoint, "references": [a] * 64})
|
|
accepts("OccurrenceEndpoint", {**endpoint, "references": [a] * 65}, False)
|
|
accepts("OccurrenceEndpoint", {**endpoint, "typo": a}, False)
|
|
accepts("OccurrenceEndpoint", {"occurrence": a}, False)
|
|
for purpose in [{"Precedence": {"guard": "ready"}}, {"Exchange": {"exchange": a}},
|
|
{"CommonWork": {"work": b}}]:
|
|
link = {"owner": a, "source": endpoint, "target": {"references": [], "occurrence": a},
|
|
"purpose": purpose}
|
|
accepts("RelationBody", {"ContextualLink": link})
|
|
accepts("RelationBody", {"ContextualLink": {**link, "extra": None}}, False)
|
|
accepts("ContextualLinkPurpose", {"CommonWork": {"work": a, "guard": ""}}, False)
|
|
accepts("OperationUse", {})
|
|
accepts("OperationUse", {"allocation": None})
|
|
accepts("OperationUse", {"allocation": a})
|
|
accepts("OperationUse", {"allocation": a, "item": b}, False)
|
|
accepts("OperationUse", {"allocation": "wrong"}, False)
|
|
message = {"exchange": a, "sender": a, "receiver": b, "message_kind": "Call"}
|
|
for operation in [None, {}, {"allocation": None}, {"allocation": a}]:
|
|
accepts("OccurrenceBody", {"Exchange": {**message, "operation": operation}})
|
|
accepts("OccurrenceBody", {"Exchange": {**message, "operation": a}}, False)
|
|
omission = {"source": a, "rationale": "Operational abstraction retains the request intent."}
|
|
accepts("OperationOmission", omission)
|
|
accepts("OperationOmission", {"source": a}, False)
|
|
accepts("TransitionRequest", {**request, "operation_omissions": [omission] * 200})
|
|
accepts("TransitionRequest", {**request, "operation_omissions": [omission] * 201}, False)
|
|
accepts("FacetSelector", {"Occurrence": "Operation"})
|
|
for facet in ["ContextOwner", "ContextSource", "ContextTarget", "ContextPurpose"]:
|
|
accepts("FacetSelector", {"Relation": facet})
|
|
accepts("FacetReview", {"Occurrence": {"Operation": {
|
|
"oa": "NotApplicable", "sa": {"Value": {"allocation": a}}}}})
|
|
assert schemas["Project"]["properties"]["schema_version"]["const"] == 7
|
|
assert schemas["MigrationReport"]["properties"]["from_schema"]["enum"] == [1, 2, 3, 4, 5, 6]
|
|
print(f"Source-7 schema fixtures: {fixture_count} passed; all {len(schemas)} component schemas well-formed")
|
|
|
|
rendered = json.dumps(spec, indent=2, ensure_ascii=False) + "\n"
|
|
if "--check" in sys.argv:
|
|
assert json.loads(PATH.read_text(encoding="utf-8")) == spec, "Run semantic_schema.py to refresh OpenAPI"
|
|
else:
|
|
PATH.write_text(rendered, encoding="utf-8")
|
|
print(f"Closed catalog: {len(families)} families, {leaf_count} scalar leaves; {len(schemas)} schemas")
|