Logan CusanoandClaude Opus 5.5
2f21728bce
- Feed/catalog error text (shown to every signed-in user via GET /cameras and
/traffic/511 status fields) carried httpx's full request URL, including
?key= for Ontario 511 and the 511NY DAA key. Query strings are now dropped.
- TfL and Calgary licences are per GEV only (TfL's terms page 403'd), the same
footing as Ontario 511, so they are UNVERIFIED (acknowledgement required)
instead of ungated attribution_required. Attribution still renders.
- The map camera layer clamps its bbox to +/-180/+/-90; zoomed-out bounds
produced a 422 ("could not reach server").
Refs server-26#198
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
313 lines
16 KiB
Python
313 lines
16 KiB
Python
"""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
|