server-26/drb-c2-core/tests/test_area_context.py

"""
Unit tests for the area_context schema and anchor (server-26#36).

Three properties carry the real risk:

  * NULLABILITY IS THE MECHANISM. Which scope an operator fills is their
    declaration of how homogeneous the system is. Merging must let a talkgroup
    narrow the system without dropping what the system already said — a
    talkgroup that sets only a town must still inherit the state, or "Ossining"
    is nationally ambiguous again.

  * NO ANCHOR IS BETTER THAN A USELESS ONE. An anchor wider than
    area_anchor_max_radius_km, or one whose resolved_from no longer matches the
    place it came from, must read as ABSENT. Verification then skips. Treating
    either as usable would rubber-stamp any location while looking like a check.

  * THE CLIENT DOES NOT WRITE SERVER FIELDS. The systems form sends
    config.talkgroups[] in full; taking it verbatim destroys the resolved anchor
    and the pending queue, which is the same bug as the ten_codes wipe.
"""
import pytest
from unittest.mock import AsyncMock, patch

from app.internal import area_context as ac

SYSTEM_AREA = {
    "county": "Westchester",
    "state": "New York",
    "local_knowledge": [{"term": "Route 9", "meaning": "state highway"}],
    "center": {"lat": 41.1, "lng": -73.8},
    "radius_km": 30.0,
    "resolved_from": "|westchester|new york",
    "resolved_at": "2026-08-23T00:00:00+00:00",
}

TG_AREA = {
    "municipality": "Ossining",
    "local_knowledge": [{"term": "Sing Sing", "meaning": "state prison"}],
    "center": {"lat": 41.16, "lng": -73.86},
    "radius_km": 6.0,
    "resolved_from": "ossining|westchester|new york",
    "resolved_at": "2026-08-23T00:00:00+00:00",
}


# -- Merging -------------------------------------------------------------------

def test_talkgroup_narrows_without_dropping_the_system():
    merged = ac.effective(SYSTEM_AREA, TG_AREA)
    assert merged["municipality"] == "Ossining"
    assert merged["county"] == "Westchester"
    assert merged["state"] == "New York", "the state must survive the narrowing"


def test_talkgroup_knowledge_ranks_first_and_dedupes():
    system = {"local_knowledge": [{"term": "Route 9"}, {"term": "Metro-North"}]}
    tg = {"local_knowledge": [{"term": "route 9", "meaning": "the local name"}]}
    merged = ac.effective(system, tg)
    assert [e["term"] for e in merged["local_knowledge"]] == ["route 9", "Metro-North"]
    assert merged["local_knowledge"][0]["meaning"] == "the local name"


def test_empty_at_both_scopes_is_legal():
    assert ac.effective(None, None) == {}
    assert ac.effective({}, {}) == {}


def test_bare_strings_are_accepted_as_terms():
    """roads[]/landmarks[] from the old shape, and anything a model returns."""
    assert ac.normalize_local_knowledge(["Route 9", "", "Route 9", 7]) == [{"term": "Route 9"}]


def test_pre_36_roads_and_landmarks_are_read_forward():
    """
    Real systems still have the old shape stored. Dropping it the day this
    shipped would silently discard ground truth an operator already entered.
    """
    legacy = {"county": "Westchester", "roads": ["Route 9"], "landmarks": ["Sing Sing"]}
    merged = ac.effective(legacy, None)
    assert [e["term"] for e in merged["local_knowledge"]] == ["Route 9", "Sing Sing"]
    assert ac.normalize(legacy) == {
        "county": "Westchester",
        "local_knowledge": [{"term": "Route 9"}, {"term": "Sing Sing"}],
    }, "and the next save writes them in the new shape"


def test_normalize_drops_client_sent_server_fields():
    out = ac.normalize({"municipality": " Ossining ", "radius_km": 5000, "center": {"lat": 0}})
    assert out == {"municipality": "Ossining"}


# -- Anchor selection ----------------------------------------------------------

def test_talkgroup_anchor_wins():
    anchor = ac.anchor_for(SYSTEM_AREA, TG_AREA)
    assert anchor == {"lat": 41.16, "lng": -73.86, "radius_km": 6.0}


