"""AZ-777 Phase 1 smoke test against the real parent-suite satellite-provider. Validates the wire C11 just plumbed through against the .NET service running inside `docker-compose.test.jetson.yml`: * TLS handshake works against the self-signed dev cert via ``SATELLITE_PROVIDER_TLS_INSECURE=1`` (development-only override — production deploys MUST use a CA-issued cert and unset the flag). * Bearer JWT minted from ``JWT_SECRET`` / ``JWT_ISSUER`` / ``JWT_AUDIENCE`` is accepted. * ``POST /api/satellite/tiles/inventory`` (AZ-505) round-trips the documented v1.0.0 schema for a 1-tile Derkachi-bbox query: response is 200 with a ``results`` array of length == request.tiles.length (Inv-2 from ``tile-inventory.md``) and per-entry shape matches the contract regardless of seed state. * The adapted C11 :class:`HttpTileDownloader` (Phase 1a) drives the same wire against the real service. When the catalog is unseeded (pre-Phase-2) the report is ``SUCCESS`` with zero downloads (every entry comes back ``present=false``); when seeded, the stubbed C6 ``write_tile_for_download`` receives one call per present tile and the report counts match. Phase 2 (Derkachi catalog seed via ``POST /api/satellite/request``) turns the second test from a "wire works" check into a "tiles actually land" check; the assertions here are written so both states (seeded / unseeded) pass cleanly so the smoke runs green as soon as Phase 1 lands and stays green through Phase 2. """ from __future__ import annotations import logging import os import ssl import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any import httpx import pytest from gps_denied_onboard.components.c11_tile_manager import ( C11Config, DownloadOutcome, DownloadRequest, HttpTileDownloader, SectorClassification, ) from gps_denied_onboard.helpers.wgs_converter import WgsConverter # ---------------------------------------------------------------------- # Skip gates # ---------------------------------------------------------------------- def _heavy_skip_reason() -> str | None: if os.environ.get("RUN_REPLAY_E2E", "").lower() not in {"1", "true", "yes", "on"}: return "AZ-777 satellite-provider smoke gated by RUN_REPLAY_E2E=1" return None _HEAVY_SKIP = pytest.mark.skipif( _heavy_skip_reason() is not None, reason=_heavy_skip_reason() or "ok", ) _INVENTORY_PATH = "/api/satellite/tiles/inventory" # Derkachi bbox center used for the 1-tile inventory query. Zoom 15 is # the upper bound of the AZ-777 Phase 2 catalog (zooms 15-18); picking # the lowest zoom keeps the per-tile ground footprint largest, which # makes a 1-tile query cover the most catalog. _DERKACHI_TILE_ZOOM = 15 _DERKACHI_LAT = 50.10 _DERKACHI_LON = 36.10 # Tight bbox that surrounds the Derkachi tile center so the C11 bbox # enumeration produces exactly one (z,x,y) coord at zoom 15. _DERKACHI_BBOX = (50.099, 36.099, 50.101, 36.101) # ---------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------- def _mint_bearer_token_or_skip() -> str: """Return a valid Bearer JWT or `pytest.skip` with an actionable reason.""" token = os.environ.get("SATELLITE_PROVIDER_API_KEY", "").strip() if token and token != "PASTE-MINTED-JWT-HERE": return token try: import jwt # type: ignore[import-untyped] except ImportError: pytest.skip( "SATELLITE_PROVIDER_API_KEY is unset and PyJWT is not " "installed in-container; install dev extras " "(`pip install -e .[dev]`) or set " "SATELLITE_PROVIDER_API_KEY via .env.test." ) secret = os.environ.get("JWT_SECRET", "").strip() issuer = os.environ.get("JWT_ISSUER", "").strip() audience = os.environ.get("JWT_AUDIENCE", "").strip() missing = [ name for name, value in [ ("JWT_SECRET", secret), ("JWT_ISSUER", issuer), ("JWT_AUDIENCE", audience), ] if not value ] if missing: pytest.skip( "Cannot mint a fallback JWT — missing env: " + ", ".join(missing) ) now = datetime.now(timezone.utc) payload = { "sub": "gps-denied-onboard-smoke", "iss": issuer, "aud": audience, "jti": uuid.uuid4().hex, "iat": int(now.timestamp()), "nbf": int(now.timestamp()), "exp": int((now + timedelta(hours=1)).timestamp()), } return str(jwt.encode(payload, secret, algorithm="HS256")) def _make_http_client(base_url: str) -> httpx.Client: """Return an httpx.Client honouring SATELLITE_PROVIDER_TLS_INSECURE.""" insecure_raw = os.environ.get("SATELLITE_PROVIDER_TLS_INSECURE", "").strip() insecure = insecure_raw.lower() in {"1", "true", "yes", "on"} verify: bool | ssl.SSLContext if insecure: # Documented dev-only path; the audit-trail WARNING lives in # `.env.test.example`. We do NOT silently disable verification # unless the env var explicitly opts in. verify = False else: verify = True return httpx.Client(base_url=base_url, verify=verify, timeout=30.0) def _resolve_satellite_provider_url() -> str: url = os.environ.get("SATELLITE_PROVIDER_URL", "").strip() if not url: pytest.skip( "SATELLITE_PROVIDER_URL is not set — the Jetson e2e compose " "env block provides https://satellite-provider:8080; running " "this smoke outside compose requires an explicit override." ) return url # ---------------------------------------------------------------------- # Test 1 — inventory POST contract (AZ-505 v1.0.0) # ---------------------------------------------------------------------- @pytest.mark.tier2 @_HEAVY_SKIP def test_smoke_satellite_provider_inventory_contract() -> None: # Arrange base_url = _resolve_satellite_provider_url() bearer = _mint_bearer_token_or_skip() tile_x, tile_y = WgsConverter.latlon_to_tile_xy( _DERKACHI_TILE_ZOOM, _DERKACHI_LAT, _DERKACHI_LON ) body = { "tiles": [ { "tileZoom": _DERKACHI_TILE_ZOOM, "tileX": tile_x, "tileY": tile_y, } ] } # Act with _make_http_client(base_url) as client: response = client.post( _INVENTORY_PATH, json=body, headers={"Authorization": f"Bearer {bearer}"}, ) # Assert — contract invariants from `tile-inventory.md` v1.0.0 assert response.status_code == 200, ( f"satellite-provider inventory POST returned " f"{response.status_code}: {response.text!r}" ) decoded = response.json() assert isinstance(decoded, dict), f"expected JSON object, got {type(decoded)}" assert "results" in decoded, f"missing 'results' key in {decoded!r}" results = decoded["results"] assert isinstance(results, list) assert len(results) == len(body["tiles"]), ( # Inv-2: response order/length matches request order/length. f"inventory response length {len(results)} != request length " f"{len(body['tiles'])}" ) entry = results[0] assert entry["tileZoom"] == _DERKACHI_TILE_ZOOM assert entry["tileX"] == tile_x assert entry["tileY"] == tile_y assert "present" in entry, f"missing 'present' in entry {entry!r}" assert isinstance(entry["present"], bool) if entry["present"]: for required in ("id", "capturedAt", "source", "resolutionMPerPx"): assert entry.get(required) is not None, ( f"present entry missing required field {required!r}: {entry!r}" ) # ---------------------------------------------------------------------- # Test 2 — C11 HttpTileDownloader against the real service # ---------------------------------------------------------------------- class _InMemoryC6Adapter: """Implements `_TileWriterLike` + `_BudgetEnforcerLike` in-memory. The Phase 1 smoke does NOT exercise the real C6 store (that's Phase 3 of AZ-777). It exercises the C11 wire end-to-end with a stub C6 so the test stays scope-clean. """ def __init__(self) -> None: self.write_calls: list[dict[str, Any]] = [] self.reserved_bytes: list[int] = [] 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: self.write_calls.append( { "zoom_level": zoom_level, "lat": lat, "lon": lon, "tile_blob_len": len(tile_blob), "sector_class": sector_class, } ) return "fresh" def tile_already_present( self, *, zoom_level: int, lat: float, lon: float ) -> bool: self.exists_calls.append((zoom_level, lat, lon)) return False def reserve_headroom(self, needed_bytes: int) -> object: self.reserved_bytes.append(int(needed_bytes)) return object() @pytest.mark.tier2 @_HEAVY_SKIP def test_smoke_c11_download_via_http_pipeline(tmp_path: Path) -> None: # Arrange base_url = _resolve_satellite_provider_url() bearer = _mint_bearer_token_or_skip() adapter = _InMemoryC6Adapter() cfg = C11Config( satellite_provider_url=base_url, service_api_key=bearer, download_http_timeout_s=30.0, download_max_5xx_retries=2, download_max_retry_after_s=60, download_resolution_floor_m_per_px=0.5, ) logger = logging.getLogger("test_az777_smoke") with _make_http_client(base_url) as http_client: downloader = HttpTileDownloader( http_client=http_client, tile_writer=adapter, # type: ignore[arg-type] budget_enforcer=adapter, # type: ignore[arg-type] logger=logger, config=cfg, ) request = DownloadRequest( flight_id=uuid.uuid4(), bbox_min_lat=_DERKACHI_BBOX[0], bbox_min_lon=_DERKACHI_BBOX[1], bbox_max_lat=_DERKACHI_BBOX[2], bbox_max_lon=_DERKACHI_BBOX[3], zoom_levels=(_DERKACHI_TILE_ZOOM,), sector_class=SectorClassification.STABLE_REAR, cache_root=tmp_path, ) # Act report = downloader.download_tiles_for_area(request) # Assert — Phase 1 smoke contract: the wire works regardless of # catalog state. Pre-Phase-2 the catalog is empty → 0 downloads. # Post-Phase-2 the catalog is seeded → ≥ 1 download. Both are valid # outcomes for this test; the test fails only on a wire / contract # break (any provider error would have raised before reaching this # assertion block). assert report.outcome == DownloadOutcome.SUCCESS assert report.tiles_requested >= 0 assert report.tiles_downloaded == len(adapter.write_calls) assert report.tiles_rejected_resolution >= 0 if report.tiles_downloaded > 0: # Phase 2 catalog landed: verify the write actually carried the # tile bytes we'd expect from a real GET /tiles/{z}/{x}/{y}. assert all(call["tile_blob_len"] > 0 for call in adapter.write_calls) assert all(call["zoom_level"] == _DERKACHI_TILE_ZOOM for call in adapter.write_calls)