70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
import json
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.modules.dwell_engine import DwellEngine
|
|
from app.modules.notifier import append_json_event, dispatch_json_event
|
|
|
|
|
|
def test_append_json_event_writes_jsonl(tmp_path):
|
|
output = tmp_path / "logs" / "events.jsonl"
|
|
|
|
append_json_event(output, {"event": "long_stay_alert", "count": 5})
|
|
append_json_event(output, {"event": "half_hour_report", "count": 3})
|
|
|
|
lines = output.read_text(encoding="utf-8").splitlines()
|
|
|
|
assert json.loads(lines[0]) == {"event": "long_stay_alert", "count": 5}
|
|
assert json.loads(lines[1]) == {"event": "half_hour_report", "count": 3}
|
|
|
|
|
|
def test_dispatch_json_event_posts_report_without_tracks(tmp_path, monkeypatch):
|
|
tz = ZoneInfo("Asia/Shanghai")
|
|
window_start = datetime(2026, 4, 15, 11, 7, tzinfo=tz)
|
|
engine = DwellEngine(
|
|
camera_id="store_cam_01",
|
|
queue_time_threshold_seconds=300,
|
|
crowded_count_threshold=5,
|
|
normal_count_threshold=2,
|
|
pause_timeout_seconds=300,
|
|
alert_cooldown_seconds=600,
|
|
report_window_start=window_start,
|
|
)
|
|
|
|
engine.process_observations(
|
|
[{"person_id": f"cust_{idx}", "role": "customer"} for idx in range(6)],
|
|
window_start,
|
|
)
|
|
events = engine.process_observations([], window_start.replace(minute=37))
|
|
report_event = next(event for event in events if event["event"] == "half_hour_report")
|
|
|
|
sent: dict[str, object] = {}
|
|
|
|
class DummyResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def fake_urlopen(req, timeout):
|
|
sent["url"] = req.full_url
|
|
sent["timeout"] = timeout
|
|
sent["payload"] = json.loads(req.data.decode("utf-8"))
|
|
return DummyResponse()
|
|
|
|
monkeypatch.setattr("app.modules.notifier.request.urlopen", fake_urlopen)
|
|
|
|
output = tmp_path / "logs" / "events.jsonl"
|
|
dispatch_json_event(
|
|
output,
|
|
report_event,
|
|
webhook_url="https://example.test/webhook",
|
|
timeout_seconds=5.0,
|
|
)
|
|
|
|
assert sent["url"] == "https://example.test/webhook"
|
|
assert sent["timeout"] == 5.0
|
|
assert sent["payload"]["event"] == "half_hour_report"
|
|
assert "tracks" not in sent["payload"]
|