ila/yovision
1
0
Fork 0
Code Issues 14 Pull Requests 14 Actions Packages Projects Releases Wiki Activity

feat: freeze event evidence contract v1 (#150)

Browse Source
This commit is contained in:
QiuSW
2026-08-31 08:45:57 +08:00
parent 49aa79f3b9
commit 2a395aa126
23 changed files with 740 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
contracts/tests/events-v1/contract_support.py
+91
View File
@@ -0,0 +1,91 @@
"""Small dependency-free validator for the JSON Schema keywords used by v1 contracts."""
from __future__ import annotations
import json
import math
import re
from datetime import datetime
from pathlib import Path
from typing import Any
def load_json(path: Path) -> Any:
return json.loads(path.read_text(encoding="utf-8"))
def canonical_bytes(value: Any) -> bytes:
"""Canonical bytes for checked-in JCS vectors (all vector numbers are JCS-safe)."""
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8")
def validate(instance: Any, schema: dict[str, Any], schema_path: Path, location: str = "$") -> list[str]:
if "$ref" in schema:
ref = schema["$ref"]
if ref.startswith("#"):
return [f"{location}: local fragments are not supported by the contract checker"]
target = (schema_path.parent / ref).resolve()
return validate(instance, load_json(target), target, location)
errors: list[str] = []
for subschema in schema.get("allOf", []):
errors.extend(validate(instance, subschema, schema_path, location))
if "anyOf" in schema and not any(not validate(instance, item, schema_path, location) for item in schema["anyOf"]):
errors.append(f"{location}: does not match anyOf")
if "not" in schema and not validate(instance, schema["not"], schema_path, location):
errors.append(f"{location}: matches forbidden schema")
if "if" in schema and not validate(instance, schema["if"], schema_path, location):
errors.extend(validate(instance, schema.get("then", {}), schema_path, location))
expected = schema.get("type")
type_ok = {
"object": lambda x: isinstance(x, dict),
"array": lambda x: isinstance(x, list),
"string": lambda x: isinstance(x, str),
"integer": lambda x: isinstance(x, int) and not isinstance(x, bool),
"number": lambda x: isinstance(x, (int, float)) and not isinstance(x, bool) and math.isfinite(x),
"boolean": lambda x: isinstance(x, bool),
}
if expected and (expected not in type_ok or not type_ok[expected](instance)):
return errors + [f"{location}: expected {expected}"]
if "const" in schema and instance != schema["const"]:
errors.append(f"{location}: expected constant {schema['const']!r}")
if "enum" in schema and instance not in schema["enum"]:
errors.append(f"{location}: value not in enum")
if isinstance(instance, dict):
required = schema.get("required", [])
errors.extend(f"{location}: missing {name}" for name in required if name not in instance)
properties = schema.get("properties", {})
if schema.get("additionalProperties") is False:
errors.extend(f"{location}: unknown property {name}" for name in instance if name not in properties)
for name, value in instance.items():
if name in properties:
errors.extend(validate(value, properties[name], schema_path, f"{location}.{name}"))
elif isinstance(instance, list):
if len(instance) < schema.get("minItems", 0):
errors.append(f"{location}: too few items")
if "maxItems" in schema and len(instance) > schema["maxItems"]:
errors.append(f"{location}: too many items")
if schema.get("uniqueItems") and len({canonical_bytes(item) for item in instance}) != len(instance):
errors.append(f"{location}: duplicate items")
for index, value in enumerate(instance):
errors.extend(validate(value, schema.get("items", {}), schema_path, f"{location}[{index}]"))
elif isinstance(instance, str):
if len(instance) < schema.get("minLength", 0):
errors.append(f"{location}: string too short")
if "maxLength" in schema and len(instance) > schema["maxLength"]:
errors.append(f"{location}: string too long")
if "pattern" in schema and re.fullmatch(schema["pattern"], instance) is None:
errors.append(f"{location}: pattern mismatch")
if schema.get("format") == "date-time":
try:
datetime.fromisoformat(instance.replace("Z", "+00:00"))
except ValueError:
errors.append(f"{location}: invalid date-time")
elif isinstance(instance, (int, float)) and not isinstance(instance, bool):
if "minimum" in schema and instance < schema["minimum"]:
errors.append(f"{location}: below minimum")
if "maximum" in schema and instance > schema["maximum"]:
errors.append(f"{location}: above maximum")
return errors
contracts/tests/events-v1/fixtures/sensitive-field.json
+17
View File
@@ -0,0 +1,17 @@
{
"schema_version": "yovision.event/v1",
"producer_id": "brain-school-a",
"source_event_id": "evt-sensitive-0001",
"site_ref": "site-school-a",
"device_ref": "camera-east-gate",
"profile_ref": "profile-main-stream",
"event_type": "dangerous_area_entered",
"occurred_at": "2026-08-31T00:00:01.125Z",
"severity": "high",
"rule": {"rule_id": "rule-east-danger", "version": "3"},
"model": {"name": "anonymous-detector", "version": "2026.08"},
"observation": {"track_id": "track-0042", "category": "person", "confidence": 0.93, "face_feature": "forbidden"},
"region": {"region_id": "region-east-danger", "kind": "area"},
"evidence": [],
"camera_password": "forbidden"
}
contracts/tests/events-v1/fixtures/unknown-version.json
+16
View File
@@ -0,0 +1,16 @@
{
"schema_version": "yovision.event/v2",
"producer_id": "brain-school-a",
"source_event_id": "evt-unknown-version-0001",
"site_ref": "site-school-a",
"device_ref": "camera-east-gate",
"profile_ref": "profile-main-stream",
"event_type": "dangerous_area_entered",
"occurred_at": "2026-08-31T00:00:01.125Z",
"severity": "high",
"rule": {"rule_id": "rule-east-danger", "version": "3"},
"model": {"name": "anonymous-detector", "version": "2026.08"},
"observation": {"track_id": "track-0042", "category": "person", "confidence": 0.93},
"region": {"region_id": "region-east-danger", "kind": "area"},
"evidence": []
}
contracts/tests/events-v1/idempotency-vectors.json
+13
View File
@@ -0,0 +1,13 @@
{
"algorithm": "RFC8785-JCS+SHA-256",
"vectors": [
{
"name": "dangerous-area-original-and-reordered-duplicate",
"fixture": "../../events/v1/examples/dangerous-area.json",
"idempotency_key": ["brain-school-a", "evt-area-20260831-0001"],
"payload_sha256": "4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1",
"conflict_patch": {"severity": "critical"},
"conflict_payload_sha256": "7076771f7827d97ef45831ae221046b8cb347f152dd222c2edd6f14a58e173b2"
}
]
}
contracts/tests/events-v1/test_contract.py
+110
View File
@@ -0,0 +1,110 @@
from __future__ import annotations
import copy
import hashlib
import sys
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
CONTRACTS = HERE.parents[1]
sys.path.insert(0, str(HERE))
from contract_support import canonical_bytes, load_json, validate # noqa: E402
class EventV1ContractTests(unittest.TestCase):
schema_path = CONTRACTS / "events" / "v1" / "event.schema.json"
schema = load_json(schema_path)
def assert_valid(self, payload: object) -> None:
self.assertEqual([], validate(payload, self.schema, self.schema_path))
def test_anonymous_area_and_line_examples_are_valid(self) -> None:
for name in ("dangerous-area.json", "directional-line-crossed.json"):
with self.subTest(name=name):
self.assert_valid(load_json(CONTRACTS / "events" / "v1" / "examples" / name))
def test_idempotency_vector_duplicate_and_conflict(self) -> None:
vectors = load_json(HERE / "idempotency-vectors.json")["vectors"]
for vector in vectors:
payload = load_json((HERE / vector["fixture"]).resolve())
self.assertEqual(vector["idempotency_key"], [payload["producer_id"], payload["source_event_id"]])
digest = hashlib.sha256(canonical_bytes(payload)).hexdigest()
self.assertEqual(vector["payload_sha256"], digest)
reordered = dict(reversed(list(payload.items())))
self.assertEqual(digest, hashlib.sha256(canonical_bytes(reordered)).hexdigest())
conflict = copy.deepcopy(payload)
conflict.update(vector["conflict_patch"])
conflict_digest = hashlib.sha256(canonical_bytes(conflict)).hexdigest()
self.assertEqual(vector["conflict_payload_sha256"], conflict_digest)
self.assertNotEqual(digest, conflict_digest)
def test_unknown_version_is_rejected(self) -> None:
payload = load_json(HERE / "fixtures" / "unknown-version.json")
self.assertTrue(validate(payload, self.schema, self.schema_path))
def test_brain_producer_sense_relay_and_bell_consumer_fixture(self) -> None:
produced = load_json(CONTRACTS / "events" / "v1" / "examples" / "dangerous-area.json")
self.assert_valid(produced)
relayed = copy.deepcopy(produced)
self.assertEqual(
(produced["producer_id"], produced["source_event_id"]),
(relayed["producer_id"], relayed["source_event_id"]),
)
self.assertEqual(canonical_bytes(produced), canonical_bytes(relayed))
bell_allowed = set(self.schema["properties"])
self.assertEqual(set(produced), bell_allowed)
self.assertNotIn("alert", produced)
self.assertNotIn("receipt", produced)
def test_sensitive_and_internal_fields_are_rejected(self) -> None:
payload = load_json(HERE / "fixtures" / "sensitive-field.json")
errors = validate(payload, self.schema, self.schema_path)
self.assertTrue(any("camera_password" in error for error in errors))
self.assertTrue(any("face_feature" in error for error in errors))
base = load_json(CONTRACTS / "events" / "v1" / "examples" / "dangerous-area.json")
for forbidden, value in {
"user_token": "forbidden", "ack_state": "acked", "local_path": "C:/forbidden",
"signed_url": "https://forbidden.invalid/object?signature=forbidden"
}.items():
with self.subTest(forbidden=forbidden):
candidate = copy.deepcopy(base)
candidate[forbidden] = value
self.assertTrue(validate(candidate, self.schema, self.schema_path))
def test_openapi_references_exist_and_responses_are_explicit(self) -> None:
path = CONTRACTS / "events" / "v1" / "openapi.json"
spec = load_json(path)
operation = spec["paths"]["/v1/events"]["post"]
self.assertEqual({"200", "201", "400", "409", "422"}, set(operation["responses"]))
refs: list[str] = []
def collect(value: object) -> None:
if isinstance(value, dict):
refs.extend(item for key, item in value.items() if key == "$ref")
for item in value.values(): collect(item)
elif isinstance(value, list):
for item in value: collect(item)
collect(spec)
self.assertTrue(refs)
for ref in refs:
self.assertTrue((path.parent / ref).resolve().is_file(), ref)
def test_duplicate_conflict_and_unknown_version_response_examples(self) -> None:
directory = CONTRACTS / "events" / "v1"
cases = (
("duplicate-result.json", "ingest-result.schema.json"),
("idempotency-conflict-problem.json", "problem.schema.json"),
("unsupported-version-problem.json", "problem.schema.json"),
)
for fixture_name, schema_name in cases:
with self.subTest(fixture=fixture_name):
schema_path = directory / schema_name
errors = validate(load_json(directory / "examples" / fixture_name), load_json(schema_path), schema_path)
self.assertEqual([], errors)
if __name__ == "__main__":
unittest.main(verbosity=2)
contracts/tests/evidence-v1/fixtures/sensitive-field.json
+12
View File
@@ -0,0 +1,12 @@
{
"schema_version": "yovision.evidence-reference/v1",
"evidence_id": "ev-sensitive-0001",
"owner_id": "sense-school-a",
"type": "snapshot",
"status": "available",
"captured_at": "2026-08-31T00:00:01.125Z",
"status_updated_at": "2026-08-31T00:00:02.450Z",
"content_type": "image/jpeg",
"integrity": {"algorithm": "sha256", "digest": "2f77668a9dfbf8d5848b9e6d7da867800b7b6790625f16b45a101f1ca1f7da75", "size_bytes": 48215},
"local_path": "forbidden"
}
contracts/tests/evidence-v1/test_contract.py
+81
View File
@@ -0,0 +1,81 @@
from __future__ import annotations
import copy
import sys
import unittest
from pathlib import Path
HERE = Path(__file__).resolve().parent
CONTRACTS = HERE.parents[1]
EVENT_SUPPORT = HERE.parent / "events-v1"
sys.path.insert(0, str(EVENT_SUPPORT))
from contract_support import load_json, validate # noqa: E402
class EvidenceV1ContractTests(unittest.TestCase):
schema_path = CONTRACTS / "evidence" / "v1" / "evidence-reference.schema.json"
schema = load_json(schema_path)
def errors_for(self, payload: object) -> list[str]:
return validate(payload, self.schema, self.schema_path)
def test_pending_success_and_failed_examples_are_valid(self) -> None:
for name in ("pending.json", "success.json", "failed.json"):
with self.subTest(name=name):
payload = load_json(CONTRACTS / "evidence" / "v1" / "examples" / name)
self.assertEqual([], self.errors_for(payload))
def test_state_specific_metadata_is_enforced(self) -> None:
available = load_json(CONTRACTS / "evidence" / "v1" / "examples" / "success.json")
del available["integrity"]
self.assertTrue(self.errors_for(available))
failed = load_json(CONTRACTS / "evidence" / "v1" / "examples" / "failed.json")
del failed["failure"]
self.assertTrue(self.errors_for(failed))
pending = load_json(CONTRACTS / "evidence" / "v1" / "examples" / "pending.json")
pending["content_type"] = "image/jpeg"
self.assertTrue(self.errors_for(pending))
def test_sensitive_access_material_and_unknown_version_are_rejected(self) -> None:
fixture = load_json(HERE / "fixtures" / "sensitive-field.json")
self.assertTrue(any("local_path" in error for error in self.errors_for(fixture)))
base = load_json(CONTRACTS / "evidence" / "v1" / "examples" / "pending.json")
for forbidden, value in {
"camera_password": "forbidden",
"user_token": "forbidden",
"signed_url": "https://forbidden.invalid/object?signature=forbidden",
"face_feature": "forbidden",
"alert_state": "acked"
}.items():
with self.subTest(forbidden=forbidden):
candidate = copy.deepcopy(base)
candidate[forbidden] = value
self.assertTrue(self.errors_for(candidate))
unknown = copy.deepcopy(base)
unknown["schema_version"] = "yovision.evidence-reference/v2"
self.assertTrue(self.errors_for(unknown))
def test_openapi_refs_and_degradation_responses(self) -> None:
path = CONTRACTS / "evidence" / "v1" / "openapi.json"
spec = load_json(path)
operation = spec["paths"]["/v1/evidence/{evidence_id}"]["get"]
self.assertEqual({"200", "404", "410"}, set(operation["responses"]))
refs: list[str] = []
def collect(value: object) -> None:
if isinstance(value, dict):
refs.extend(item for key, item in value.items() if key == "$ref")
for item in value.values(): collect(item)
elif isinstance(value, list):
for item in value: collect(item)
collect(spec)
for ref in refs:
self.assertTrue((path.parent / ref).resolve().is_file(), ref)
if __name__ == "__main__":
unittest.main(verbosity=2)