111 lines
5.1 KiB
Python
111 lines
5.1 KiB
Python
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)
|