def test_system_anchor_used_when_talkgroup_sets_no_place():
    anchor = ac.anchor_for(SYSTEM_AREA, {"local_knowledge": [{"term": "Post 4"}]})
    assert anchor == {"lat": 41.1, "lng": -73.8, "radius_km": 30.0}


def test_no_anchor_when_nothing_is_configured():
    assert ac.anchor_for({}, {}) is None


def test_stale_anchor_reads_as_absent():
    """
    Someone edited the town and the refresh has not run yet. The stored centre
    is for the OLD place, so using it would validate locations against an area
    the channel no longer covers.
    """
    stale = {**TG_AREA, "municipality": "Croton"}
    assert ac.anchor_for(SYSTEM_AREA, stale) is None


def test_anchor_key_ignores_case_and_padding():
    assert ac.anchor_key({"municipality": " OSSINING "}) == ac.anchor_key({"municipality": "ossining"})


# -- Anchor resolution ---------------------------------------------------------

def _maps(viewport_span_deg: float):
    """A geocode response whose viewport spans roughly the given degrees."""
    payload = {
        "status": "OK",
        "results": [{
            "geometry": {
                "location": {"lat": 41.0, "lng": -73.0},
                "viewport": {
                    "northeast": {"lat": 41.0 + viewport_span_deg, "lng": -73.0 + viewport_span_deg},
                    "southwest": {"lat": 41.0 - viewport_span_deg, "lng": -73.0 - viewport_span_deg},
                },
            }
        }],
    }

    class _Resp:
        def raise_for_status(self): pass
        def json(self): return payload

    class _Client:
        async def __aenter__(self): return self
        async def __aexit__(self, *a): return False
        async def get(self, *a, **k): return _Resp()

    return patch("httpx.AsyncClient", lambda *a, **k: _Client())


@pytest.fixture(autouse=True)
def _clear_cache():
    ac._anchor_cache.clear()
    with patch.object(ac.settings, "google_maps_api_key", "test-key"):
        yield
    ac._anchor_cache.clear()


@pytest.mark.asyncio
async def test_small_place_produces_an_anchor():
    with _maps(0.05):
        anchor = await ac.resolve_anchor({"municipality": "Ossining", "state": "New York"})
    assert anchor is not None
    assert anchor["radius_km"] < 10
    assert anchor["resolved_from"] == "ossining||new york"


@pytest.mark.asyncio
async def test_statewide_place_produces_no_anchor():
    """
    A radius that covers a state would confirm any location inside it. Storing
    it would make the geocode check worse than useless — it would look like
    verification and pass everything.
    """
    with _maps(4.0), patch.object(ac.settings, "area_anchor_max_radius_km", 60.0):
        assert await ac.resolve_anchor({"state": "Colorado"}) is None


@pytest.mark.asyncio
async def test_no_place_never_calls_maps():
    with patch("httpx.AsyncClient") as client:
        assert await ac.resolve_anchor({"local_knowledge": [{"term": "Post 4"}]}) is None
    client.assert_not_called()


@pytest.mark.asyncio
async def test_refresh_skips_scopes_whose_place_is_unchanged():
    doc = {"area_context": SYSTEM_AREA, "config": {"talkgroups": [{"id": 1, "area_context": TG_AREA}]}}
    with patch("httpx.AsyncClient") as client:
        assert await ac.refresh_anchors(doc) == {}
    client.assert_not_called()


@pytest.mark.asyncio
async def test_editing_the_system_place_re_anchors_its_talkgroups():
    """
    A talkgroup's anchor derives from its EFFECTIVE place, so changing the
    system's county silently changes what every talkgroup should be anchored to.
    """
    doc = {
        "area_context": {"county": "Putnam", "state": "New York"},
        "config": {"talkgroups": [{"id": 1, "area_context": {"municipality": "Ossining"}}]},
    }
    with _maps(0.05):
        patch_out = await ac.refresh_anchors(doc)
    tg = patch_out["config"]["talkgroups"][0]
    assert tg["area_context"]["resolved_from"] == "ossining|putnam|new york"
    assert tg["area_context"]["center"]["lat"] == 41.0


