mirror of
https://github.com/BigBodyCobain/Shadowbroker.git
synced 2026-06-05 13:58:15 +02:00
test(cctv): Madrid KML HTTPS-first fallback; clarify KiwiSDR #364 docs
Adds unit coverage for MadridCityIngestor catalog fetch order. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -37,10 +37,9 @@ _SOURCE_URL_HTTPS = "https://rx.linkfanel.net/kiwisdr_com.js"
|
||||
_CACHE_FILE = Path(__file__).resolve().parent.parent / "data" / "kiwisdr_cache.json"
|
||||
# Bundled fallback — shipped with the codebase so the KiwiSDR layer always
|
||||
# has something to render even when the upstream is unreachable, returns
|
||||
# garbage, or appears to have been tampered with. Issue #206: the upstream
|
||||
# only speaks HTTP, so we can't rely on TLS for integrity — instead we
|
||||
# validate the response's shape and fall back to this bundle if it doesn't
|
||||
# look right.
|
||||
# garbage, or appears to have been tampered with. Issue #206 / #364: try HTTPS
|
||||
# first, then HTTP; we still validate shape and fall back to this bundle if the
|
||||
# payload does not look right.
|
||||
_BUNDLED_FALLBACK = Path(__file__).resolve().parent.parent / "data" / "kiwisdr_directory.json"
|
||||
|
||||
# Minimum number of receivers we expect from a healthy upstream response.
|
||||
@@ -226,9 +225,8 @@ def _load_bundled_fallback() -> list[dict]:
|
||||
def fetch_kiwisdr_nodes() -> list[dict]:
|
||||
"""Return the KiwiSDR receiver list, refreshed at most once per day.
|
||||
|
||||
Layered fallback (issue #206 — upstream is HTTP-only, so we defend with
|
||||
content validation + bundled static directory rather than trying to
|
||||
upgrade the transport):
|
||||
Layered fallback (issue #206 / #364 — HTTPS first, HTTP fallback, plus
|
||||
content validation + bundled static directory):
|
||||
|
||||
1. In-memory cache (handled by @cached on this function)
|
||||
2. On-disk cache if <24h old
|
||||
|
||||
@@ -0,0 +1,27 @@
|
||||
"""Madrid CCTV KML prefers HTTPS (#363)."""
|
||||
from __future__ import annotations
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from services.cctv_pipeline import MadridCityIngestor
|
||||
|
||||
|
||||
def test_madrid_fetch_kml_tries_https_before_http():
|
||||
ingestor = MadridCityIngestor()
|
||||
calls: list[str] = []
|
||||
|
||||
def fake_fetch(url, **kwargs):
|
||||
calls.append(url)
|
||||
if url == ingestor.KML_URL_HTTPS:
|
||||
raise ConnectionError("tls handshake failed")
|
||||
res = MagicMock()
|
||||
res.status_code = 200
|
||||
res.content = b'<?xml version="1.0"?><kml xmlns="http://www.opengis.net/kml/2.2"></kml>'
|
||||
res.raise_for_status = MagicMock()
|
||||
return res
|
||||
|
||||
with patch("services.cctv_pipeline.fetch_with_curl", side_effect=fake_fetch):
|
||||
response = ingestor._fetch_kml()
|
||||
|
||||
assert response.status_code == 200
|
||||
assert calls == [ingestor.KML_URL_HTTPS, ingestor.KML_URL_HTTP]
|
||||
Reference in New Issue
Block a user