""" An incident must not lie about what it is or where it is — server-26#23 / #26. Both defects come from the 2026-08-20 production dump (CORRELATION_REVIEW_0820.md) and both are the same shape: a field of the incident header re-derived from whichever call linked most recently. * `location` (the label) and `location_coords` (the map pin) were two independent last-write-wins fields. A call could move one and not the other, so they drifted: 5 of 6 incidents in the dump were pinned somewhere other than the place they were labelled. `b9b4f392` said "100 South Mosher" and pinned `Westmed`. * `location` was never validated, so "Flames from 49" put `location: "49"` on `9d376ffe` and the summarizer wrote "reported at location 49". * `title` was re-derived from every classified call, so `b9b4f392` — opened on a suspect search at 80 Grasslands Road — was named after its third call, and `f5190670` after the thirteenth of its thirteen events. The rules under test: 1. label and pin are ONE value, written together on every path; 2. a pin is only ever kept next to the label it was geocoded from; 3. a string with no word in it is not a place; 4. the title names the founding event and only ever escalates. """ import pytest from datetime import datetime, timedelta, timezone from unittest.mock import AsyncMock, patch from app.internal.incident_correlator import ( _build_context, _create_incident, _update_incident, _resolve_location_pair, _verified_pin, clean_location, location_is_unit, ) NOW = datetime(2026, 8, 20, 7, 25, 0, tzinfo=timezone.utc) # TG 383 from the dump: "Ch 1 (Patched with 155.310)" — a shared dispatch # backbone, which is where every one of these chains happened. DISPATCH_TG = "Ch 1 (Patched with 155.310)" GRASSLANDS = {"lat": 41.0891, "lng": -73.8010} WESTMED = {"lat": 41.0348, "lng": -73.7629} # --------------------------------------------------------------------------- # Harness — incidents are stored with merge=True, so folding each write back # into the dict is exactly what Firestore does between calls. # --------------------------------------------------------------------------- async def _create(**call) -> dict: with patch("app.internal.incident_correlator.fstore") as mock_fstore: mock_fstore.doc_set = AsyncMock() await _create_incident( call.get("call_id", "call-0"), "org-1", call.get("incident_type", "police"), 383, DISPATCH_TG, "sys-1", call.get("tags", []), call.get("location"), call.get("coords"), call.get("units", []), [], None, call.get("severity", "routine"), call.get("now", NOW), ) return dict(mock_fstore.doc_set.await_args.args[2]) async def _link(inc: dict, **call) -> dict: """Run one call through _update_incident, fold the write back, return it.""" with patch("app.internal.incident_correlator.fstore") as mock_fstore: mock_fstore.doc_set = AsyncMock() await _update_incident( inc, call.get("call_id", "call-n"), 383, "sys-1", call.get("tags", []), call.get("location"), call.get("coords"), call.get("units", []), [], None, call.get("now", NOW), talkgroup_name=DISPATCH_TG, incident_type=call.get("incident_type"), call_severity=call.get("severity", "routine"), ) updates = dict(mock_fstore.doc_set.await_args.args[2]) inc.update(updates) return updates def _assert_pin_matches_label(doc: dict): """The invariant: a pin exists only alongside the label it was geocoded from.""" if doc.get("location_coords") is not None: assert doc.get("location"), "pin with no label" assert doc.get("location_coords_source") == doc["location"], ( f"pin sourced from {doc.get('location_coords_source')!r} " f"but incident is labelled {doc['location']!r}" ) # --------------------------------------------------------------------------- # server-26#23 — the label and the pin are one value # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_label_and_pin_stay_consistent_across_a_chain_of_calls(): """ Replays `b9b4f392` exactly: a suspect search at 80 Grasslands Road, then an EMS transport to Westmed, then a brand-new open-911 dispatch at 100 South Mosher. Production ended up labelled "100 South Mosher" and pinned at Westmed — the label from one call, the pin from another. """ inc = await _create( tags=["suspect-search"], location="80 Grasslands Road", coords=GRASSLANDS, severity="moderate", incident_type="police", ) _assert_pin_matches_label(inc) await _link( # 07:41 — "one female to Westmed" inc, call_id="call-ems", tags=["ems-transport"], location="Westmed", coords=WESTMED, incident_type="ems", now=NOW + timedelta(minutes=16), ) _assert_pin_matches_label(inc) await _link( # 07:55 — "open 911 line", a different job entirely inc, call_id="call-911", tags=["open-911"], location="100 South Mosher", coords=None, incident_type="police", now=NOW + timedelta(minutes=30), ) _assert_pin_matches_label(inc) assert inc["location"] == "80 Grasslands Road" assert inc["location_coords"] == GRASSLANDS # Every place anyone named is still recorded — that is what the map path # is drawn from; it just isn't the incident's own location. assert inc["location_mentions"] == [ "80 Grasslands Road", "Westmed", "100 South Mosher", ] @pytest.mark.asyncio async def test_a_later_call_never_moves_the_pin_without_the_label(): """The direct mechanism: coords updating on their own.""" inc = await _create(tags=["suspect-search"], location="80 Grasslands Road", coords=None, incident_type="police") assert inc["location_coords"] is None updates = await _link(inc, tags=["ems-transport"], location="Westmed", coords=WESTMED, incident_type="ems") assert updates["location"] == "80 Grasslands Road" assert updates["location_coords"] is None _assert_pin_matches_label(inc) @pytest.mark.asyncio async def test_a_pin_can_still_be_filled_in_for_the_same_place(): """ The one permitted change. Geocoding is not deterministic in practice — it needs the node's position, an API quota and a response — so the same address can fail on one call and resolve on the next. Filling in a pin the incident never had is not a move; matching is deliberately by exact label, so it can never quietly re-point at a different street. """ inc = await _create(tags=["welfare-check"], location="55 Hyman Hills Road", coords=None, incident_type="police") assert inc["location_coords"] is None await _link(inc, tags=["welfare-check"], location="55 Hyman Hills Road", coords=GRASSLANDS, incident_type="police") assert inc["location"] == "55 Hyman Hills Road" assert inc["location_coords"] == GRASSLANDS _assert_pin_matches_label(inc) def test_a_pin_that_cannot_be_tied_to_the_label_is_not_shown(): """ Every incident written before this change carries a pin with no record of where it came from — and the dump says 5 of 6 of those are wrong. An unverifiable pin is dropped, not displayed: a missing pin reads as missing data, a wrong one reads as fact. """ legacy = {"location": "100 South Mosher", "location_coords": WESTMED} assert _verified_pin(legacy) is None resolved = _resolve_location_pair(legacy, None, None) assert resolved["location"] == "100 South Mosher" assert resolved["location_coords"] is None assert resolved["location_coords_source"] is None tagged = {"location": "Westmed", "location_coords": WESTMED, "location_coords_source": "westmed"} assert _verified_pin(tagged) == WESTMED # --------------------------------------------------------------------------- # server-26#23 — "49" is not a place # --------------------------------------------------------------------------- @pytest.mark.parametrize("junk", [ "49", # 9d376ffe, from "Fire received. Flames from 49." "10-24", # a ten-code "5-5-2", # a unit designator " ", "", None, "1", ]) def test_bare_numbers_are_rejected_as_locations(junk): assert clean_location(junk) is None @pytest.mark.parametrize("place", [ "80 Grasslands Road", "Westmed", "Rt 9", "226 East Main Street, apartment number 1", ]) def test_real_place_names_survive(place): assert clean_location(place) == place @pytest.mark.asyncio async def test_a_bare_number_never_reaches_the_correlator(): """ Rejected at the context boundary, so it is not a location in the fit tests, the thin-call test, the LLM prompt or the incident — and its coordinates go with it, because they were geocoded from that very string. """ with patch("app.internal.incident_correlator.fstore") as mock_fstore: mock_fstore.doc_get = AsyncMock(return_value={}) mock_fstore.collection_list = AsyncMock(return_value=[]) ctx = await _build_context( call_id="call-49", units=None, vehicles=None, cleared_units=None, location_coords={"lat": 41.0, "lng": -73.8}, reference_time=NOW, system_id="sys-1", talkgroup_id=383, talkgroup_name=DISPATCH_TG, tags=[], incident_type="fire", location="49", reassignment=False, create_if_new=True, ) assert ctx["location"] is None assert ctx["location_coords"] is None @pytest.mark.asyncio async def test_a_bare_number_never_becomes_an_incident_location_or_title(): inc = await _create(tags=["flames"], location="49", coords=None, incident_type="fire") assert inc["location"] is None assert inc["location_coords"] is None assert "49" not in inc["title"] assert inc["location_mentions"] == [] # --------------------------------------------------------------------------- # server-26#26 — the title names the founding event # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_an_unrelated_later_call_does_not_rename_the_incident(): """`b9b4f392` again, from the title's side.""" inc = await _create(tags=["suspect-search"], location="80 Grasslands Road", coords=GRASSLANDS, severity="moderate", incident_type="police") assert inc["title"] == "Suspect Search at 80 Grasslands Road" await _link(inc, tags=["ems-transport"], location="Westmed", coords=WESTMED, incident_type="ems", severity="routine") await _link(inc, tags=["open-911"], location="100 South Mosher", incident_type="police", severity="moderate") assert inc["title"] == "Suspect Search at 80 Grasslands Road" assert inc["title_tag"] == "Suspect Search" @pytest.mark.asyncio async def test_routine_status_traffic_never_touches_the_title(): inc = await _create(tags=["welfare-check"], location="55 Hyman Hills Road", incident_type="police") updates = await _link(inc, tags=[], incident_type=None, units=["6-Adam"]) assert "title" not in updates @pytest.mark.asyncio async def test_a_worse_event_takes_the_title_over(): """ The one case where the header must change: a check-condition that turns into a structure fire is a structure fire. Monotonic like _max_severity — a calmer later call can never take it back. """ inc = await _create(tags=["check-condition"], location="226 East Main Street", severity="minor", incident_type="police") assert inc["title"] == "Check Condition at 226 East Main Street" await _link(inc, tags=["structure-fire"], incident_type="fire", severity="major") assert inc["title"] == "Structure Fire at 226 East Main Street" assert inc["severity"] == "major" await _link(inc, tags=["ems-transport"], incident_type="ems", severity="routine") assert inc["title"] == "Structure Fire at 226 East Main Street" @pytest.mark.asyncio async def test_a_placeholder_title_is_filled_in_not_overwritten(): """ An incident that opened on a call with no content tags is named after its type ("Police — "). That is a placeholder, not an event name, so the first classified call may name it — and only the first. """ inc = await _create(tags=[], location=None, incident_type="police") assert inc["title"] == f"Police — {DISPATCH_TG}" assert inc["title_tag"] is None await _link(inc, tags=["vehicle-accident"], location="Airport Road", incident_type="police", severity="minor") assert inc["title"] == "Vehicle Accident at Airport Road" await _link(inc, tags=["disabled-vehicle"], location="Yonkers Avenue", incident_type="police", severity="minor") assert inc["title"] == "Vehicle Accident at Airport Road" @pytest.mark.asyncio async def test_the_title_picks_up_an_address_learned_later(): """ Same event, new information — not a rename. The founding call classified the event but named no place; a later call on the same event does. """ inc = await _create(tags=["welfare-check"], location=None, incident_type="police") assert inc["title"] == f"Welfare Check — {DISPATCH_TG}" await _link(inc, tags=["welfare-check"], location="55 Hyman Hills Road", incident_type="police") assert inc["title"] == "Welfare Check at 55 Hyman Hills Road" assert inc["location"] == "55 Hyman Hills Road" @pytest.mark.asyncio async def test_a_legacy_incidents_title_is_not_claimed_by_the_next_call(): """ Incidents created before this change have no `title_tag`, so their founding event is unrecoverable. Their existing title is treated as the founding one rather than handed to whichever call links next. """ legacy = { "incident_id": "b9b4f392", "title": "Suspect Search at 80 Grasslands Road", "location": "80 Grasslands Road", "call_ids": ["call-0"], "started_at": NOW.isoformat(), "updated_at": NOW.isoformat(), } updates = await _link(legacy, tags=["open-911"], location="100 South Mosher", incident_type="police", severity="routine") assert "title" not in updates assert updates["location"] == "80 Grasslands Road" # ── server-26#52: a unit call-sign must never become a map pin ──────────────── # # "Post 1-2" passed clean_location (it has a word in it), geocoded against the # Ossining anchor and produced a confident pin in the right town for an event # with no known location. It was in the same incident's `units` all along. @pytest.mark.parametrize("location,units", [ ("Post 1-2", ["1-2", "Lincoln", "Post 1-2"]), # the dump's actual incident ("post 1-2", ["Post 1-2"]), # case-blind ("Post 1-2.", ["Post 1-2"]), # punctuation-blind ("Engine 4", ["Engine 4", "Ladder 1"]), ]) def test_location_matching_a_unit_is_rejected(location, units): assert location_is_unit(location, units) is True @pytest.mark.parametrize("location,units", [ ("Water Street", ["1-2", "Post 1-2"]), # a real place, same incident ("South High", []), # no units extracted ("Riverdale Station", ["Lincoln"]), ("", ["Post 1-2"]), # nothing to compare (None, ["Post 1-2"]), ]) def test_real_places_survive_the_unit_check(location, units): assert location_is_unit(location, units) is False def test_unit_check_does_not_match_on_substrings(): """`1-2` is a unit; "1-2 Main Street" is an address that contains it.""" assert location_is_unit("1-2 Main Street", ["1-2"]) is False