central/tests/test_gui_adapter_edit.py

"""v0.9.9 — adapter edit form: generic model_list editor (fixes the
describe_fields list[BaseModel] 500 across 4 adapters), TomTom bbox validation,
and quota gating."""

from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from starlette.datastructures import FormData
from starlette.requests import Request

from central.gui import templates as gui_templates
from central.gui.form_descriptors import describe_fields
from central.gui.routes import adapters_edit_form, adapters_edit_submit
from central.adapters.tomtom_incidents import (
    TomTomIncidentsAdapter, TomTomIncidentsSettings, TOMTOM_FREE_TIER_CALLS_PER_MONTH,
)

_BBOXES = [
    {"name": "treasure_valley_ext", "min_lon": -117.05, "min_lat": 43.30,
     "max_lon": -116.00, "max_lat": 44.30, "state_code": "ID", "cadence_s": 3600},
    {"name": "mountain_home_corridor", "min_lon": -116.00, "min_lat": 42.65,
     "max_lon": -114.85, "max_lat": 43.40, "state_code": "ID", "cadence_s": 5400},
    {"name": "magic_valley_burley", "min_lon": -114.85, "min_lat": 42.30,
     "max_lon": -113.40, "max_lat": 42.95, "state_code": "ID", "cadence_s": 3600},
]
_SETTINGS = {"bboxes": _BBOXES, "api_key_alias": "tomtom"}
# A small (~2.3k km^2) valid box used as the base for POST tests.
_SMALL = {"min_lon": "-117.0", "min_lat": "43.0", "max_lon": "-116.5", "max_lat": "43.5"}


def _render(name, ctx):
    req = Request({"type": "http", "method": "GET", "path": "/", "headers": [], "query_string": b""})
    return gui_templates.TemplateResponse(request=req, name=name, context=ctx).body.decode()


def _ctx(settings, fields, quota, name="tomtom_incidents", display="TomTom"):
    return {
        "operator": SimpleNamespace(username="admin"), "csrf_token": "x",
        "adapter": {"name": name, "display_name": display, "description": "", "enabled": True,
                    "cadence_s": 1800, "settings": settings, "paused_at": None,
                    "updated_at": None, "last_error": None},
        "fields": fields, "api_keys": [{"alias": "tomtom"}], "errors": None, "form_data": None,
        "tile_url": "https://t/{z}/{x}/{y}.png", "tile_attribution": "OSM",
        "api_key_missing": False, "requires_api_key_alias": "tomtom",
        "preview_rows": None, "preview_error": None, "quota": quota,
    }


class TestModelListRender:
    def test_tomtom_renders_rows_and_quota(self):
        fields = describe_fields(TomTomIncidentsAdapter.settings_schema, _SETTINGS)
        quota = TomTomIncidentsAdapter.quota_estimate(TomTomIncidentsSettings(**_SETTINGS), 1800)
        out = _render("adapters_edit.html", _ctx(_SETTINGS, fields, quota))
        assert "treasure_valley_ext" in out and "magic_valley_burley" in out
        assert "model-list" in out and 'name="bboxes-0-name"' in out
        assert "free tier" in out  # quota panel rendered

    def test_nws_region_intact_no_model_list(self):
        from central.adapters.nws import NWSSettings
        s = {"contact_email": "a@b.com",
             "region": {"north": 49.5, "south": 31.0, "east": -102.0, "west": -124.5}}
        out = _render("adapters_edit.html",
                      _ctx(s, describe_fields(NWSSettings, s), None, name="nws", display="NWS"))
        assert "region-map" in out                 # region picker intact
        assert "model-list" not in out and "free tier" not in out

    def test_tomtom_flow_renders_tiles_editor(self):
        from central.adapters.tomtom_flow import TomTomFlowAdapter
        s = {"tiles": [{"z": 10, "x": 181, "y": 373}], "api_key_alias": "tomtom"}
        out = _render("adapters_edit.html",
                      _ctx(s, describe_fields(TomTomFlowAdapter.settings_schema, s), None,
                           name="tomtom_flow", display="Flow"))
        assert "model-list" in out and 'name="tiles-0-z"' in out


class TestDescribeFieldsModelList:
    def test_bbox_is_model_list_with_subfields(self):
        bf = next(f for f in describe_fields(TomTomIncidentsSettings, _SETTINGS) if f.name == "bboxes")
        assert bf.widget == "model_list"
        assert [s.name for s in bf.sub_fields] == \
            ["name", "min_lon", "min_lat", "max_lon", "max_lat", "state_code", "cadence_s"]
        assert bf.current_value[0]["name"] == "treasure_valley_ext"


def _post_req(pairs, cadence_s="1800"):
    req = MagicMock()
    req.state.operator = SimpleNamespace(id=1, username="admin")
    req.state.csrf_token = "x"
    req.form = AsyncMock(return_value=FormData(
        [("csrf_token", "x"), ("enabled", "on"), ("cadence_s", cadence_s)] + pairs))
    return req