# -- Client writes -------------------------------------------------------------

def test_merge_config_preserves_the_anchor_and_the_pending_queue():
    existing = {"talkgroups": [{
        "id": 9048,
        "area_context": TG_AREA,
        ac.PENDING_KEY: [{"term": "Snowden Avenue"}],
    }]}
    # What the systems form actually sends: no anchor, no pending queue.
    incoming = {"talkgroups": [{"id": 9048, "name": "Ossining PD",
                                "area_context": {"municipality": "Ossining"}}]}
    merged = ac.merge_config(incoming, existing)
    tg = merged["talkgroups"][0]
    assert tg["area_context"]["center"] == TG_AREA["center"]
    assert tg[ac.PENDING_KEY] == [{"term": "Snowden Avenue"}]
    assert tg["name"] == "Ossining PD", "the client still owns the fields it owns"


def test_merge_config_drops_an_emptied_area():
    existing = {"talkgroups": [{"id": 1, "area_context": TG_AREA}]}
    merged = ac.merge_config({"talkgroups": [{"id": 1}]}, existing)
    assert "area_context" not in merged["talkgroups"][0]


# -- Pending terms -------------------------------------------------------------

def _store(doc):
    saved = {}

    async def _get(_col, _id):
        return doc

    async def _update(_col, _id, patch):
        saved.update(patch)

    return saved, patch.multiple(
        "app.internal.firestore", doc_get=AsyncMock(side_effect=_get),
        doc_update=AsyncMock(side_effect=_update),
    )


@pytest.mark.asyncio
async def test_pending_terms_land_on_the_talkgroup():
    doc = {"config": {"talkgroups": [{"id": 9048}]}, "vocabulary": []}
    saved, store = _store(doc)
    with store:
        assert await ac.add_pending("sys-1", 9048, [{"term": "Snowden Avenue"}]) == 1
    assert saved["config"]["talkgroups"][0][ac.PENDING_KEY][0]["term"] == "Snowden Avenue"
    assert "vocabulary" not in saved, "nothing writes to the system"


@pytest.mark.asyncio
async def test_already_known_terms_are_not_re_proposed():
    doc = {
        "vocabulary": ["Metro-North"],
        "area_context": {"local_knowledge": [{"term": "Route 9"}]},
        "config": {"talkgroups": [{"id": 9048, "local_knowledge_pending": [{"term": "Sing Sing"}]}]},
    }
    saved, store = _store(doc)
    with store:
        queued = await ac.add_pending("sys-1", 9048, [
            {"term": "route 9"}, {"term": "Metro-North"}, {"term": "sing sing"},
        ])
    assert queued == 0
    assert saved == {}


@pytest.mark.asyncio
async def test_approving_writes_to_the_talkgroup_and_never_the_system():
    """
    Blast radius: the same term at system level misleads every channel on the
    system, including one 400km away on a statewide system.
    """
    doc = {"config": {"talkgroups": [{"id": 9048, ac.PENDING_KEY: [
        {"term": "Snowden Avenue", "meaning": "residential street"}]}]}}
    saved, store = _store(doc)
    with store:
        assert await ac.resolve_pending("sys-1", 9048, "snowden avenue", approve=True) is True
    tg = saved["config"]["talkgroups"][0]
    assert tg["area_context"]["local_knowledge"] == [
        {"term": "Snowden Avenue", "meaning": "residential street"}
    ]
    assert tg[ac.PENDING_KEY] == []
    assert "vocabulary" not in saved and "area_context" not in saved


@pytest.mark.asyncio
async def test_dismissing_adds_nothing():
    doc = {"config": {"talkgroups": [{"id": 9048, ac.PENDING_KEY: [{"term": "Optum"}]}]}}
    saved, store = _store(doc)
    with store:
        assert await ac.resolve_pending("sys-1", 9048, "Optum", approve=False) is True
    tg = saved["config"]["talkgroups"][0]
    assert tg[ac.PENDING_KEY] == []
    assert not (tg.get("area_context") or {}).get("local_knowledge")