Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 50 additions & 7 deletions scripts/validate_control_plane_wrapper_ids.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,8 +7,12 @@
canonical `$id` values under `https://schemas.srcos.ai/v2/control-plane/...`.
- The wrappers `allOf`-wrap legacy `*.schema.json` files.

We validate that a minimal instance can be validated against a schema that `$ref`s
one of the canonical wrapper `$id` values.
We validate **two properties**:

1) Wrapper `$id` values resolve from the local schema registry.
2) Wrapper-internal `$ref` targets resolve (e.g., the legacy `*.schema.json` files).

Additionally, we run one end-to-end validation using IncidentEvent (minimal instance).

Exit:
- 0 on success
Expand All @@ -29,6 +33,19 @@ def load_json(path: str) -> dict:
return json.load(f)


def collect_refs(obj) -> set[str]:
refs: set[str] = set()
if isinstance(obj, dict):
for k, v in obj.items():
if k == "$ref" and isinstance(v, str):
refs.add(v)
refs |= collect_refs(v)
elif isinstance(obj, list):
for it in obj:
refs |= collect_refs(it)
return refs


def main() -> int:
repo = Path(__file__).resolve().parents[1]
schema_dir = repo / "schemas"
Expand Down Expand Up @@ -72,12 +89,38 @@ def resolve_remote(self, uri):

resolver = LocalRegistry(base_uri=base_uri, referrer={}, store=registry)

# Minimal canonical wrapper `$id` resolution test: IncidentEvent
wrapper_id = "https://schemas.srcos.ai/v2/control-plane/IncidentEvent.json"
wrappers = [
"https://schemas.srcos.ai/v2/control-plane/IncidentEvent.json",
"https://schemas.srcos.ai/v2/control-plane/MeshSkill.json",
"https://schemas.srcos.ai/v2/control-plane/EnrollmentToken.json",
"https://schemas.srcos.ai/v2/control-plane/SkillExecutionEvent.json",
]

# 1) Resolution checks (wrapper id + internal refs)
for wid in wrappers:
try:
schema = resolver.resolve_remote(wid)
except Exception as e:
print(f"FAIL: cannot resolve wrapper $id: {wid}: {e}", file=sys.stderr)
return 1

if isinstance(schema, dict):
# Ensure internal refs resolve (allOf wraps legacy schema)
for r in sorted(collect_refs(schema)):
try:
resolver.resolve_remote(r)
except Exception as e:
print(f"FAIL: wrapper {wid} contains unresolved $ref {r}: {e}", file=sys.stderr)
return 1

print(f"OK: wrapper resolved: {wid}")

# 2) End-to-end instance validation (IncidentEvent)
incident_wrapper = wrappers[0]

schema_ref = {
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$ref": wrapper_id,
"$ref": incident_wrapper,
}

instance = {
Expand All @@ -91,10 +134,10 @@ def resolve_remote(self, uri):
try:
validate(instance, schema_ref, resolver=resolver)
except Exception as e:
print(f"FAIL: canonical wrapper $id resolution failed: {e}", file=sys.stderr)
print(f"FAIL: IncidentEvent wrapper end-to-end validation failed: {e}", file=sys.stderr)
return 1

print("OK: control-plane canonical wrapper $id resolution")
print("OK: IncidentEvent wrapper end-to-end validation")
return 0


Expand Down