def _pool(settings=_SETTINGS):
    conn = AsyncMock()
    conn.fetchrow.side_effect = [
        {"name": "tomtom_incidents", "enabled": True, "cadence_s": 1800, "settings": settings,
         "paused_at": None, "updated_at": None, "last_error": None},
        {"map_tile_url": None, "map_attribution": None},
    ]
    conn.fetch.return_value = [{"alias": "tomtom"}]
    pool = MagicMock()
    pool.acquire.return_value.__aenter__ = AsyncMock(return_value=conn)
    pool.acquire.return_value.__aexit__ = AsyncMock(return_value=None)
    return pool, conn


def _bbox_pairs(i, name, state="ID", cadence=None, **coords):
    c = {**_SMALL, **coords}
    p = [(f"bboxes-{i}-name", name), (f"bboxes-{i}-state_code", state),
         (f"bboxes-{i}-min_lon", c["min_lon"]), (f"bboxes-{i}-min_lat", c["min_lat"]),
         (f"bboxes-{i}-max_lon", c["max_lon"]), (f"bboxes-{i}-max_lat", c["max_lat"])]
    if cadence is not None:
        p.append((f"bboxes-{i}-cadence_s", cadence))
    return p


async def _submit_expect_422(pairs, cadence_s="1800"):
    pool, _ = _pool()
    tmpl = MagicMock(); tmpl.TemplateResponse.return_value = MagicMock()
    with patch("central.gui.routes._get_templates", return_value=tmpl), \
         patch("central.gui.routes.get_pool", return_value=pool), \
         patch("central.gui.routes.adapter_has_resolved_api_key",
               new=AsyncMock(return_value=(True, "tomtom"))):
        await adapters_edit_submit(_post_req(pairs + [("api_key_alias", "tomtom")], cadence_s),
                                   "tomtom_incidents")
    ca = tmpl.TemplateResponse.call_args
    assert ca.kwargs["status_code"] == 422
    return ca.kwargs["context"]["errors"]


class TestBboxValidation:
    @pytest.mark.asyncio
    async def test_valid_persists(self):
        pool, conn = _pool()
        captured = {}
        async def cap(q, *a):
            if "UPDATE config.adapters" in q:
                captured["s"] = a[2]
        conn.execute.side_effect = cap
        pairs = _bbox_pairs(0, "metro", cadence="3600") + [("api_key_alias", "tomtom")]
        with patch("central.gui.routes.get_pool", return_value=pool), \
             patch("central.gui.routes.write_audit", new=AsyncMock()):
            res = await adapters_edit_submit(_post_req(pairs), "tomtom_incidents")
        assert res.status_code == 302
        assert captured["s"]["bboxes"][0]["name"] == "metro"

    @pytest.mark.asyncio
    async def test_oversized_422(self):
        errs = await _submit_expect_422(
            _bbox_pairs(0, "big", min_lon="-120", min_lat="30", max_lon="-100", max_lat="45"))
        assert "bboxes" in errs

    @pytest.mark.asyncio
    async def test_degenerate_422(self):
        errs = await _submit_expect_422(_bbox_pairs(0, "bad", min_lon="-100", max_lon="-110"))
        assert "bboxes" in errs

    @pytest.mark.asyncio
    async def test_duplicate_names_422(self):
        errs = await _submit_expect_422(_bbox_pairs(0, "dup") + _bbox_pairs(1, "dup"))
        assert "bboxes" in errs

    @pytest.mark.asyncio
    async def test_cadence_floor_422(self):
        errs = await _submit_expect_422(_bbox_pairs(0, "metro", cadence="30"))
        assert "bboxes" in errs

    @pytest.mark.asyncio
    async def test_quota_block_422(self):
        # top cadence 300, one bbox cadence 60 -> 8,640/mo > 2,500 cap
        errs = await _submit_expect_422(_bbox_pairs(0, "metro", cadence="60"), cadence_s="300")
        assert "bboxes" in errs and "free-tier" in errs["bboxes"]


class TestGetRoute:
    @pytest.mark.asyncio
    async def test_get_200_with_quota_in_context(self):
        pool, _ = _pool()
        tmpl = MagicMock(); tmpl.TemplateResponse.return_value = MagicMock(status_code=200)
        req = MagicMock()
        req.state.operator = SimpleNamespace(id=1, username="admin")
        req.state.csrf_token = "x"
        with patch("central.gui.routes._get_templates", return_value=tmpl), \
             patch("central.gui.routes.get_pool", return_value=pool), \
             patch("central.gui.routes.adapter_has_resolved_api_key",
                   new=AsyncMock(return_value=(True, "tomtom"))):
            await adapters_edit_form(req, "tomtom_incidents")
        ctx = tmpl.TemplateResponse.call_args.kwargs["context"]
        assert ctx["quota"]["calls_per_month"] == 1920  # 720+480+720
        assert ctx["quota"]["blocked"] is False