"""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]