mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 23:01:13 +00:00
Oleksandr Bezdieniezhnykh
5def1a3eb3
Implement the AC-8.4 and AC-NEW-6 blackbox scenarios for mid-flight tile generation, dedup, landing-time upload, and freshness gating. Helpers: - runner/helpers/mid_flight_tile_evaluator.py — pure-logic evaluators for tile generation rate, Mode B Fact #105 schema check, footprint+ GSD dedup (via geo.distance_m), upload-audit reconciliation, and the AC-5/AC-6 capture_utc + freshness-gate checks. - runner/helpers/mock_suite_sat_audit.py — httpx wrapper for the mock-suite-sat-service /tiles/audit endpoint with strict response- shape validation. Scenarios: - tests/positive/test_ft_p_17_mid_flight_tiles.py - tests/negative/test_ft_n_06_mid_flight_freshness.py Both skip when sitl_replay_ready is false and fail loudly when fixture records are missing (tests-as-gates discipline). 52 new unit tests (41 evaluator + 11 audit client) cover every helper branch. Review: PASS_WITH_WARNINGS (2 Low — duplicate haversine carry-over, upstream production dependency surface). Co-authored-by: Cursor <cursoragent@cursor.com>
77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
"""HTTP client for the mock-suite-sat-service audit log.
|
|
|
|
Thin wrapper over ``GET /tiles/audit`` (and its alias ``GET /mock/audit``)
|
|
that the FT-P-17 scenario uses to verify every generated tile was
|
|
accepted with HTTP 202 at landing-time upload.
|
|
|
|
Reading the audit log is a one-shot, end-of-run operation, so the
|
|
helper is synchronous httpx — no streaming, no retries (the service is
|
|
co-located in the compose harness and a failure to reach it is itself
|
|
a test failure, not something to paper over).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
DEFAULT_TIMEOUT_S = 10.0
|
|
DEFAULT_AUDIT_PATH = "/tiles/audit"
|
|
|
|
|
|
def fetch_audit(
|
|
base_url: str,
|
|
run_id: str,
|
|
*,
|
|
timeout_s: float = DEFAULT_TIMEOUT_S,
|
|
audit_path: str = DEFAULT_AUDIT_PATH,
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> list[dict[str, Any]]:
|
|
"""Return the ``entries`` list from the mock-suite-sat-service audit log.
|
|
|
|
The endpoint returns ``{"run_id": str, "entries": [...]}``; this
|
|
helper unwraps the list. An empty list is a legal response (the
|
|
SUT may not have uploaded anything yet).
|
|
|
|
Raises ``RuntimeError`` on non-2xx HTTP status or a malformed
|
|
response shape — the scenario test wants those to fail loudly.
|
|
|
|
``transport`` is a unit-test seam: pass an
|
|
``httpx.MockTransport`` to feed canned responses without spinning
|
|
up the real service.
|
|
"""
|
|
if not base_url:
|
|
raise RuntimeError("fetch_audit: base_url must be a non-empty string")
|
|
if not run_id:
|
|
raise RuntimeError("fetch_audit: run_id must be a non-empty string")
|
|
url = base_url.rstrip("/") + audit_path
|
|
client_kwargs: dict[str, Any] = {"timeout": timeout_s}
|
|
if transport is not None:
|
|
client_kwargs["transport"] = transport
|
|
with httpx.Client(**client_kwargs) as client:
|
|
resp = client.get(url, params={"run_id": run_id})
|
|
if resp.status_code >= 300:
|
|
raise RuntimeError(
|
|
f"fetch_audit: {url}?run_id={run_id} returned HTTP "
|
|
f"{resp.status_code}: body={resp.text[:200]!r}"
|
|
)
|
|
try:
|
|
body = resp.json()
|
|
except ValueError as exc:
|
|
raise RuntimeError(
|
|
f"fetch_audit: {url}?run_id={run_id} body is not valid JSON: {exc}"
|
|
) from exc
|
|
if not isinstance(body, dict):
|
|
raise RuntimeError(
|
|
f"fetch_audit: {url}?run_id={run_id} body is not a JSON object: "
|
|
f"got {type(body).__name__}"
|
|
)
|
|
entries = body.get("entries")
|
|
if not isinstance(entries, list):
|
|
raise RuntimeError(
|
|
f"fetch_audit: {url}?run_id={run_id} body missing 'entries' list: "
|
|
f"keys={list(body.keys())}"
|
|
)
|
|
return entries
|