""" server-26#300 — confidence gate before any push alert. CrimeRadar pushed "shots fired at an elementary school" after STT heard "fire alarm" as "firearm". A single uncorroborated transcript hit on a high-severity term must be held, not pushed; a corroborated incident releases it; low-confidence matches never push. """ import pytest from unittest.mock import AsyncMock, patch from app.internal import alerter from tests.fake_firestore import FakeFirestore def _ev(**over): base = {"incident_ids": [], "max_calls": 0, "severity": None, "stt_confidence": None, "not_speech": False} base.update(over) return base # ── the pure decision ─────────────────────────────────────────────────────── def test_talkgroup_only_rule_always_pushes(): v = alerter._assess(["talkgroup:9048"], [], _ev(stt_confidence=0.1)) assert v["push"] and v["confidence"] == 1.0 def test_transcript_only_firearm_single_call_is_held(): v = alerter._assess(["firearm"], [], _ev()) assert not v["push"] and v["hold_reason"] == "needs_corroboration" assert v["high_severity"] is True def test_tagged_shots_fired_still_needs_corroboration(): v = alerter._assess(["shots-fired"], ["shots-fired"], _ev(max_calls=1)) assert not v["push"] and v["confidence"] == 0.8 def test_corroborated_high_severity_pushes(): v = alerter._assess(["shots-fired"], ["shots-fired"], _ev(max_calls=2)) assert v["push"] def test_major_incident_makes_any_content_match_high_severity(): v = alerter._assess(["school"], [], _ev(severity="major")) assert v["high_severity"] and not v["push"] def test_ordinary_tag_match_pushes(): v = alerter._assess(["mva"], ["mva"], _ev(max_calls=1)) assert v["push"] and not v["high_severity"] def test_low_stt_confidence_holds_even_ordinary_matches(): v = alerter._assess(["mva"], ["mva"], _ev(stt_confidence=0.1)) assert not v["push"] and v["hold_reason"] == "low_confidence" assert v["confidence"] == round(0.8 * 0.1 / 0.3, 3) def test_typical_radio_logprob_does_not_hold_a_transcript_match(): """exp(avg_logprob) of ~0.4 is normal on vocoded audio (review of #300).""" v = alerter._assess(["mva"], [], _ev(stt_confidence=0.4)) assert v["push"] def test_not_speech_never_pushes_content(): v = alerter._assess(["mva"], ["mva"], _ev(not_speech=True)) assert not v["push"] # ── end to end through Firestore ──────────────────────────────────────────── RULE = {"rule_id": "r1", "name": "Weapons", "enabled": True, "org_id": "org-1", "keywords": ["firearm", "shots-fired"], "discord_webhook": "https://discord.example/wh"} @pytest.fixture def fs(monkeypatch): fake = FakeFirestore.install(monkeypatch) fake.put("alert_rules", "r1", dict(RULE)) return fake def _posts(): posted = [] class _Client: async def __aenter__(self): return self async def __aexit__(self, *a): return False async def post(self, url, json=None, **k): posted.append(json) return posted, patch("httpx.AsyncClient", lambda *a, **k: _Client()) async def _dispatch(call_id, tags, transcript): await alerter.check_and_dispatch(call_id=call_id, node_id="n1", talkgroup_id=1, talkgroup_name="PD Dispatch", tags=tags, transcript=transcript) @pytest.mark.asyncio async def test_crimeradar_case_is_held_then_released_by_corroboration(fs): fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"]}) fs.put("incidents", "inc-1", {"call_ids": ["c1"], "severity": "moderate"}) posted, http = _posts() with http: await _dispatch("c1", [], "report of a firearm at the elementary school") assert posted == [], "a single uncorroborated firearm hit must not push" assert fs.data.get("alert_events", {}) == {} (hold,) = fs.data["alert_holds"].values() assert hold["status"] == "held" and hold["hold_reason"] == "needs_corroboration" assert hold["incident_id"] == "inc-1" # A second call joins the incident; it does not itself mention a firearm. fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"], "severity": "moderate"}) with http: await _dispatch("c2", [], "units on scene, stand by") assert len(posted) == 1 and "Weapons" in posted[0]["content"] (event,) = fs.data["alert_events"].values() assert event["released_from_hold"] is True and event["corroborated_by_call_id"] == "c2" assert fs.get("alert_holds", hold["alert_id"])["status"] == "released" @pytest.mark.asyncio async def test_low_confidence_hold_is_not_released_by_corroboration(fs): fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"], "stt_confidence": 0.1}) fs.put("incidents", "inc-1", {"call_ids": ["c1"]}) posted, http = _posts() with http: await _dispatch("c1", [], "firearm") (hold,) = fs.data["alert_holds"].values() assert hold["hold_reason"] == "low_confidence" fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"]}) with http: await _dispatch("c2", [], "units on scene") assert posted == [] assert fs.get("alert_holds", hold["alert_id"])["status"] == "held" @pytest.mark.asyncio async def test_rule_pushed_now_supersedes_its_hold_instead_of_alerting_twice(fs): fs.put("calls", "c1", {"org_id": "org-1", "incident_ids": ["inc-1"]}) fs.put("incidents", "inc-1", {"call_ids": ["c1"]}) posted, http = _posts() with http: await _dispatch("c1", ["shots-fired"], "shots fired") fs.put("calls", "c2", {"org_id": "org-1", "incident_ids": ["inc-1"]}) fs.put("incidents", "inc-1", {"call_ids": ["c1", "c2"]}) with http: await _dispatch("c2", ["shots-fired"], "more shots fired") assert len(posted) == 1 (hold,) = fs.data["alert_holds"].values() assert hold["status"] == "superseded" @pytest.mark.asyncio async def test_held_alerts_are_listed_for_admins(fs): from app.routers import alerts as alerts_router fs.put("alert_holds", "h1", {"alert_id": "h1", "status": "held", "triggered_at": "2026-10-27T01:00:00"}) fs.put("alert_holds", "h2", {"alert_id": "h2", "status": "released", "triggered_at": "2026-10-27T02:00:00"}) out = await alerts_router.list_held_alerts(status="held", org_id=None, _={}) assert [h["alert_id"] for h in out] == ["h1"]