tegwick
072a7fa550
The snapshot and manifest schemas, both example dataspaces, and the shared text renderer now pass the fixture validator and the negative contract tests. JSON, NDJSON, and Markdown frontmatter publish a snapshot only when rights, identity, and references allow it. CONTENT-WP-0001 and CONTENT-WP-0003 are marked finished against those artifacts. CONTENT-WP-0000-T04 stays open. SCOPE.md and the 19:05 scope-against-intent note record the stop. Assistant: grok Assistant-Session: 01a130f1-abfb-7d71-9cd6-a8b75ec7f95a
236 lines
9.1 KiB
Python
236 lines
9.1 KiB
Python
"""Validate a v0.1 snapshot against its manifest.
|
|
|
|
Publication checks live here. Callers pass documents they already loaded.
|
|
`validate_dataspace` is the file entry point used by `scripts/validate_fixtures.py`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import jsonschema
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
SNAPSHOT_SCHEMA_PATH = ROOT / "contracts" / "snapshot.schema.json"
|
|
MANIFEST_SCHEMA_PATH = ROOT / "contracts" / "manifest.schema.json"
|
|
SUPPORTED_VERSIONS = frozenset({"0.1"})
|
|
ENTITY_ID_PREFIX = "urn:ce:"
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def _validator(schema_path: Path) -> jsonschema.Draft202012Validator:
|
|
schema = load_json(schema_path)
|
|
return jsonschema.Draft202012Validator(
|
|
schema,
|
|
format_checker=jsonschema.Draft202012Validator.FORMAT_CHECKER,
|
|
)
|
|
|
|
|
|
def schema_errors(document: dict, schema_path: Path) -> list[str]:
|
|
validator = _validator(schema_path)
|
|
found = sorted(validator.iter_errors(document), key=lambda err: list(err.path))
|
|
messages = []
|
|
for err in found:
|
|
where = "/".join(str(part) for part in err.path) or "(root)"
|
|
messages.append(f"schema {where}: {err.message}")
|
|
return messages
|
|
|
|
|
|
def version_error(document: dict, label: str) -> str | None:
|
|
version = document.get("schemaVersion")
|
|
if version in SUPPORTED_VERSIONS:
|
|
return None
|
|
return (
|
|
f"{label} schemaVersion {version!r} is not supported; "
|
|
f"expected one of {', '.join(sorted(SUPPORTED_VERSIONS))}"
|
|
)
|
|
|
|
|
|
def resolve_claims(claims: list[dict], key: str, policy: str) -> dict[str, Any]:
|
|
"""Choose what a display field may show for one subject.
|
|
|
|
`missing` means no claim exists. `unknown` means a claim records that the
|
|
value is not known. Differing asserted values are a contradiction. The
|
|
manifest policy `show-all` keeps every asserted value, `withhold` keeps
|
|
none, and `preferred` keeps the claim marked preferred or withholds when
|
|
none is marked.
|
|
"""
|
|
relevant = [claim for claim in claims if claim.get("key") == key]
|
|
if not relevant:
|
|
return {"state": "missing", "values": []}
|
|
asserted = [claim for claim in relevant if claim.get("status") == "asserted"]
|
|
values = []
|
|
for claim in asserted:
|
|
value = claim.get("value")
|
|
if value not in values:
|
|
values.append(value)
|
|
if len(values) > 1:
|
|
if policy == "show-all":
|
|
return {"state": "contradicted", "values": values}
|
|
if policy == "preferred":
|
|
chosen = [claim.get("value") for claim in asserted if claim.get("preferred") is True]
|
|
if len(chosen) == 1:
|
|
return {"state": "present", "values": chosen}
|
|
return {"state": "contradicted", "values": []}
|
|
return {"state": "contradicted", "values": []}
|
|
if values:
|
|
return {"state": "present", "values": values}
|
|
if any(claim.get("status") == "unknown" for claim in relevant):
|
|
return {"state": "unknown", "values": []}
|
|
if any(claim.get("status") == "withheld" for claim in relevant):
|
|
return {"state": "withheld", "values": []}
|
|
return {"state": "missing", "values": []}
|
|
|
|
|
|
def _cycles(snapshot: dict, predicates: set[str]) -> list[str]:
|
|
if not predicates:
|
|
return []
|
|
adjacency: dict[str, list[tuple[str, str]]] = defaultdict(list)
|
|
for relation in snapshot.get("relations", []):
|
|
if relation.get("predicate") in predicates:
|
|
adjacency[relation["from"]].append((relation["to"], relation["id"]))
|
|
color: dict[str, int] = {}
|
|
errors: list[str] = []
|
|
|
|
def walk(node: str) -> None:
|
|
color[node] = 1
|
|
for nxt, relation_id in adjacency.get(node, []):
|
|
state = color.get(nxt, 0)
|
|
if state == 1:
|
|
errors.append(
|
|
f"taxonomic cycle via {relation_id}: {node} -> {nxt}"
|
|
)
|
|
elif state == 0:
|
|
walk(nxt)
|
|
color[node] = 2
|
|
|
|
for start in list(adjacency):
|
|
if color.get(start, 0) == 0:
|
|
walk(start)
|
|
return errors
|
|
|
|
|
|
def graph_errors(snapshot: dict, manifest: dict) -> list[str]:
|
|
errors: list[str] = []
|
|
version = version_error(snapshot, "snapshot")
|
|
if version:
|
|
errors.append(version)
|
|
manifest_version = version_error(manifest, "manifest")
|
|
if manifest_version:
|
|
errors.append(manifest_version)
|
|
|
|
kinds = [item.get("id") for item in snapshot.get("kinds", [])]
|
|
if len(kinds) != len(set(kinds)):
|
|
errors.append("duplicate kind id")
|
|
kind_ids = set(kinds)
|
|
entity_ids = [item.get("id") for item in snapshot.get("entities", [])]
|
|
if len(entity_ids) != len(set(entity_ids)):
|
|
errors.append("duplicate entity id")
|
|
entities = {item.get("id"): item for item in snapshot.get("entities", []) if item.get("id")}
|
|
sources = {item.get("id") for item in snapshot.get("sources", [])}
|
|
claim_ids = [item.get("id") for item in snapshot.get("claims", [])]
|
|
if len(claim_ids) != len(set(claim_ids)):
|
|
errors.append("duplicate claim id")
|
|
|
|
dataspace_id = (snapshot.get("dataspace") or {}).get("id")
|
|
if manifest.get("id") != dataspace_id:
|
|
errors.append(
|
|
f"manifest id {manifest.get('id')!r} does not match dataspace {dataspace_id!r}"
|
|
)
|
|
|
|
labels = manifest.get("relationLabels") or {}
|
|
views = manifest.get("kindViews") or {}
|
|
for kind_id in kind_ids:
|
|
if kind_id not in views:
|
|
errors.append(f"kind {kind_id} has no manifest kindView")
|
|
for featured in (manifest.get("home") or {}).get("featuredKinds", []):
|
|
if featured not in kind_ids:
|
|
errors.append(f"featured kind {featured} is not in the snapshot")
|
|
|
|
for entity in snapshot.get("entities", []):
|
|
entity_id = entity.get("id", "")
|
|
if not str(entity_id).startswith(ENTITY_ID_PREFIX):
|
|
errors.append(f"entity id {entity_id} is not a urn:ce identifier")
|
|
if entity.get("kindId") not in kind_ids:
|
|
errors.append(f"{entity_id}: undeclared kind {entity.get('kindId')}")
|
|
if not entity.get("sourceRefs"):
|
|
errors.append(f"{entity_id}: missing source")
|
|
for source_id in entity.get("sourceRefs", []):
|
|
if source_id not in sources:
|
|
errors.append(f"{entity_id}: missing source {source_id}")
|
|
for claim_id in entity.get("claimRefs", []):
|
|
if claim_id not in claim_ids:
|
|
errors.append(f"{entity_id}: missing claim {claim_id}")
|
|
|
|
known_entities = set(entities)
|
|
for relation in snapshot.get("relations", []):
|
|
relation_id = relation.get("id")
|
|
predicate = relation.get("predicate")
|
|
if predicate not in labels:
|
|
errors.append(f"relation {relation_id}: predicate {predicate} has no label")
|
|
for field in ("from", "to"):
|
|
target = relation.get(field)
|
|
if target not in known_entities:
|
|
errors.append(f"relation {relation_id}: dangling {field} {target}")
|
|
refs = relation.get("sourceRefs") or []
|
|
if not refs:
|
|
errors.append(f"relation {relation_id}: missing source")
|
|
for source_id in refs:
|
|
if source_id not in sources:
|
|
errors.append(f"relation {relation_id}: missing source {source_id}")
|
|
|
|
for claim in snapshot.get("claims", []):
|
|
claim_id = claim.get("id")
|
|
subject = claim.get("subjectId")
|
|
if subject not in known_entities:
|
|
errors.append(f"claim {claim_id}: dangling subject {subject}")
|
|
evidence = claim.get("evidenceRefs") or []
|
|
if not evidence:
|
|
errors.append(f"claim {claim_id}: missing source")
|
|
for source_id in evidence:
|
|
if source_id not in sources:
|
|
errors.append(f"claim {claim_id}: missing source {source_id}")
|
|
|
|
acyclic = set(manifest.get("acyclicPredicates") or [])
|
|
errors.extend(_cycles(snapshot, acyclic))
|
|
return errors
|
|
|
|
|
|
def size_errors(snapshot: dict, *, min_kinds: int = 3, min_entities: int = 8) -> list[str]:
|
|
errors = []
|
|
kind_count = len(snapshot.get("kinds") or [])
|
|
entity_count = len(snapshot.get("entities") or [])
|
|
if kind_count < min_kinds:
|
|
errors.append(f"needs at least {min_kinds} kinds, found {kind_count}")
|
|
if entity_count < min_entities:
|
|
errors.append(f"needs at least {min_entities} entities, found {entity_count}")
|
|
return errors
|
|
|
|
|
|
def validate_documents(
|
|
snapshot: dict,
|
|
manifest: dict,
|
|
*,
|
|
require_size: bool = False,
|
|
) -> list[str]:
|
|
errors = schema_errors(snapshot, SNAPSHOT_SCHEMA_PATH)
|
|
errors.extend(schema_errors(manifest, MANIFEST_SCHEMA_PATH))
|
|
errors.extend(graph_errors(snapshot, manifest))
|
|
if require_size:
|
|
errors.extend(size_errors(snapshot))
|
|
return errors
|
|
|
|
|
|
def validate_dataspace(directory: Path, *, require_size: bool = False) -> list[str]:
|
|
snapshot_path = directory / "snapshot.json"
|
|
manifest_path = directory / "manifest.json"
|
|
snapshot = load_json(snapshot_path)
|
|
manifest = load_json(manifest_path)
|
|
errors = validate_documents(snapshot, manifest, require_size=require_size)
|
|
return [f"{directory.name}: {message}" for message in errors]
|