azaion/gps-denied-onboard
1
0
Fork 0
mirror of https://github.com/azaion/gps-denied-onboard.git synced 2026-06-21 21:51:13 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
4f0d8bdcd9fe71e3e9c243cb711cc50c3f8976dd
gps-denied-onboard/tests/unit/c11_tile_manager/test_tile_downloader.py
T
Oleksandr Bezdieniezhnykh b15454b9a9 [AZ-777] Phase 1 hotfix (z/x/y) + Phase 2 Derkachi seed + ops 
Phase 1 hotfix:
- C11 HttpTileDownloader adapted to satellite-provider v2.0.0
  z/x/y inventory contract (bulk POST keyed by slippy-map coords).
- Unit tests rewritten to exercise the new inventory schema.
- E2E smoke test updated to match the v2.0.0 wire.

Phase 2 (Derkachi seed + smoke-validated on Jetson):
- tests/fixtures/derkachi_c6/{README,bbox.yaml,seed_region.py}
  drives POST /api/satellite/region against satellite-provider
  with Google Maps as the imagery source. Smoke run produced
  4 regions, 175 tiles, inventory 32/32.
- scripts/mint_dev_jwt.py + run-tests-jetson.sh auto-mint and
  export SATELLITE_PROVIDER_API_KEY using JWT_SECRET / JWT_ISSUER
  / JWT_AUDIENCE env vars (no host port mappings; e2e-runner
  reaches SP via internal docker network only).

Spec amendment: AZ-777 todo spec updated to record the
Google Maps imagery source decision and STOP-gate state.

AZ-777 Phase 3+ work is superseded by Epic AZ-835 (see next
commit).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-22 17:39:21 +03:00

776 lines
27 KiB
Python
Raw Blame History

