import json from datetime import datetime, timezone import numpy as np import pytest from v1.alerts import ( AlertDispatcher, AlertSink, EventArtifactWriter, annotate_frame, screenshot_label, ) from v1.fall_state import FallEvent, FallState from v1.view_model import MonitorViewState, PersonOverlay, Point, StatusColor class RecordingSink(AlertSink): def __init__(self): self.sounds = [] self.popups = [] def play_sound(self, event): self.sounds.append(event.event_id) def show_popup(self, record): self.popups.append(record.event.event_id) def _event(event_id="FALL-000001", track_id="P-0003"): return FallEvent( event_id=event_id, track_id=track_id, config_version="cfg-abc123", suspected_at_monotonic=0.0, confirmed_at_monotonic=1.8, latency_seconds=1.8, ) def _confirmed_view(): overlay = PersonOverlay( track_id="P-0003", state=FallState.CONFIRMED, color=StatusColor.CRITICAL, box_xyxy=(20.0, 20.0, 80.0, 180.0), keypoints=(Point(30.0, 40.0), Point(40.0, 40.0)), skeleton_segments=((Point(30.0, 40.0), Point(40.0, 40.0)),), label="P-0003 · CONFIRMED", ) return MonitorViewState( connected=True, status_text="在线", status_color=StatusColor.SUCCESS, has_frame=True, people=(overlay,), events=(), highest_state=FallState.CONFIRMED, ) def _writer(tmp_path): fixed = datetime(2026, 7, 21, 8, 30, 0, tzinfo=timezone.utc) return EventArtifactWriter(tmp_path, source_id="lobby-camera-01", clock=lambda: fixed) def _image(): return np.full((200, 240, 3), 30, dtype=np.uint8) def test_confirmed_event_fires_each_side_effect_once(tmp_path): sink = RecordingSink() dispatcher = AlertDispatcher(_writer(tmp_path), sink) event = _event() first = dispatcher.dispatch([event], _image(), _confirmed_view()) second = dispatcher.dispatch([event], _image(), _confirmed_view()) assert len(first) == 1 assert second == () assert sink.sounds == ["FALL-000001"] assert sink.popups == ["FALL-000001"] log_path = tmp_path / "20260721" / "events.jsonl" assert log_path.read_text(encoding="utf-8").count("\n") == 1 assert (tmp_path / "20260721" / "FALL-000001.png").is_file() def test_jsonl_record_has_fields_and_no_source_address(tmp_path): dispatcher = AlertDispatcher(_writer(tmp_path)) dispatcher.dispatch([_event()], _image(), _confirmed_view()) line = (tmp_path / "20260721" / "events.jsonl").read_text(encoding="utf-8").strip() record = json.loads(line) assert record["event_id"] == "FALL-000001" assert record["config_version"] == "cfg-abc123" assert record["source_id"] == "lobby-camera-01" assert record["state"] == "CONFIRMED" assert record["confirmed_at_utc"] == "2026-07-21T08:30:00+00:00" assert record["screenshot"] == "20260721/FALL-000001.png" assert "rtsp" not in line.lower() assert "password" not in line.lower() def test_missing_image_produces_no_side_effect(tmp_path): sink = RecordingSink() dispatcher = AlertDispatcher(_writer(tmp_path), sink) records = dispatcher.dispatch([_event()], None, _confirmed_view()) assert records == () assert sink.sounds == [] assert not (tmp_path / "20260721").exists() def test_distinct_events_produce_two_log_lines(tmp_path): dispatcher = AlertDispatcher(_writer(tmp_path)) dispatcher.dispatch([_event("FALL-000001")], _image(), _confirmed_view()) dispatcher.dispatch([_event("FALL-000002", "P-0009")], _image(), _confirmed_view()) lines = (tmp_path / "20260721" / "events.jsonl").read_text(encoding="utf-8").strip().split("\n") assert len(lines) == 2 def test_screenshot_label_is_ascii_only(): label = screenshot_label("P-0001", "CONFIRMED") assert label == "P-0001 CONFIRMED" assert all(ord(ch) < 128 for ch in label) def test_screenshot_label_strips_non_ascii_separator(): # The live-view label uses a middle dot; the screenshot label must not. label = screenshot_label("P-0001", "CONFIRMED") assert "·" not in label def test_annotate_draws_red_border_only_for_confirmed(): image = np.full((60, 80, 3), 30, dtype=np.uint8) confirmed = annotate_frame(image, _confirmed_view()) corner = confirmed[0, 0].tolist() assert corner == [40, 40, 198] # critical red in BGR assert confirmed.shape == image.shape