Merge branch 'main' into fix/153-ai-request-timeouts

2026-10-04 18:29:00 -04:00
parent 278edf48dc 30bca25ecc
commit 8d9fe01828
33 changed files with 3683 additions and 141 deletions
@@ -0,0 +1,312 @@
+"""Live ops wall camera providers: registry, licence gate, admin routes, adapters.
+
+What is held here:
+  * every provider defaults OFF, and an off provider is never fetched;
+  * stubs and PROHIBITED sources can never be enabled;
+  * a gated licence needs ``acknowledge_licence``, which is stored with who/when
+    and audited -- and an audit failure refuses the enable (fail closed);
+  * an acknowledgement given for a different licence status no longer counts;
+  * the admin routes are admin-only;
+  * each implemented adapter parses its (verified) upstream shape, pins imagery
+    to the agency's hosts, and treats an all-unparseable payload as an error.
+"""
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+from app.internal.auth import require_admin_token, require_service_or_firebase_token
+from app.internal.cameras import base, general, regional, registry
+from app.internal.cameras.base import Licence
+from app.main import app
+
+client = TestClient(app)
+ADMIN = {"uid": "admin-1", "email": "admin@example.com", "role": "admin"}
+
+
+class FakeStore:
+    def __init__(self):
+        self.docs = {}
+        self.audit = []
+
+    async def doc_get(self, col, doc):
+        return self.docs.get((col, doc))
+
+    async def doc_set(self, col, doc, data, merge=True):
+        cur = self.docs.setdefault((col, doc), {})
+        if col == "audit_log":
+            self.audit.append(data)
+            return
+        for k, v in data.items():   # one-level deep merge is all registry needs
+            if merge and isinstance(v, dict) and isinstance(cur.get(k), dict):
+                cur[k] = {**cur[k], **v}
+            else:
+                cur[k] = v
+
+
+@pytest.fixture(autouse=True)
+def store():
+    s = FakeStore()
+    registry.invalidate_cache()
+    for cat in registry.CATALOGS.values():
+        cat.items, cat.fetched_at, cat.error, cat._next_attempt = [], None, None, 0.0
+    with patch.object(registry.fstore, "doc_get", s.doc_get), patch.object(registry.fstore, "doc_set", s.doc_set), \
+            patch("app.internal.audit.fstore.doc_set", s.doc_set):
+        yield s
+    registry.invalidate_cache()
+    app.dependency_overrides.clear()
+
+
+def run(coro):
+    return asyncio.run(coro)
+
+
+# ── registry / gate ──────────────────────────────────────────────────────────
+
+def test_every_catalogued_source_is_registered_and_defaults_off():
+    listed = run(registry.list_providers())
+    ids = {p["id"] for p in listed}
+    assert {"ny511", "nyc_dot", "westchester_dpw", "caltrans", "deldot", "tfl", "ontario511", "drivebc",
+            "calgary", "nsw", "fintraffic", "txdot", "austin", "tallinn", "tarktee", "warendorf"} <= ids
+    assert all(p["enabled"] is False and p["serving"] is False for p in listed)
+    assert all(p["licence"]["label"] for p in listed)
+
+
+def test_off_provider_is_never_fetched():
+    fetch = AsyncMock(return_value=[])
+    with patch.object(registry.PROVIDERS["tfl"], "fetch_catalog", fetch):
+        out = run(registry.cameras_in_bbox((51.4, -0.2, 51.6, 0.1)))
+    assert out["cameras"] == [] and out["providers"] == [] and fetch.await_count == 0
+
+
+def test_stub_and_prohibited_cannot_be_enabled():
+    for pid in ("nyc_dot", "westchester_dpw", "ctroads", "trafficland"):
+        with pytest.raises(registry.ProviderError) as e:
+            run(registry.set_enabled(pid, True, ("u", "e"), acknowledge_licence=True))
+        assert e.value.code == 409
+
+
+def test_attribution_licence_enables_without_acknowledgement_and_is_audited(store):
+    out = run(registry.set_enabled("nsw", True, ("u1", "a@x")))
+    assert out["serving"] is True and out["acknowledgement"] is None
+    assert store.audit[-1]["action"] == "camera_provider.enable"
+    assert store.audit[-1]["actor_email"] == "a@x"
+
+
+def test_gated_licence_requires_acknowledgement(store):
+    with pytest.raises(registry.ProviderError) as e:
+        run(registry.set_enabled("ny511", True, ("u1", "a@x")))
+    assert e.value.code == 409 and e.value.extra["requires_acknowledgement"] is True
+    assert "Developer Access Agreement" in e.value.extra["warning"]
+    assert not store.audit and not run(registry.is_serving("ny511"))
+
+    out = run(registry.set_enabled("ny511", True, ("u1", "a@x"), acknowledge_licence=True))
+    assert out["serving"] is True
+    ack = out["acknowledgement"]
+    assert ack["by_uid"] == "u1" and ack["by_email"] == "a@x" and ack["at"]
+    assert ack["licence_status"] == Licence.AGREEMENT_REQUIRED.value
+    assert store.audit[-1]["action"] == "camera_provider.enable_acknowledged"
+    assert store.audit[-1]["details"]["provider_id"] == "ny511"
+
+
+def test_licences_not_read_first_hand_are_gated_but_keep_attribution():
+    # TfL (terms page 403'd) and Calgary (per GEV only) are not confirmed, so an
+    # admin must acknowledge before they serve; their credit still renders.
+    for pid in ("tfl", "calgary", "ontario511"):
+        lic = registry.PROVIDERS[pid].meta.licence
+        assert lic.status == Licence.UNVERIFIED and lic.gated, pid
+    assert registry.PROVIDERS["tfl"].meta.licence.attribution == "Powered by TfL Open Data"
+    with pytest.raises(registry.ProviderError) as e:
+        run(registry.set_enabled("tfl", True, ("u", "e")))
+    assert e.value.extra["requires_acknowledgement"] is True
+
+
+def test_gated_enable_refused_when_audit_cannot_be_written(store):
+    with patch.object(registry, "_audit", AsyncMock(side_effect=RuntimeError("firestore down"))):
+        with pytest.raises(registry.ProviderError) as e:
+            run(registry.set_enabled("deldot", True, ("u", "e"), acknowledge_licence=True))
+    assert e.value.code == 503
+    assert not run(registry.is_serving("deldot"))
+
+
+def test_disable_clears_acknowledgement(store):
+    run(registry.set_enabled("deldot", True, ("u", "e"), acknowledge_licence=True))
+    out = run(registry.set_enabled("deldot", False, ("u", "e")))
+    assert out["serving"] is False and out["acknowledgement"] is None
+
+
+def test_stale_acknowledgement_stops_serving():
+    p = registry.PROVIDERS["deldot"]
+    entry = {"enabled": True, "acknowledgement": {"licence_status": "non_commercial"}}
+    assert registry.effective(p, entry) == (False, "licence acknowledgement missing or out of date")
+    entry["acknowledgement"]["licence_status"] = p.meta.licence.status.value
+    assert registry.effective(p, entry) == (True, None)
+
+
+def test_unreadable_settings_fail_closed():
+    with patch.object(registry.fstore, "doc_get", AsyncMock(side_effect=RuntimeError("boom"))):
+        registry.invalidate_cache()
+        assert run(registry.get_settings()) == {}
+
+
+def test_query_returns_attribution_and_skips_providers_outside_bbox(store):
+    run(registry.set_enabled("nsw", True, ("u", "e")))
+    run(registry.set_enabled("tfl", True, ("u", "e"), acknowledge_licence=True))
+    cam = base.make_camera("nsw", "x", -33.87, 151.2, name="Sydney", image_url="https://webcams.transport.nsw.gov.au/a.jpg")
+    nsw_fetch, tfl_fetch = AsyncMock(return_value=[cam]), AsyncMock(return_value=[])
+    with patch.object(registry.PROVIDERS["nsw"], "fetch_catalog", nsw_fetch), \
+            patch.object(registry.PROVIDERS["tfl"], "fetch_catalog", tfl_fetch):
+        out = run(registry.cameras_in_bbox((-34.0, 151.0, -33.7, 151.4)))
+    assert [c["id"] for c in out["cameras"]] == ["nsw:x"]
+    by_id = {p["id"]: p for p in out["providers"]}
+    assert by_id["nsw"]["attribution"] == "Transport for NSW, CC BY 4.0"
+    assert tfl_fetch.await_count == 0   # London never fetched for a Sydney bbox
+    # ...but its credit is still listed, for cameras the client holds from another view
+    assert by_id["tfl"]["attribution"] == "Powered by TfL Open Data" and by_id["tfl"]["total_in_bbox"] == 0
+    assert out["serving_count"] == 2
+
+
+# ── admin routes ─────────────────────────────────────────────────────────────
+
+def test_admin_routes_require_admin():
+    def deny():
+        raise HTTPException(403, "Admin access required")
+    app.dependency_overrides[require_admin_token] = deny
+    assert client.get("/admin/camera-providers").status_code == 403
+    assert client.put("/admin/camera-providers/nsw", json={"enabled": True}).status_code == 403
+    assert client.get("/admin/camera-providers", headers={}).status_code == 403
+
+
+def test_admin_route_toggle_and_gate(store):
+    app.dependency_overrides[require_admin_token] = lambda: ADMIN
+    r = client.put("/admin/camera-providers/caltrans", json={"enabled": True})
+    assert r.status_code == 409 and r.json()["requires_acknowledgement"] is True
+    r = client.put("/admin/camera-providers/caltrans", json={"enabled": True, "acknowledge_licence": True})
+    assert r.status_code == 200 and r.json()["acknowledgement"]["by_email"] == "admin@example.com"
+    assert client.put("/admin/camera-providers/nope", json={"enabled": True}).status_code == 404
+    listed = client.get("/admin/camera-providers").json()["providers"]
+    assert next(p for p in listed if p["id"] == "caltrans")["serving"] is True
+
+
+def test_cameras_route_rejects_inverted_bbox():
+    app.dependency_overrides[require_service_or_firebase_token] = lambda: {"uid": "v"}
+    assert client.get("/cameras", params={"south": 2, "west": 0, "north": 1, "east": 1}).status_code == 400
+    r = client.get("/cameras", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4})
+    assert r.status_code == 200 and r.json()["cameras"] == []
+
+
+# ── adapters ─────────────────────────────────────────────────────────────────
+
+def _with_payload(payload):
+    return patch.object(general, "get_json", AsyncMock(return_value=payload))
+
+
+def test_heading_from_text():
+    assert base.heading_from_text("Northbound") == 0
+    assert base.heading_from_text("WB") == 270
+    assert base.heading_from_text("SE") == 135
+    assert base.heading_from_text("Both directions") is None
+    assert base.heading_from_text("Unknown") is None
+
+
+def test_https_on_pins_hosts():
+    assert base.https_on("https://cwwp2.dot.ca.gov/x.jpg", "dot.ca.gov")
+    assert base.https_on("http://cwwp2.dot.ca.gov/x.jpg", "dot.ca.gov") is None
+    assert base.https_on("https://dot.ca.gov.evil.com/x.jpg", "dot.ca.gov") is None
+
+
+def test_caltrans_adapter():
+    row = {"cctv": {"inService": "true", "index": "1", "location": {
+        "district": "4", "locationName": "TV102 -- I-580 : West of SR-24", "latitude": "37.82539",
+        "longitude": "-122.27291", "direction": "West", "route": "I-580"},
+        "imageData": {"streamingVideoURL": "https://wzmedia.dot.ca.gov/D4/x.stream/playlist.m3u8",
+                      "static": {"currentImageURL": "https://cwwp2.dot.ca.gov/data/d4/cctv/image/a/a.jpg"}}}}
+    off = {"cctv": {**row["cctv"], "inService": "false"}}
+    with _with_payload({"data": [row, off]}):
+        cams = run(general.CaltransProvider().fetch_catalog())
+    assert len(cams) == len(general.CaltransProvider.DISTRICTS)   # same row per district fetch
+    c = cams[0]
+    assert c["id"] == "caltrans:d4-TV102" and c["heading_deg"] == 270 and c["video_url"].endswith(".m3u8")
+
+
+def test_deldot_adapter_requires_https_hls():
+    good = {"id": "KCAM001", "title": "DE 1", "lat": 38.99, "lon": -75.44, "enabled": True, "status": "Active",
+            "urls": {"m3u8s": "https://video.deldot.gov:443/live/KCAM001.stream/playlist.m3u8",
+                     "m3u8": "http://video.deldot.gov:1935/live/KCAM001.stream/playlist.m3u8"}}
+    down = {**good, "id": "KCAM002", "status": "Unavailable"}
+    with _with_payload({"videoCameras": [good, down]}):
+        cams = run(general.DelDotProvider().fetch_catalog())
+    assert [c["id"] for c in cams] == ["deldot:KCAM001"] and cams[0]["image_url"] is None
+
+
+def test_drivebc_calgary_tfl_nsw_fintraffic_adapters():
+    with _with_payload([{"id": 1176, "name": "Lougheed", "is_on": True, "should_appear": True, "orientation": "W",
+                         "location": {"coordinates": [-122.5, 49.2]}, "highway_description": "Hwy 7"}]):
+        assert run(general.DriveBcProvider().fetch_catalog())[0]["image_url"] == "https://www.drivebc.ca/images/1176.jpg"
+    with _with_payload([{"camera_url": {"url": "http://trafficcam.calgary.ca/loc86.jpg"}, "camera_location": "Stoney",
+                         "point": {"coordinates": [-113.97, 50.9]}, "quadrant": "SE"}]):
+        c = run(general.CalgaryProvider().fetch_catalog())[0]
+        assert c["image_url"] == "https://trafficcam.calgary.ca/loc86.jpg" and c["id"] == "calgary:loc86"
+    with _with_payload([{"id": "JamCams_00002.00865", "commonName": "A406", "lat": 51.6, "lon": -0.016,
+                         "additionalProperties": [{"key": "available", "value": "true"},
+                                                  {"key": "imageUrl", "value": general.TflProvider.IMAGE_PREFIX + "a.jpg"}]}]):
+        assert run(general.TflProvider().fetch_catalog())[0]["id"] == "tfl:00002.00865"
+    with _with_payload({"features": [{"id": "abc", "geometry": {"coordinates": [151.1, -34.0]},
+                                      "properties": {"title": "5 Ways", "direction": "W",
+                                                     "href": "https://webcams.transport.nsw.gov.au/x.jpeg"}}]}):
+        assert run(general.NswProvider().fetch_catalog())[0]["heading_deg"] == 270
+    with _with_payload({"features": [{"geometry": {"coordinates": [23.99, 60.05, 0]},
+                                      "properties": {"id": "C01503", "name": "kt51", "collectionStatus": "GATHERING",
+                                                     "presets": [{"id": "C0150301", "inCollection": True},
+                                                                 {"id": "C0150302", "inCollection": False}]}}]}):
+        cams = run(general.FintrafficProvider().fetch_catalog())
+        assert [c["image_url"] for c in cams] == ["https://weathercam.digitraffic.fi/C0150301.jpg"]
+
+
+def test_schema_change_is_an_error_not_an_empty_catalog():
+    with _with_payload([{"renamed": True}]):
+        with pytest.raises(ValueError, match="schema"):
+            run(general.DriveBcProvider().fetch_catalog())
+
+
+def test_ontario_needs_its_key(monkeypatch):
+    monkeypatch.setattr(general.settings, "ontario511_api_key", None, raising=False)
+    with pytest.raises(RuntimeError, match="ONTARIO511_API_KEY"):
+        run(general.Ontario511Provider().fetch_catalog())
+
+
+def test_ny511_provider_wraps_legacy_client_and_carries_daa_metadata():
+    row = {"id": "NYSDOT-1", "lat": 41.03, "lon": -73.76, "name": "I-287", "roadway": "I-287",
+           "direction": "Northbound", "image_url": "https://x/1", "video_url": "javascript:alert(1)"}
+    with patch.object(regional.ny511, "_fetch", AsyncMock(return_value=[row])):
+        cams = run(regional.Ny511Provider().fetch_catalog())
+    assert cams[0]["id"] == "ny511:NYSDOT-1" and cams[0]["heading_deg"] == 0
+    assert cams[0]["video_url"] is None   # non-https never reaches the browser
+    lic = regional.Ny511Provider.meta.licence
+    assert lic.attribution == "powered by 511NY" and lic.attribution_url == "https://511ny.org"
+    assert "with all faults" in lic.disclaimer and lic.gated
+
+
+def test_refresh_error_never_echoes_an_api_key():
+    import httpx
+    req = httpx.Request("GET", "https://511on.ca/api/v2/get/cameras", params={"format": "json", "key": "SECRET123"})
+    with pytest.raises(httpx.HTTPStatusError) as raised:   # httpx's own message carries the full URL
+        httpx.Response(403, request=req).raise_for_status()
+    err = raised.value
+    cat = registry.CATALOGS["ontario511"]
+    with patch.object(cat.provider, "fetch_catalog", AsyncMock(side_effect=err)):
+        run(cat.get())
+    assert "SECRET123" not in cat.error and "403" in cat.error and "511on.ca/api/v2/get/cameras" in cat.error
+
+
+def test_failed_refresh_keeps_last_good_catalog():
+    cat = registry.CATALOGS["nsw"]
+    cam = base.make_camera("nsw", "x", -33.8, 151.2)
+    with patch.object(cat.provider, "fetch_catalog", AsyncMock(return_value=[cam])):
+        run(cat.get())
+    cat._next_attempt = 0.0
+    with patch.object(cat.provider, "fetch_catalog", AsyncMock(side_effect=RuntimeError("down"))):
+        run(cat.get())
+    assert cat.items == [cam] and "down" in cat.error and cat.fetched_at
@@ -0,0 +1,203 @@
+"""
+server-26#195 — reject must revoke: delete the node's dynsec (MQTT) login and
+its node_keys entry, /upload must refuse a node that is not approved, and
+dynsec.reconcile_all() must delete rejected nodes' logins on startup.
+"""
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi.testclient import TestClient
+
+from app.main import app
+from app.internal import dynsec
+from app.internal.auth import require_admin_token, require_service_or_firebase_token
+from app.routers import nodes, upload
+
+client = TestClient(app)
+
+
+def setup_function():
+    app.dependency_overrides[require_admin_token] = lambda: {"admin": True}
+    app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True}
+
+
+def teardown_function():
+    app.dependency_overrides.pop(require_admin_token, None)
+    app.dependency_overrides.pop(require_service_or_firebase_token, None)
+
+
+def _fake_store(node, key_doc):
+    store = {"nodes": node, "node_keys": key_doc}
+    calls = []
+
+    async def doc_get(c, i):
+        return store.get(c)
+
+    async def doc_update(c, i, data):
+        calls.append(("doc_update", c, data))
+        store[c] = {**(store[c] or {}), **data}
+
+    async def doc_delete(c, i):
+        calls.append(("doc_delete", c))
+        store[c] = None
+
+    async def doc_set(c, i, data, merge=False):
+        store[c] = data
+
+    return store, calls, doc_get, doc_update, doc_delete, doc_set
+
+
+# ---------------------------------------------------------------------------
+# reject_node
+# ---------------------------------------------------------------------------
+
+def test_reject_revokes_mqtt_login_and_key_in_fail_closed_order():
+    store, calls, doc_get, doc_update, doc_delete, _ = _fake_store(
+        {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"})
+
+    async def delete_client(node_id):
+        calls.append(("dynsec_delete", node_id))
+
+    with patch.object(nodes.fstore, "doc_get", doc_get), \
+         patch.object(nodes.fstore, "doc_update", doc_update), \
+         patch.object(nodes.fstore, "doc_delete", doc_delete), \
+         patch.object(nodes.dynsec, "delete_node_client", delete_client):
+        resp = client.post("/nodes/n1/reject")
+
+    assert resp.status_code == 200
+    assert resp.json() == {"ok": True, "revoked": True}
+    assert calls == [
+        ("doc_update", "nodes", {"approval_status": "rejected"}),
+        ("dynsec_delete", "n1"),
+        ("doc_delete", "node_keys"),
+    ]
+    assert store["node_keys"] is None
+    assert store["nodes"]["approval_status"] == "rejected"
+
+
+def test_reject_dynsec_failure_is_502_and_keeps_key_but_status_is_already_rejected():
+    store, calls, doc_get, doc_update, doc_delete, _ = _fake_store(
+        {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "live-key"})
+
+    with patch.object(nodes.fstore, "doc_get", doc_get), \
+         patch.object(nodes.fstore, "doc_update", doc_update), \
+         patch.object(nodes.fstore, "doc_delete", doc_delete), \
+         patch.object(nodes.dynsec, "delete_node_client",
+                      AsyncMock(side_effect=dynsec.DynsecError("timeout"))):
+        resp = client.post("/nodes/n1/reject")
+
+    assert resp.status_code == 502
+    assert "Retry reject" in resp.json()["detail"]
+    assert store["nodes"]["approval_status"] == "rejected"
+    assert store["node_keys"] == {"node_id": "n1", "api_key": "live-key"}
+    assert ("doc_delete", "node_keys") not in calls
+
+
+def test_reject_unknown_node_is_404_and_touches_nothing():
+    with patch.object(nodes.fstore, "doc_get", AsyncMock(return_value=None)), \
+         patch.object(nodes.fstore, "doc_update", AsyncMock()) as doc_update, \
+         patch.object(nodes.fstore, "doc_delete", AsyncMock()) as doc_delete, \
+         patch.object(nodes.dynsec, "delete_node_client", AsyncMock()) as delete_client:
+        resp = client.post("/nodes/ghost/reject")
+    assert resp.status_code == 404
+    doc_update.assert_not_awaited()
+    doc_delete.assert_not_awaited()
+    delete_client.assert_not_awaited()
+
+
+def test_reject_requires_admin():
+    app.dependency_overrides.pop(require_admin_token, None)
+    resp = client.post("/nodes/n1/reject")
+    assert resp.status_code in (401, 403)
+
+
+def test_reapprove_after_reject_mints_a_new_key_never_the_revoked_one():
+    store, _, doc_get, doc_update, doc_delete, doc_set = _fake_store(
+        {"node_id": "n1", "approval_status": "approved"}, {"node_id": "n1", "api_key": "old-key"})
+
+    with patch.object(nodes.fstore, "doc_get", doc_get), \
+         patch.object(nodes.fstore, "doc_update", doc_update), \
+         patch.object(nodes.fstore, "doc_delete", doc_delete), \
+         patch.object(nodes.fstore, "doc_set", doc_set), \
+         patch.object(nodes.dynsec, "delete_node_client", AsyncMock()), \
+         patch.object(nodes.dynsec, "upsert_node_client", AsyncMock()) as upsert, \
+         patch.object(nodes.mqtt_handler, "publish_node_key", MagicMock()):
+        assert client.post("/nodes/n1/reject").status_code == 200
+        assert client.post("/nodes/n1/approve").status_code == 200
+
+    new_key = store["node_keys"]["api_key"]
+    assert new_key != "old-key" and len(new_key) == 64
+    upsert.assert_awaited_once_with("n1", new_key)
+
+
+# ---------------------------------------------------------------------------
+# /upload approval gate
+# ---------------------------------------------------------------------------
+
+def _upload(node_doc, key="live-key"):
+    store = {"node_keys": {"node_id": "n1", "api_key": "live-key"}, "nodes": node_doc}
+
+    async def doc_get(c, i):
+        return store.get(c)
+
+    with patch.object(upload.fstore, "doc_get", doc_get), \
+         patch.object(upload, "upload_audio", AsyncMock(return_value=None)) as up:
+        resp = client.post(
+            "/upload",
+            data={"call_id": "c1", "node_id": "n1"},
+            files={"file": ("a.mp3", b"\x00\x01", "audio/mpeg")},
+            headers={"Authorization": f"Bearer {key}"},
+        )
+    return resp, up
+
+
+def test_upload_refuses_valid_key_from_rejected_node():
+    resp, up = _upload({"node_id": "n1", "approval_status": "rejected"})
+    assert resp.status_code == 403
+    up.assert_not_awaited()
+
+
+def test_upload_refuses_valid_key_from_pending_or_missing_node():
+    for node_doc in ({"node_id": "n1", "approval_status": "pending"}, {"node_id": "n1"}, None):
+        resp, up = _upload(node_doc)
+        assert resp.status_code == 403, node_doc
+        up.assert_not_awaited()
+
+
+def test_upload_accepts_approved_node():
+    resp, up = _upload({"node_id": "n1", "approval_status": "approved"})
+    assert resp.status_code == 200
+    up.assert_awaited_once()
+
+
+def test_upload_wrong_key_still_401_before_approval_check():
+    resp, up = _upload({"node_id": "n1", "approval_status": "approved"}, key="wrong")
+    assert resp.status_code == 401
+    up.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# dynsec.reconcile_all
+# ---------------------------------------------------------------------------
+
+def test_reconcile_deletes_rejected_logins_and_still_syncs_approved():
+    async def collection_list(c, **filters):
+        status = filters.get("approval_status")
+        return {
+            "rejected": [{"node_id": "bad1"}, {"node_id": "bad2"}],
+            "approved": [{"node_id": "good"}],
+        }[status]
+
+    async def doc_get(c, i):
+        return {"node_id": i, "api_key": "k-" + i}
+
+    delete_client = AsyncMock(side_effect=[dynsec.DynsecError("boom"), None])
+    with patch.object(dynsec.fstore, "collection_list", collection_list), \
+         patch.object(dynsec.fstore, "doc_get", doc_get), \
+         patch.object(dynsec, "delete_node_client", delete_client), \
+         patch.object(dynsec, "upsert_node_client", AsyncMock()) as upsert:
+        asyncio.run(dynsec.reconcile_all())
+
+    # one failed revoke does not stop the rest, nor the approved sync
+    assert [c.args[0] for c in delete_client.await_args_list] == ["bad1", "bad2"]
+    upsert.assert_awaited_once_with("good", "k-good")
@@ -43,11 +43,14 @@ def test_normalize_camera_skips_disabled_blocked_and_zero_coords():
    assert ny511.normalize_camera({**CAM, "Latitude": 0, "Longitude": 0}) is None


-def test_normalize_event_parses_day_first_dates_and_truncates_description():
+def test_normalize_event_parses_day_first_dates_and_keeps_description_whole():
    e = ny511.normalize_event(EVENT)
    assert e["start_local"] == "2026-09-28T09:00:00"   # DD/MM, not MM/DD
    assert e["planned_end_local"] is None
-    assert len(e["description"]) == ny511._DESCRIPTION_MAX + 1
+    # DAA: "You may not alter the substance of a report" -- no 500-char cut any more.
+    assert e["description"] == EVENT["Description"]
+    long = ny511.normalize_event({**EVENT, "Description": "y" * (ny511._DESCRIPTION_MAX + 50)})
+    assert len(long["description"]) == ny511._DESCRIPTION_MAX + 1   # pathological only


 def test_failed_refresh_keeps_last_good_data_and_reports_error():
@@ -80,14 +83,29 @@ def test_schema_change_is_an_error_not_an_empty_layer():
            assert "schema" in str(e)


+def _serving(value: bool):
+    from app.internal.cameras import registry
+    return patch.object(registry, "is_serving", AsyncMock(return_value=value))
+
+
+def test_endpoint_is_off_until_the_ny511_provider_is_enabled():
+    with _serving(False):
+        r = client.get("/traffic/511", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4})
+    body = r.json()
+    assert r.status_code == 200 and body["cameras"] == [] and body["events"] == []
+    assert body["events_status"]["disabled"] is True
+
+
 def test_endpoint_filters_to_bbox_and_reports_status():
    far = {**CAM, "ID": "NYSDOT-2", "Latitude": 42.9, "Longitude": -78.8}   # Buffalo
    for feed, rows, norm in ((ny511.cameras, [CAM, far], ny511.normalize_camera), (ny511.events, [EVENT], ny511.normalize_event)):
        feed.items = [norm(r) for r in rows]
        feed.fetched_at, feed._next_attempt = 1.0, float("inf")
-    r = client.get("/traffic/511", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4})
+    with _serving(True):
+        r = client.get("/traffic/511", params={"south": 40.9, "west": -74.0, "north": 41.4, "east": -73.4})
    assert r.status_code == 200
    body = r.json()