"""AZ-316 / AZ-777 ``HttpTileDownloader`` unit tests.
Covers AC-1 .. AC-12 plus the throughput NFR from
``_docs/02_tasks/todo/AZ-316_c11_tile_downloader.md``, against the
AZ-777 contract update (POST ``/api/satellite/tiles/inventory`` for
bulk lookup + GET ``/tiles/{z}/{x}/{y}`` for body download — see
``../satellite-provider/_docs/02_document/contracts/api/tile-inventory.md``
v1.0.0). Uses :class:`httpx.MockTransport` for deterministic HTTP
responses, a list-backed log handler for log capture, and stub C6
stores so this suite never depends on AZ-303 / AZ-305 / AZ-307 /
AZ-308 internals.
"""
from __future__ import annotations
import json
import logging
import math
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any
from uuid import uuid4
import httpx
import pytest
from gps_denied_onboard.components.c11_tile_manager import (
C11Config,
CacheBudgetExceededError,
DownloadOutcome,
DownloadRequest,
HttpTileDownloader,
RateLimitedError,
SatelliteProviderError,
SectorClassification,
request_hash,
)
_BASE_URL = "https://parent-suite.test"
_INVENTORY_PATH = "/api/satellite/tiles/inventory"
_TILES_PATH_PREFIX = "/tiles/"
_API_KEY = "test-api-key-001"
# Mirror of c11.tile_downloader._DEFAULT_ESTIMATED_TILE_BYTES (AZ-777):
# the inventory contract no longer returns content-length hints, so the
# AZ-308 budget pre-check reserves this constant per `present=true`
# tile. Keep in sync with the production module.
_DEFAULT_ESTIMATED_TILE_BYTES = 50_000
# ----------------------------------------------------------------------
# Stubs / fakes
# ----------------------------------------------------------------------
class _StubFreshnessRejection(Exception):
"""Composition-root would surface c6's ``FreshnessRejectionError``;
we mirror it locally so the structural check in the downloader
matches by class name without importing c6.
"""
# Re-name the class so the structural check (`__class__.__name__ ==
# "FreshnessRejectionError"`) inside the downloader matches.
_StubFreshnessRejection.__name__ = "FreshnessRejectionError"
class _StubTileWriter:
"""Captures `write_tile_for_download` calls + scripts the freshness label.
AZ-777: keyed by *call index* rather than `(z, lat, lon)` strings.
The downloader now derives (lat, lon) from the slippy-map coord, so
tests can no longer fabricate arbitrary lat/lons in their fixtures;
the call-order is the contract.
"""
def __init__(
self,
*,
labels_by_index: dict[int, str] | None = None,
rejected_indices: set[int] | None = None,
) -> None:
self.labels_by_index = labels_by_index or {}
self.rejected_indices = rejected_indices or set()
self.write_calls: list[dict[str, Any]] = []
self.exists_calls: list[tuple[int, float, float]] = []
def write_tile_for_download(
self,
*,
tile_blob: bytes,
zoom_level: int,
lat: float,
lon: float,
tile_size_meters: float,
tile_size_pixels: int,
capture_timestamp: datetime,
content_sha256_hex: str,
sector_class: str,
) -> str:
call_index = len(self.write_calls)
self.write_calls.append(
{
"call_index": call_index,
"zoom_level": zoom_level,
"lat": lat,
"lon": lon,
"tile_blob_len": len(tile_blob),
"content_sha256_hex": content_sha256_hex,
"sector_class": sector_class,
}
)
if call_index in self.rejected_indices:
raise _StubFreshnessRejection(f"freshness rejected call_index={call_index}")
return self.labels_by_index.get(call_index, "fresh")
def tile_already_present(self, *, zoom_level: int, lat: float, lon: float) -> bool:
self.exists_calls.append((zoom_level, lat, lon))
return False
class _StubBudgetEnforcer:
"""Records `reserve_headroom` calls; raises pre-baked exception when set."""
def __init__(self, raise_on_call: Exception | None = None) -> None:
self.calls: list[int] = []
self._raise = raise_on_call
def reserve_headroom(self, needed_bytes: int) -> object:
self.calls.append(int(needed_bytes))
if self._raise is not None:
raise self._raise
return object()
def _build_downloader(
*,
transport: httpx.MockTransport,
tile_writer: _StubTileWriter | None = None,
budget_enforcer: _StubBudgetEnforcer | None = None,
config: C11Config | None = None,
sleep_recorder: list[float] | None = None,
backoff_schedule_s: tuple[float, ...] | None = None,
) -> tuple[
HttpTileDownloader,
list[logging.LogRecord],
_StubTileWriter,
_StubBudgetEnforcer,
list[float],
]:
log_records: list[logging.LogRecord] = []
class _Handler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
log_records.append(record)
logger = logging.getLogger(f"test_az316_{id(log_records)}")
logger.handlers.clear()
logger.addHandler(_Handler())
logger.setLevel(logging.DEBUG)
logger.propagate = False
writer = tile_writer or _StubTileWriter()
enforcer = budget_enforcer or _StubBudgetEnforcer()
sleeps = sleep_recorder if sleep_recorder is not None else []
cfg = config or C11Config(
satellite_provider_url=_BASE_URL,
service_api_key=_API_KEY,
download_http_timeout_s=5.0,
download_max_5xx_retries=4,
download_max_retry_after_s=600,
download_resolution_floor_m_per_px=0.5,
)
client = httpx.Client(transport=transport, base_url=_BASE_URL)
downloader = HttpTileDownloader(
http_client=client,
tile_writer=writer, # type: ignore[arg-type]
budget_enforcer=enforcer, # type: ignore[arg-type]
logger=logger,
config=cfg,
sleep=lambda s: sleeps.append(s),
backoff_schedule_s=backoff_schedule_s,
)
return downloader, log_records, writer, enforcer, sleeps
_DEFAULT_TEST_BBOX = (45.0, -122.5, 45.5, -122.0)
def _make_request(
*,
flight_id: Any | None = None,
cache_root: Path,
zoom_levels: tuple[int, ...] = (14,),
bbox: tuple[float, float, float, float] = _DEFAULT_TEST_BBOX,
) -> DownloadRequest:
return DownloadRequest(
flight_id=flight_id or uuid4(),
bbox_min_lat=bbox[0],
bbox_min_lon=bbox[1],
bbox_max_lat=bbox[2],
bbox_max_lon=bbox[3],
zoom_levels=zoom_levels,
sector_class=SectorClassification.STABLE_REAR,
cache_root=cache_root,
)
def _tile_center_latlon_from_zxy(zoom: int, x: int, y: int) -> tuple[float, float]:
"""Mirror of c11.tile_downloader._tile_center_latlon for test assertions.
Both sides compute lat/lon from the slippy-map (z, x, y) tuple, so
stub freshness/label maps can be keyed on the *expected* lat/lon
without depending on the production helper at import time.
"""
n = 1 << int(zoom)
lat_n = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * int(y) / n))))
lat_s = math.degrees(math.atan(math.sinh(math.pi * (1.0 - 2.0 * (int(y) + 1) / n))))
lon_w = int(x) / n * 360.0 - 180.0
lon_e = (int(x) + 1) / n * 360.0 - 180.0
return (lat_n + lat_s) / 2.0, (lon_w + lon_e) / 2.0
def _inventory_entry_for_coord(
*,
zoom: int,
x: int,
y: int,
present: bool = True,
resolution_m_per_px: float = 0.5,
captured_at: datetime | None = None,
) -> dict[str, Any]:
if not present:
return {
"z": int(zoom),
"x": int(x),
"y": int(y),
"locationHash": str(uuid4()),
"present": False,
"id": None,
"capturedAt": None,
"source": None,
"flightId": None,
"resolutionMPerPx": None,
}
captured = captured_at or datetime(2026, 5, 13, 0, 0, 0, tzinfo=timezone.utc)
return {
"z": int(zoom),
"x": int(x),
"y": int(y),
"locationHash": str(uuid4()),
"present": True,
"id": str(uuid4()),
"capturedAt": captured.isoformat(),
"source": "google_maps",
"flightId": None,
"resolutionMPerPx": float(resolution_m_per_px),
}
def _make_inventory_handler(
*,
present_count: int | None = None,
resolution_override_for_first_n: tuple[int, float] | None = None,
tile_response_factory: Any = None,
inventory_response_override: httpx.Response | None = None,
) -> Any:
"""Route requests by method+path: POST inventory vs GET /tiles/{z}/{x}/{y}.
The inventory handler echoes the request's tile coords in
response order (per contract invariant Inv-2 / Inv-3). The
`present_count` knob marks the FIRST N entries as
``present=true`` and the rest as ``present=false``; ``None`` means
"all present". `resolution_override_for_first_n=(K, RES)` overrides
the resolution of the first K present entries (used by the AZ-316
resolution-gate test).
"""
def _handler(request: httpx.Request) -> httpx.Response:
path = request.url.path
method = request.method.upper()
if method == "POST" and path.endswith(_INVENTORY_PATH):
if inventory_response_override is not None:
return inventory_response_override
body = json.loads(request.content.decode("utf-8"))
tiles_in = body["tiles"]
results: list[dict[str, Any]] = []
for i, t in enumerate(tiles_in):
is_present = present_count is None or i < int(present_count)
if (
is_present
and resolution_override_for_first_n is not None
and i < int(resolution_override_for_first_n[0])
):
resolution = float(resolution_override_for_first_n[1])
else:
resolution = 0.5
results.append(
_inventory_entry_for_coord(
zoom=int(t["z"]),
x=int(t["x"]),
y=int(t["y"]),
present=is_present,
resolution_m_per_px=resolution,
)
)
return httpx.Response(200, json={"results": results})
if method == "GET" and path.startswith(_TILES_PATH_PREFIX):
if tile_response_factory is None:
return httpx.Response(200, content=b"\xff\xd8\xff\xe0fake-jpeg")
return tile_response_factory(request)
return httpx.Response(
404,
json={"detail": f"test handler: unexpected {method} {path}"},
)
return _handler
# ----------------------------------------------------------------------
# AC-1: 100-tile happy path
# ----------------------------------------------------------------------
def test_ac1_100_tile_happy_path_writes_all(tmp_path: Path) -> None:
# Arrange — bbox at zoom 14 produces N coord candidates; the
# stub marks the first 100 as `present=true` and the rest absent.
transport = httpx.MockTransport(_make_inventory_handler(present_count=100))
(downloader, _logs, writer, enforcer, _sleeps) = _build_downloader(transport=transport)
request = _make_request(cache_root=tmp_path)
# Act
report = downloader.download_tiles_for_area(request)
# Assert
assert report.outcome == DownloadOutcome.SUCCESS
assert report.tiles_requested == 100
assert report.tiles_downloaded == 100
assert report.tiles_rejected_resolution == 0
assert report.tiles_rejected_freshness == 0
assert report.tiles_downgraded == 0
assert len(writer.write_calls) == 100
assert enforcer.calls == [_DEFAULT_ESTIMATED_TILE_BYTES * 100]
# ----------------------------------------------------------------------
# AC-2: resolution gate rejects sub-spec tiles BEFORE write
# ----------------------------------------------------------------------
def test_ac2_resolution_gate_rejects_sub_spec_tiles(tmp_path: Path) -> None:
# Arrange — 50 present tiles; first 10 below the 0.5 m/px floor.
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=50,
resolution_override_for_first_n=(10, 0.3),
)
)
(downloader, log_records, writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
# Act
report = downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
assert report.tiles_rejected_resolution == 10
assert report.tiles_downloaded == 40
assert len(writer.write_calls) == 40
res_warnings = [
r for r in log_records if getattr(r, "kind", "") == "c11.download.resolution_rejected"
]
assert len(res_warnings) == 10
# ----------------------------------------------------------------------
# AC-3: c6 freshness rejection counted, not propagated
# ----------------------------------------------------------------------
def test_ac3_freshness_rejections_counted_and_run_continues(tmp_path: Path) -> None:
# Arrange — 10 present tiles; c6 rejects the first 5 writes.
transport = httpx.MockTransport(_make_inventory_handler(present_count=10))
writer = _StubTileWriter(rejected_indices={0, 1, 2, 3, 4})
(downloader, log_records, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, tile_writer=writer
)
# Act
report = downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
assert report.tiles_rejected_freshness == 5
assert report.tiles_downloaded == 5
assert report.outcome == DownloadOutcome.SUCCESS
summary_warns = [
r
for r in log_records
if getattr(r, "kind", "") == "c11.download.freshness_rejected_summary"
]
assert len(summary_warns) == 1
# ----------------------------------------------------------------------
# AC-4: stable_rear stale tiles are surfaced as DOWNGRADED
# ----------------------------------------------------------------------
def test_ac4_downgraded_label_counted_and_persisted(tmp_path: Path) -> None:
# Arrange — 5 present tiles; c6 returns "downgraded" for the first 3 writes.
transport = httpx.MockTransport(_make_inventory_handler(present_count=5))
writer = _StubTileWriter(labels_by_index={0: "downgraded", 1: "downgraded", 2: "downgraded"})
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, tile_writer=writer
)
# Act
report = downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
assert report.tiles_downgraded == 3
assert report.tiles_downloaded == 5
# ----------------------------------------------------------------------
# AC-5: 429 honours Retry-After
# ----------------------------------------------------------------------
def test_ac5_429_honours_retry_after(tmp_path: Path) -> None:
# Arrange — single present tile; tile GET returns 429 once then 200.
state = {"attempts": 0}
def _factory(request: httpx.Request) -> httpx.Response:
state["attempts"] += 1
if state["attempts"] == 1:
return httpx.Response(429, headers={"Retry-After": "30"})
return httpx.Response(200, content=b"\xff\xd8tile")
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
sleeps: list[float] = []
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, sleep_recorder=sleeps
)
# Act
report = downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
assert state["attempts"] == 2
assert sleeps and sleeps[0] >= 30
assert report.retry_count >= 1
assert report.outcome == DownloadOutcome.SUCCESS
# ----------------------------------------------------------------------
# AC-6: persistent 5xx aborts with structured error
# ----------------------------------------------------------------------
def test_ac6_persistent_5xx_raises_satellite_provider_error(tmp_path: Path) -> None:
# Arrange — single present tile; tile GET always 503.
state = {"attempts": 0}
def _factory(request: httpx.Request) -> httpx.Response:
state["attempts"] += 1
return httpx.Response(503)
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
# Act / Assert
with pytest.raises(SatelliteProviderError):
downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
assert state["attempts"] >= 5
# ----------------------------------------------------------------------
# AC-7: 401 fails fast (no retry)
# ----------------------------------------------------------------------
def test_ac7_401_fails_fast_no_retry(tmp_path: Path) -> None:
# Arrange — single present tile; tile GET returns 401.
state = {"attempts": 0}
def _factory(request: httpx.Request) -> httpx.Response:
state["attempts"] += 1
return httpx.Response(401)
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
(downloader, _log_records, _writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
# Act / Assert
with pytest.raises(SatelliteProviderError):
downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
assert state["attempts"] == 1
# ----------------------------------------------------------------------
# AC-8: idempotent re-run after success
# ----------------------------------------------------------------------
def test_ac8_idempotent_rerun_yields_no_op(tmp_path: Path) -> None:
# Arrange — 5 present tiles; both runs reach the same handler.
state = {"tile_gets": 0}
def _factory(request: httpx.Request) -> httpx.Response:
state["tile_gets"] += 1
return httpx.Response(200, content=b"\xff\xd8tile")
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=5,
tile_response_factory=_factory,
)
)
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
request = _make_request(cache_root=tmp_path)
# Act — first run
first = downloader.download_tiles_for_area(request)
first_get_count = state["tile_gets"]
# Act — second run (same request, same cache_root → idempotent)
second = downloader.download_tiles_for_area(request)
# Assert
assert first.outcome == DownloadOutcome.SUCCESS
assert second.outcome == DownloadOutcome.IDEMPOTENT_NO_OP
assert state["tile_gets"] == first_get_count, "second run must NOT fetch"
assert second.tiles_downloaded == first.tiles_downloaded
# ----------------------------------------------------------------------
# AC-9: cache-budget pre-check aborts before any GET
# ----------------------------------------------------------------------
def test_ac9_cache_budget_pre_check_aborts(tmp_path: Path) -> None:
# Arrange — single present tile; budget enforcer rejects up-front.
transport_state = {"tile_gets": 0}
def _factory(request: httpx.Request) -> httpx.Response:
transport_state["tile_gets"] += 1
return httpx.Response(200, content=b"\xff\xd8tile")
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
enforcer = _StubBudgetEnforcer(raise_on_call=CacheBudgetExceededError("no headroom"))
(downloader, _log_records, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, budget_enforcer=enforcer
)
# Act / Assert
with pytest.raises(CacheBudgetExceededError):
downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
assert transport_state["tile_gets"] == 0
assert enforcer.calls == [_DEFAULT_ESTIMATED_TILE_BYTES]
# ----------------------------------------------------------------------
# AC-11: service API key never logged in plaintext
# ----------------------------------------------------------------------
def test_ac11_service_api_key_never_appears_in_logs(tmp_path: Path) -> None:
# Arrange — exercise the failure path so the provider-failed ERROR
# log fires (the code that explicitly redacts the auth header). The
# inventory POST returns 401 → fast-fail before any tile GET.
transport = httpx.MockTransport(
_make_inventory_handler(
inventory_response_override=httpx.Response(401),
)
)
(downloader, log_records, _writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
with pytest.raises(SatelliteProviderError):
downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
flat = " ".join(r.getMessage() + json.dumps(getattr(r, "kv", {})) for r in log_records)
assert _API_KEY not in flat
assert "Bearer ***" in flat
# ----------------------------------------------------------------------
# AC-12: journal survives mid-batch crash; re-run completes the rest
# ----------------------------------------------------------------------
def test_ac12_partial_journal_resumed_on_rerun(tmp_path: Path) -> None:
# Arrange — 10 present tiles; first run completes, then we truncate
# the journal to simulate a crash after 4 writes (clear
# `completed_at_iso`) so the second run must complete the remaining
# 6 tiles instead of short-circuiting on AC-8 idempotence.
state = {"tile_gets": 0}
def _factory(request: httpx.Request) -> httpx.Response:
state["tile_gets"] += 1
return httpx.Response(200, content=b"\xff\xd8tile")
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=10,
tile_response_factory=_factory,
)
)
(downloader, _logs, writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
request = _make_request(cache_root=tmp_path)
# First run — completes
downloader.download_tiles_for_area(request)
after_first = state["tile_gets"]
write_calls_after_first = len(writer.write_calls)
assert after_first == 10
assert write_calls_after_first == 10
# Simulate crash: rewrite the journal as "completed 4 of 10 tiles,
# NOT yet completed" (clear `completed_at_iso`).
rh = request_hash(
request.flight_id,
request.bbox_min_lat,
request.bbox_min_lon,
request.bbox_max_lat,
request.bbox_max_lon,
tuple(request.zoom_levels),
request.sector_class,
_API_KEY,
)
journal_path = tmp_path / ".c11/journal" / f"{request.flight_id}__{rh}.json"
raw = json.loads(journal_path.read_text("utf-8"))
raw["completed_at_iso"] = None
raw["tile_ids_completed"] = sorted(raw["tile_ids_completed"])[:4]
raw["tile_counts"]["tiles_downloaded"] = 4
journal_path.write_text(json.dumps(raw), encoding="utf-8")
# Reset transport counter (writer still records cumulative calls)
state["tile_gets"] = 0
writer.write_calls.clear()
# Act — second run must fetch only the missing 6
second = downloader.download_tiles_for_area(request)
# Assert
assert state["tile_gets"] == 6
assert len(writer.write_calls) == 6
assert second.outcome == DownloadOutcome.SUCCESS
assert second.tiles_downloaded == 6
# ----------------------------------------------------------------------
# Retry-After HTTP-date form (Risk 1 from the spec)
# ----------------------------------------------------------------------
def test_429_retry_after_http_date_form_parses(tmp_path: Path) -> None:
# Arrange — single present tile; first tile GET 429 with HTTP-date Retry-After.
state = {"attempts": 0}
future = (datetime.now(timezone.utc) + timedelta(seconds=20)).strftime(
"%a, %d %b %Y %H:%M:%S GMT"
)
def _factory(request: httpx.Request) -> httpx.Response:
state["attempts"] += 1
if state["attempts"] == 1:
return httpx.Response(429, headers={"Retry-After": future})
return httpx.Response(200, content=b"\xff\xd8tile")
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
sleeps: list[float] = []
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, sleep_recorder=sleeps
)
# Act
report = downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# Assert
assert state["attempts"] == 2
assert sleeps and sleeps[0] >= 0
assert report.outcome == DownloadOutcome.SUCCESS
# ----------------------------------------------------------------------
# 429 budget exhaustion → RateLimitedError
# ----------------------------------------------------------------------
def test_429_budget_exhaustion_raises_rate_limited_error(tmp_path: Path) -> None:
# Arrange — single present tile; tile GET always 429 with very long
# Retry-After. After the configured retry budget is exhausted the
# client raises RateLimitedError.
def _factory(request: httpx.Request) -> httpx.Response:
return httpx.Response(429, headers={"Retry-After": "300"})
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1,
tile_response_factory=_factory,
)
)
cfg = C11Config(
satellite_provider_url=_BASE_URL,
service_api_key=_API_KEY,
download_http_timeout_s=5.0,
download_max_5xx_retries=4,
download_max_retry_after_s=400,
download_resolution_floor_m_per_px=0.5,
)
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(
transport=transport, config=cfg
)
# Act / Assert
with pytest.raises(RateLimitedError):
downloader.download_tiles_for_area(_make_request(cache_root=tmp_path))
# ----------------------------------------------------------------------
# NFR — bookkeeping throughput on a 1000-tile happy path
# ----------------------------------------------------------------------
def test_nfr_throughput_1000_tiles_under_budget(tmp_path: Path) -> None:
# Arrange — 1000 present tiles. Use a bbox wide enough to enumerate
# ≥1000 tile coords at zoom 14 (≈0.022° per tile at ~zoom 14).
transport = httpx.MockTransport(
_make_inventory_handler(
present_count=1000,
tile_response_factory=lambda r: httpx.Response(200, content=b"\xff\xd8tile"),
)
)
(downloader, _logs, _writer, _enforcer, _sleeps) = _build_downloader(transport=transport)
# Big bbox at zoom 14: ~ 32x32 tile span on this latitude is enough.
# 1° ≈ 45 tiles at zoom 14 in latitude → 0.75° gives ≈ 33 tiles → ~1089 tiles.
request = _make_request(
cache_root=tmp_path,
bbox=(44.0, -123.0, 44.75, -122.25),
)
import time as _time
t0 = _time.perf_counter()
report = downloader.download_tiles_for_area(request)
elapsed = _time.perf_counter() - t0
# Assert — budget is generous; the goal is to catch an O(n^2)
# bookkeeping regression, not to certify wall-clock throughput.
assert report.outcome == DownloadOutcome.SUCCESS
assert report.tiles_downloaded == 1000
assert elapsed < 10.0, f"1000-tile bookkeeping took {elapsed:.2f}s"
Reference in New Issue View Git Blame Copy Permalink