31 lines
1.7 KiB
Python
31 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
|
|
from yovision_brain.events import JsonLinesSink, candidate_from_decision
|
|
from yovision_brain.rules import NormalizedPoint, RuleDecision
|
|
from yovision_brain.vision import BoundingBox, DetectorMetadata, TrackedObject
|
|
|
|
|
|
def test_internal_event_id_is_stable_and_payload_is_safe() -> None:
|
|
track = TrackedObject("track-000001", BoundingBox(1, 2, 3, 4), "anonymous_target", 1.0, 7, 123)
|
|
decision = RuleDecision("yard", "danger_area", track.track_id, "entered", True, "entered polygon", "rules-v1", "main", 10, 10, NormalizedPoint(0.2, 0.4), 123)
|
|
metadata = DetectorMetadata("detector", "1", "first-party", "no external weights", "none")
|
|
first = candidate_from_decision(decision, track, logical_input_id="camera-01", detector=metadata)
|
|
second = candidate_from_decision(decision, track, logical_input_id="camera-01", detector=metadata)
|
|
assert first == second
|
|
assert first.event_id.startswith("brain-local-")
|
|
payload = json.dumps(first.to_dict())
|
|
for forbidden in ("password", "rtsp://", "evidence", "face"):
|
|
assert forbidden not in payload.lower()
|
|
|
|
|
|
def test_json_lines_sink_writes_one_canonical_line() -> None:
|
|
track = TrackedObject("track-000001", BoundingBox(0, 0, 1, 1), "anonymous_target", 1.0, 0, 0)
|
|
decision = RuleDecision("yard", "danger_area", track.track_id, "entered", True, "entered", "v1", "main", 2, 2, NormalizedPoint(0.25, 0.5), 0)
|
|
candidate = candidate_from_decision(decision, track, logical_input_id="synthetic", detector=DetectorMetadata("d", "1", "first", "none", "none"))
|
|
stream = io.StringIO()
|
|
JsonLinesSink(stream).write(candidate)
|
|
assert json.loads(stream.getvalue())["schema"] == "brain.internal.event-candidate/v1"
|