149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from yovision_brain.events import INTERNAL_EVENT_SCHEMA, InternalEventCandidate
|
|
from yovision_brain.integration.event_export import EventExportError, canonical_json, export_event, payload_sha256
|
|
|
|
ROOT = Path(__file__).resolve().parents[4]
|
|
|
|
|
|
def _candidate(**changes: object) -> InternalEventCandidate:
|
|
values: dict[str, object] = {
|
|
"schema": INTERNAL_EVENT_SCHEMA,
|
|
"event_id": "evt-area-20260831-0001",
|
|
"logical_input_id": "camera-east-gate",
|
|
"event_type": "danger_area_entered",
|
|
"occurred_at_ns": 1_788_134_401_125_000_000,
|
|
"rule_id": "rule-east-danger",
|
|
"rule_version": "3",
|
|
"model_name": "anonymous-detector",
|
|
"model_version": "2026.08",
|
|
"profile_id": "profile-main-stream",
|
|
"frame_width": 100,
|
|
"frame_height": 100,
|
|
"track_id": "track-0042",
|
|
"observation": {
|
|
"category": "anonymous_target",
|
|
"confidence": 0.93,
|
|
"box": {"left": 12, "top": 20, "right": 31, "bottom": 74},
|
|
"anchor": {"x": 0.21, "y": 0.74},
|
|
},
|
|
"reason": "entered polygon",
|
|
}
|
|
values.update(changes)
|
|
return InternalEventCandidate(**values) # type: ignore[arg-type]
|
|
|
|
|
|
def _fixture(relative: str) -> dict[str, object]:
|
|
return json.loads((ROOT / relative).read_text(encoding="utf-8"))
|
|
|
|
|
|
def test_brain_mapper_matches_frozen_producer_fixture_and_jcs_digest() -> None:
|
|
pending = _fixture("contracts/evidence/v1/examples/pending.json")
|
|
expected = _fixture("contracts/events/v1/examples/dangerous-area.json")
|
|
event = export_event(
|
|
_candidate(),
|
|
producer_id="brain-school-a",
|
|
site_ref="site-school-a",
|
|
severity="high",
|
|
evidence=[pending],
|
|
region_ref="region-east-danger",
|
|
)
|
|
|
|
assert event == expected
|
|
assert payload_sha256(event) == "4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1"
|
|
|
|
|
|
def test_retry_mapping_preserves_original_identity_and_payload() -> None:
|
|
candidate = _candidate()
|
|
arguments = {
|
|
"producer_id": "brain-school-a",
|
|
"site_ref": "site-school-a",
|
|
"severity": "high",
|
|
"evidence": [_fixture("contracts/evidence/v1/examples/pending.json")],
|
|
"region_ref": "region-east-danger",
|
|
}
|
|
first = export_event(candidate, **arguments)
|
|
retry = export_event(candidate, **arguments)
|
|
|
|
assert (first["producer_id"], first["source_event_id"]) == (
|
|
"brain-school-a",
|
|
candidate.event_id,
|
|
)
|
|
assert retry == first
|
|
assert payload_sha256(retry) == payload_sha256(first)
|
|
|
|
|
|
@pytest.mark.parametrize("fixture", ["pending.json", "failed.json"])
|
|
def test_evidence_degradation_states_are_exported_unchanged(fixture: str) -> None:
|
|
evidence = _fixture(f"contracts/evidence/v1/examples/{fixture}")
|
|
candidate = _candidate()
|
|
if fixture == "failed.json":
|
|
candidate = _candidate(
|
|
event_id="evt-line-20260831-0002",
|
|
logical_input_id="camera-north-corridor",
|
|
event_type="directional_line_crossed",
|
|
occurred_at_ns=1_788_134_590_000_000_000,
|
|
rule_id="rule-north-one-way",
|
|
rule_version="1",
|
|
track_id="track-0088",
|
|
observation={"category": "anonymous_target", "confidence": 0.88},
|
|
)
|
|
event = export_event(
|
|
candidate,
|
|
producer_id="brain-school-a",
|
|
site_ref="site-school-a",
|
|
severity="medium" if fixture == "failed.json" else "high",
|
|
evidence=[evidence],
|
|
region_ref="line-north-one-way" if fixture == "failed.json" else "region-east-danger",
|
|
crossing_direction="b_to_a" if fixture == "failed.json" else None,
|
|
)
|
|
|
|
assert event["evidence"] == [evidence]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"unsafe",
|
|
[
|
|
{"local_path": "D:/captures/frame.jpg"},
|
|
{"url": "https://example.invalid/signed"},
|
|
{"token": "not-a-real-token"},
|
|
],
|
|
)
|
|
def test_evidence_rejects_paths_urls_and_sensitive_fields(unsafe: dict[str, object]) -> None:
|
|
evidence = _fixture("contracts/evidence/v1/examples/pending.json")
|
|
evidence.update(unsafe)
|
|
with pytest.raises(EventExportError, match="forbidden"):
|
|
export_event(
|
|
_candidate(),
|
|
producer_id="brain-school-a",
|
|
site_ref="site-school-a",
|
|
severity="high",
|
|
evidence=[evidence],
|
|
region_ref="region-east-danger",
|
|
)
|
|
|
|
|
|
def test_failed_evidence_cannot_claim_success_content() -> None:
|
|
evidence = _fixture("contracts/evidence/v1/examples/failed.json")
|
|
evidence["content_type"] = "video/mp4"
|
|
with pytest.raises(EventExportError, match="failed evidence"):
|
|
export_event(
|
|
_candidate(),
|
|
producer_id="brain-school-a",
|
|
site_ref="site-school-a",
|
|
severity="high",
|
|
evidence=[evidence],
|
|
region_ref="region-east-danger",
|
|
)
|
|
|
|
|
|
def test_jcs_normalizes_number_lexemes_and_negative_zero() -> None:
|
|
assert canonical_json({"small": 1e-7, "fixed": 1e20, "zero": -0.0}) == (
|
|
'{"fixed":100000000000000000000,"small":1e-7,"zero":0}'
|
|
)
|