+    assert body["credit"]["attribution"] == "powered by 511NY" and "as is" in body["credit"]["disclaimer"]
    assert [c["id"] for c in body["cameras"]] == ["NYSDOT-1"]
    assert body["cameras_status"] == {"fetched_at": 1.0, "error": None, "total_in_bbox": 1, "truncated": False}
    assert len(body["events"]) == 1
@@ -96,3 +114,24 @@ def test_endpoint_filters_to_bbox_and_reports_status():
 def test_endpoint_rejects_inverted_bbox():
    r = client.get("/traffic/511", params={"south": 41.4, "west": -74.0, "north": 40.9, "east": -73.4})
    assert r.status_code == 400
+
+
+def test_xml_inventory_maps_onto_legacy_fields():
+    xml = """<?xml version="1.0"?><cameras xmlns="urn:x">
+      <camera><ID>NYSDOT-9</ID><Latitude>41.05</Latitude><Longitude>-73.78</Longitude>
+        <Name>I-287 at Exit 4</Name><RoadwayName>I-287</RoadwayName><DirectionOfTravel>Eastbound</DirectionOfTravel>
+        <ImageUrl>https://example.invalid/still.jpg</ImageUrl><Disabled>false</Disabled></camera>
+      <camera id="NYSDOT-10" latitude="41.1" longitude="-73.7"><Disabled>true</Disabled></camera>
+    </cameras>"""
+    rows = ny511.parse_inventory_xml(xml)
+    cams = [ny511.normalize_camera(r) for r in rows]
+    assert cams[0]["id"] == "NYSDOT-9" and cams[0]["image_url"] == "https://example.invalid/still.jpg"
+    assert cams[1] is None   # disabled flag honoured from XML text
+
+
+def test_xml_parser_refuses_html_and_entities():
+    import pytest
+    with pytest.raises(ValueError, match="HTML"):
+        ny511.parse_inventory_xml("<!doctype html><html><body>511NY</body></html>")
+    with pytest.raises(ValueError, match="DTD"):
+        ny511.parse_inventory_xml('<?xml version="1.0"?><!DOCTYPE x [<!ENTITY a "b">]><x>&a;</x>')