71 lines
3.4 KiB
Python
71 lines
3.4 KiB
Python
"""Cross-language invariants for Brain-to-Bell event ingress v1."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
CONTRACT = ROOT / "docs" / "contracts" / "brain-event-ingress-v1.openapi.json"
|
|
EVENT_SCHEMA = ROOT / "docs" / "raw" / "contracts" / "event-v0.1.schema.json"
|
|
|
|
|
|
class BrainEventIngressContractTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.document = json.loads(CONTRACT.read_text(encoding="utf-8"))
|
|
cls.event_schema = json.loads(EVENT_SCHEMA.read_text(encoding="utf-8"))
|
|
|
|
def test_candidate_is_exact_final_top_level_shape_minus_bell_id(self) -> None:
|
|
candidate = self.document["components"]["schemas"]["EventV01Candidate"]
|
|
expected = set(self.event_schema["required"]) - {"id"}
|
|
self.assertEqual(expected, set(candidate["required"]))
|
|
self.assertEqual(expected, set(candidate["properties"]))
|
|
self.assertFalse(candidate["additionalProperties"])
|
|
self.assertIn("id", self.event_schema["required"])
|
|
|
|
def test_transport_auth_and_statuses_are_frozen(self) -> None:
|
|
operation = self.document["paths"]["/internal/v1/event-candidates"]["post"]
|
|
self.assertEqual({"200", "201", "400", "401", "403", "409", "413", "422", "503"}, set(operation["responses"]))
|
|
header_names = {
|
|
self.document["components"]["parameters"][parameter["$ref"].split("/")[-1]]["name"]
|
|
for parameter in operation["parameters"]
|
|
}
|
|
self.assertEqual(
|
|
{"X-YoVision-Key-Id", "X-YoVision-Timestamp", "X-YoVision-Nonce", "X-YoVision-Signature"},
|
|
header_names,
|
|
)
|
|
|
|
def test_numeric_safety_bounds_match_both_implementations(self) -> None:
|
|
signature = self.document["x-yovision-signature"]
|
|
delivery = self.document["x-yovision-delivery"]
|
|
self.assertEqual((300, 600, 32), (signature["clock_skew_seconds"], signature["nonce_receipt_ttl_seconds"], signature["minimum_secret_bytes"]))
|
|
self.assertEqual((1048576, 10, 1, 300, 100, 10000), (
|
|
delivery["maximum_body_bytes"], delivery["deadline_seconds"], delivery["initial_retry_seconds"],
|
|
delivery["maximum_retry_seconds"], delivery["maximum_attempts"], delivery["maximum_queued"],
|
|
))
|
|
brain = (ROOT / "Brain" / "yovision_brain" / "ingress.py").read_text(encoding="utf-8")
|
|
bell = (ROOT / "Bell" / "internal" / "ingress" / "ingress.go").read_text(encoding="utf-8")
|
|
for marker in (
|
|
"/internal/v1/event-candidates", "X-YoVision-Key-Id", "X-YoVision-Timestamp",
|
|
"X-YoVision-Nonce", "X-YoVision-Signature",
|
|
):
|
|
self.assertIn(marker, brain)
|
|
self.assertIn(marker, bell)
|
|
self.assertIn("MAX_QUEUED = 10_000", brain)
|
|
self.assertIn("MAX_ATTEMPTS = 100", brain)
|
|
self.assertIn("300*time.Second", bell)
|
|
|
|
def test_event_key_is_bound_to_producer_and_separate_from_audit(self) -> None:
|
|
key_shape = self.document["x-yovision-signature"]["key_document_shape"]["keys"][0]
|
|
self.assertEqual({"key_id", "producer_id", "secret_base64url"}, set(key_shape))
|
|
bell_main = (ROOT / "Bell" / "cmd" / "bell-api" / "main.go").read_text(encoding="utf-8")
|
|
self.assertIn("BELL_EVENT_INGRESS_KEYS_FILE", bell_main)
|
|
self.assertIn("BELL_AUDIT_KEYS_FILE", bell_main)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|