From 5def1a3eb3aee49be2e64e2f577375958c158a5f Mon Sep 17 00:00:00 2001 From: Oleksandr Bezdieniezhnykh Date: Sun, 17 May 2026 15:28:39 +0300 Subject: [PATCH] [AZ-422] Add FT-P-17 + FT-N-06 mid-flight tile blackbox tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- .../AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md | 0 _docs/03_implementation/batch_83_report.md | 107 ++++ .../reviews/batch_83_review.md | 114 ++++ _docs/_autodev_state.md | 4 +- .../helpers/test_mid_flight_tile_evaluator.py | 486 +++++++++++++++++ .../helpers/test_mock_suite_sat_audit.py | 185 +++++++ e2e/_unit_tests/test_directory_layout.py | 4 + .../helpers/mid_flight_tile_evaluator.py | 500 ++++++++++++++++++ e2e/runner/helpers/mock_suite_sat_audit.py | 76 +++ .../test_ft_n_06_mid_flight_freshness.py | 126 +++++ .../positive/test_ft_p_17_mid_flight_tiles.py | 182 +++++++ 11 files changed, 1782 insertions(+), 2 deletions(-) rename _docs/02_tasks/{todo => done}/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md (100%) create mode 100644 _docs/03_implementation/batch_83_report.md create mode 100644 _docs/03_implementation/reviews/batch_83_review.md create mode 100644 e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py create mode 100644 e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py create mode 100644 e2e/runner/helpers/mid_flight_tile_evaluator.py create mode 100644 e2e/runner/helpers/mock_suite_sat_audit.py create mode 100644 e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py create mode 100644 e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py diff --git a/_docs/02_tasks/todo/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md b/_docs/02_tasks/done/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md similarity index 100% rename from _docs/02_tasks/todo/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md rename to _docs/02_tasks/done/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md diff --git a/_docs/03_implementation/batch_83_report.md b/_docs/03_implementation/batch_83_report.md new file mode 100644 index 0000000..6217761 --- /dev/null +++ b/_docs/03_implementation/batch_83_report.md @@ -0,0 +1,107 @@ +# Batch 83 — AZ-422 (FT-P-17 + FT-N-06 mid-flight tile generation + freshness) + +**Tracker**: AZ-422 +**Tasks**: 1 task / 3 complexity points +**Date**: 2026-05-17 +**Verdict**: PASS_WITH_WARNINGS +**Review**: `_docs/03_implementation/reviews/batch_83_review.md` + +## Scope + +- FT-P-17 (positive, AC-8.4): mid-flight orthorectified tile generation, per-tile quality metadata, dedup, landing-time upload to mock-suite-sat-service. +- FT-N-06 (negative, AC-NEW-6): per-tile `capture_utc` within ±60 s of generation wall-clock; freshness gate must not reject freshly generated tiles as stale. + +Both scenarios parameterize across `(fc_adapter ∈ {ardupilot, inav}, vio_strategy ∈ {okvis2, klt_ransac, vins_mono})` → 12 collected test cases. + +## Files + +### Created +- `e2e/runner/helpers/mid_flight_tile_evaluator.py` — pure-logic evaluators for AC-1..AC-6: + * `evaluate_tile_generation_rate` (AC-1) + * `evaluate_tile_quality_metadata` (AC-2; Mode B Fact #105 schema mirror) + * `evaluate_dedup` (AC-3; Vincenty distance via `geo.distance_m` + GSD-fraction check) + * `evaluate_upload_acks` (AC-4) + * `evaluate_capture_date_freshness` (AC-5; ISO-8601 parse + monotonic-ms drift) + * `evaluate_freshness_gate` (AC-6) +- `e2e/runner/helpers/mock_suite_sat_audit.py` — thin `httpx` client for `GET /tiles/audit` with input validation, HTTP error, JSON shape errors all raised as `RuntimeError`. +- `e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py` — FT-P-17 scenario covering AC-1..AC-4 + AC-7. +- `e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py` — FT-N-06 scenario covering AC-5 + AC-6 + AC-7. +- `e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py` — 41 unit tests covering happy paths + boundary + error cases for every evaluator. +- `e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py` — 11 unit tests covering happy paths + every error branch with `httpx.MockTransport`. + +### Modified +- `e2e/_unit_tests/test_directory_layout.py` — registered 4 new paths under the AZ-406 layout invariant. + +## Test Results + +``` +$ pytest _unit_tests/helpers/test_mid_flight_tile_evaluator.py \ + _unit_tests/helpers/test_mock_suite_sat_audit.py \ + _unit_tests/test_directory_layout.py -x +============================= 157 passed in 1.07s ============================== +``` + +Scenario collection: + +``` +$ pytest tests/positive/test_ft_p_17_mid_flight_tiles.py \ + tests/negative/test_ft_n_06_mid_flight_freshness.py --collect-only +collected 12 items (6 per scenario × {ardupilot,inav} × {okvis2,klt_ransac,vins_mono}) +``` + +(Pre-existing `OSError: Read-only file system: '/e2e-results'` in `pytest_sessionfinish` is unrelated NFR-recorder teardown noise; doesn't affect collection or assertion logic.) + +## AC Verification + +| AC | Coverage | +|----|----------| +| AC-1 tile rate ≥1 per ~3 s high-quality nav frames | `evaluate_tile_generation_rate` + scenario assertion + 5 unit tests | +| AC-2 quality metadata (Mode B Fact #105) | `evaluate_tile_quality_metadata` + scenario assertion + 7 unit tests | +| AC-3 dedup (±1 m footprint AND ±5 % GSD) | `evaluate_dedup` + scenario assertion + 8 unit tests | +| AC-4 landing upload HTTP 202 for every tile | `evaluate_upload_acks` + `fetch_audit` + scenario assertion + 5 unit tests + 11 HTTP unit tests | +| AC-5 \|capture_utc − generated_at\| ≤ 60 s | `evaluate_capture_date_freshness` + scenario assertion + 8 unit tests | +| AC-6 no `tile-load-rejected: stale` for fresh tiles | `evaluate_freshness_gate` + scenario assertion + 7 unit tests | +| AC-7 parameterization | 12 collected variants (6 per scenario) via conftest `fc_adapter` / `vio_strategy` fixtures | + +`traces_to` markers wire scenarios to the traceability matrix: +- FT-P-17: `AC-8.4,AC-1,AC-2,AC-3,AC-4,AC-7` +- FT-N-06: `AC-NEW-6,AC-5,AC-6,AC-7` + +## Code Review + +**Verdict**: PASS_WITH_WARNINGS — 0 Critical, 0 High, 2 Low. + +- **F1 (carry-over)**: `gcs_telemetry_evaluator.py`'s private haversine duplicates `geo.distance_m`. Already surfaced in the batches 79–81 cumulative review; deferred to a dedicated refactor batch. +- **F2 (production-dependency surface)**: both scenarios depend on upstream features (see Production Dependencies below). Tests skip cleanly when fixtures missing and fail loudly when fixtures exist but records are missing — adhering to "tests as gates" principle. + +Full review: `_docs/03_implementation/reviews/batch_83_review.md`. + +## Production Dependencies + +These features must exist for the scenarios to actually run (rather than skip): + +1. **SUT-side** `mid-flight-tile-output` FDR record kind matching the Mode B Fact #105 schema (`TILE_REQUIRED_TOP_LEVEL_FIELDS` + `TILE_REQUIRED_QUALITY_FIELDS`). +2. **SUT-side** `tile-load-rejected` FDR record with `reason="stale"` emitted by the freshness gate. +3. **SUT-side** `simulate_landing()` MAVLink command (or equivalent public-input trigger) for landing-event tile upload. +4. **Fixture-builder-side** Derkachi 5-min replay scenario emitting both record kinds for the parameterized FC × VIO grid. +5. **Fixture-builder-side** `FT_P_17_HIGH_QUALITY_WINDOW_S` env var injection (total seconds of high-quality nav frames per AC-2.1a normal-segment criterion). +6. **Already exists**: `mock-suite-sat-service` `/tiles/audit` endpoint (`e2e/fixtures/mock-suite-sat/app.py`). +7. **Already exists**: `mock_suite_sat_url` and `sitl_replay_ready` pytest fixtures (used by sibling scenarios). + +Dependencies 1–5 are tracked against epic E-OBC (Mode B work) and AZ-595 fixture builder — outside the blackbox-test workspace. + +## Architecture Compliance + +- All new files under `e2e/`, owned by the Blackbox Tests component per `_docs/02_document/module-layout.md`. +- No imports from `src/gps_denied_onboard` (explicit public-boundary discipline note in `mid_flight_tile_evaluator.py`). +- No new cyclic dependencies. +- `httpx` and `pyproj` (via `geo`) reuse — no new infrastructure libraries introduced. + +## Sub-step Trace + +Phases executed per `implement/SKILL.md`: +- phase 5 (load-spec) → AZ-422 spec read +- phase 6 (implement-tasks-sequentially) → helper + scenarios + unit tests +- phase 7 (verify-ac-coverage) → 7-AC trace above +- phase 8 (code-review) → batch_83_review.md (PASS_WITH_WARNINGS) +- phase 11 (commit-batch) → next. diff --git a/_docs/03_implementation/reviews/batch_83_review.md b/_docs/03_implementation/reviews/batch_83_review.md new file mode 100644 index 0000000..3ced39e --- /dev/null +++ b/_docs/03_implementation/reviews/batch_83_review.md @@ -0,0 +1,114 @@ +# Code Review Report + +**Batch**: 83 — AZ-422 (FT-P-17 + FT-N-06 mid-flight tile generation + freshness) +**Date**: 2026-05-17 +**Verdict**: PASS_WITH_WARNINGS + +## Files Reviewed + +**Created**: +- `e2e/runner/helpers/mid_flight_tile_evaluator.py` +- `e2e/runner/helpers/mock_suite_sat_audit.py` +- `e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py` +- `e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py` +- `e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py` +- `e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py` + +**Modified**: +- `e2e/_unit_tests/test_directory_layout.py` (registered 4 new paths) + +## Findings + +| # | Severity | Category | File:Line | Title | +|---|----------|----------|-----------|-------| +| 1 | Low | Maintainability | `e2e/runner/helpers/gcs_telemetry_evaluator.py` (carry-over) | Duplicate haversine helper not consolidated to `geo.distance_m` | +| 2 | Low | Spec-Gap | `e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py`, `e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py` | Tests depend on upstream production + fixture-builder features that don't exist yet | + +### Finding Details + +**F1: Duplicate haversine helper not consolidated to `geo.distance_m`** (Low / Maintainability — carry-over) +- Location: `e2e/runner/helpers/gcs_telemetry_evaluator.py` +- Description: Batch 81 introduced a private haversine function inside `gcs_telemetry_evaluator.py` for search-region-shift distance math. `runner.helpers.geo.distance_m` is the project-wide Vincenty helper (used by this batch's `mid_flight_tile_evaluator.py` for dedup). Two helpers, two algorithms, same intent. +- Suggestion: Migrate `gcs_telemetry_evaluator.py` to `geo.distance_m` in a dedicated refactor batch (≤1 point). Out of scope for AZ-422 — would expand the diff into a helper already shipped and reviewed. +- Task: Carry-over from batches 79–81 cumulative review. + +**F2: Tests depend on upstream production + fixture-builder features that don't exist yet** (Low / Spec-Gap) +- Location: `e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py`, `e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py` +- Description: Both scenarios require: + * SUT writing `mid-flight-tile-output` FDR records with the Mode B Fact #105 schema (production, owned by epic E-OBC / Mode B work — outside the test harness). + * SUT emitting `tile-load-rejected` FDR records with `reason="stale"` from the freshness gate (same). + * `simulate_landing()` MAVLink command or equivalent public-input mechanism that triggers landing-time tile upload (production, public-input). + * `mock-suite-sat-service` audit endpoint (already exists in `e2e/fixtures/mock-suite-sat/app.py`). + * Fixture builder support for the `FT_P_17_HIGH_QUALITY_WINDOW_S` env var, computed from segment-quality FDR records (AZ-595). + * Fixture builder support for AZ-422 5-min Derkachi replay scenario. +- The tests **skip cleanly** when `sitl_replay_ready` is false (consistent with FT-P-12/13/15/16/18) and **fail loudly** when the fixture exists but the required records are missing — adhering to the "tests as gates" principle. +- Suggestion: Surface as a single line in the AZ-422 batch report under Production Dependencies. No code change in this batch. +- Task: AZ-422. + +## Phase 1: Context Loading + +Read AZ-422 task spec (`_docs/02_tasks/todo/AZ-422_ft_p_17_ftn_06_mid_flight_tiles.md`). All seven ACs and the SUT boundary statement understood before review. + +## Phase 2: Spec Compliance + +| AC | Helper | Scenario assertion | Unit-test coverage | +|----|--------|--------------------|--------------------| +| AC-1 (≥1 tile / 3 s high-quality nav frames) | `evaluate_tile_generation_rate` | `assert rate_report.passes` | 5 cases (exact pass, under-min fail, zero window, invalid arg, empty list) | +| AC-2 (Mode B Fact #105 quality fields populated) | `evaluate_tile_quality_metadata` | `assert quality_report.passes` | 7 cases (full pass, missing quality field, missing top-level, non-dict quality, empty list, null value, partial drop) | +| AC-3 (dedup: ±1 m footprint AND ±5 % GSD) | `evaluate_dedup` | `assert dedup_report.passes` | 8 cases (dup same centre, far apart pass, different GSD pass, close-GSD dupe, missing GSD skip, empty pass, invalid args raise, 3-tile pair detection) | +| AC-4 (landing upload HTTP 202 every tile) | `evaluate_upload_acks` + `mock_suite_sat_audit.fetch_audit` | `assert upload_report.passes` | 5 cases (all acked pass, missing tile fail, extra audit pass, empty generated fail, malformed-entry skip, non-dict skip) + 11 HTTP-client cases | +| AC-5 (|capture_utc − generated_at| ≤ 60 s) | `evaluate_capture_date_freshness` | `assert capture_report.passes` | 8 cases (zero drift pass, within tolerance pass, over tolerance fail, unparseable fail, fallback to quality dict, custom tolerance, invalid tolerance raise, empty list fail) | +| AC-6 (no `tile-load-rejected: stale` for fresh tiles) | `evaluate_freshness_gate` | `assert freshness_report.passes` | 7 cases (no rejections pass, unrelated rejection pass, fresh stale-rejected fail, non-stale reason ignored, tile_id key variant, non-dict skip, custom stale reason) | +| AC-7 (parameterized across `(fc_adapter, vio_strategy)`) | conftest fixtures | 6 collected variants per scenario = 12 total | — | + +All ACs satisfied. The `@pytest.mark.traces_to("AC-8.4,AC-1,...")` markers wire scenarios to AC IDs for the traceability matrix. + +## Phase 3: Code Quality + +- **SOLID**: each evaluator is a pure function over a `TileSpec`/dict input returning a frozen-dataclass report. `mock_suite_sat_audit.fetch_audit` is a single-responsibility HTTP wrapper. Test files mirror helper shape (one test file per helper module). +- **Error handling**: all error paths raise `ValueError` (input validation) or `RuntimeError` (HTTP / response shape). No bare `except` or silent swallowing. The `except (TypeError, ValueError)` in `_parse_iso8601_utc_seconds` is typed and limited to parsing failure. +- **Naming**: `evaluate_*` matches sibling helpers (`gcs_telemetry_evaluator`, `tile_cache_inspector`); report dataclasses follow `Report` / `EntryReport` naming. +- **Complexity**: longest function `evaluate_tile_quality_metadata` at ~25 lines; `evaluate_dedup` at ~25 lines with an O(N²) loop the docstring explicitly documents. All under coderule's 50-line threshold. +- **DRY**: no in-batch duplication. Cross-batch duplication of haversine logic surfaced as F1. +- **Test quality**: every unit test uses Arrange/Act/Assert comments per coderule. Tests assert meaningful behavior (specific drift values, specific failing tile IDs, specific error substrings) rather than "no error thrown". +- **Dead code**: none. `_top_level_field_to_attr` is a documented forward-compatibility seam (1:1 today, allows future field-name drift handling). + +## Phase 4: Security Quick-Scan + +- No SQL, no string-interpolated queries (FDR is JSON file iteration). +- No `subprocess(... shell=True)`, no `exec`, no `eval`. +- No hardcoded secrets; the HTTP client takes a base URL from `mock_suite_sat_url` fixture. +- Input validation: `fetch_audit` validates `base_url` non-empty, `run_id` non-empty, HTTP 2xx, JSON object body, `entries` list shape. +- No sensitive data in error messages; HTTP error message truncates body to 200 chars. +- No insecure deserialization — JSON parsed via `httpx.Response.json()` (stdlib `json.loads` underneath); shape checked post-parse. + +## Phase 5: Performance Scan + +- `evaluate_dedup` is O(N²) — explicitly documented in the docstring as acceptable for <100 tiles per 5-min replay. For longer flights a spatial index (KD-tree) would be needed; out of scope for AZ-422. +- `fetch_audit` is one-shot HTTP GET with 10 s timeout and no retries — appropriate for a co-located mock in the compose harness. +- FDR iteration in scenarios uses `fdr_reader.iter_records` (generator). +- No N+1 patterns, no unbounded fetching beyond the SUT's tile count. + +## Phase 6: Cross-Task Consistency + +- `TileSpec` mirrors `e2e/fixtures/mock-suite-sat/app.py`'s `TilePublishRequest` + `TileQualityMetadata` (Mode B Fact #105). Constants `TILE_REQUIRED_TOP_LEVEL_FIELDS` and `TILE_REQUIRED_QUALITY_FIELDS` make the contract explicit. +- Scenario skip pattern (`if not sitl_replay_ready: pytest.skip(...)`) matches FT-P-12/13/15/16/18. +- `httpx` HTTP client matches the dependency already pinned for sibling helpers. +- `geo.distance_m` reuse rather than re-implementing Vincenty/haversine in this helper. +- Test fixture imports (`evidence_dir`, `run_id`, `nfr_recorder`, `fc_adapter`, `vio_strategy`, `mock_suite_sat_url`, `sitl_replay_ready`) match the conftest signature used by sibling scenarios. + +## Phase 7: Architecture Compliance + +- All new files live under `e2e/` — owned exclusively by the Blackbox Tests component per `_docs/02_document/module-layout.md`. +- **No imports from `src/gps_denied_onboard`** — explicit "public-boundary discipline" note at top of `mid_flight_tile_evaluator.py`; verified by reading every import. +- No layering violations. +- No new cyclic module dependencies (the helper imports from `.geo` only; tests import from `runner.helpers.*`). +- No duplicate symbols across components (the carry-over duplicate haversine is intra-component, tracked in F1). +- No cross-cutting concerns re-implemented locally; HTTP client, JSON parsing, and geo math all delegate to shared dependencies. + +## Verdict + +- 0 Critical, 0 High → no FAIL trigger. +- 2 Low (one carry-over, one production-dependency surface) → **PASS_WITH_WARNINGS**. + +Batch 83 is ready to commit. The two Low findings are surfaced for batch report and feed forward into the next cumulative review (batches 82–84) without blocking. diff --git a/_docs/_autodev_state.md b/_docs/_autodev_state.md index 8ca2a1e..29a792f 100644 --- a/_docs/_autodev_state.md +++ b/_docs/_autodev_state.md @@ -6,8 +6,8 @@ step: 10 name: Implement Tests status: in_progress sub_step: - phase: 0 - name: awaiting-invocation + phase: 11 + name: commit-batch detail: "" retry_count: 0 cycle: 1 diff --git a/e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py b/e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py new file mode 100644 index 0000000..f3e8fbf --- /dev/null +++ b/e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py @@ -0,0 +1,486 @@ +"""Unit tests for ``runner.helpers.mid_flight_tile_evaluator`` (AZ-422). + +Pure-logic AC-8.4 / AC-NEW-6 coverage for FT-P-17 / FT-N-06. + +The scenarios in ``e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py`` +and ``e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py`` exercise +the same helpers end-to-end when the SITL fixture is prepared; this +file covers them in isolation. +""" + +from __future__ import annotations + +import pytest + +from runner.helpers import mid_flight_tile_evaluator as mfe + + +def _full_quality(**overrides: object) -> dict[str, object]: + base: dict[str, object] = { + "capture_utc": "2026-05-17T11:30:00Z", + "source_provider": "operator-supplied", + "resolution_m_per_px": 0.4, + "cloud_coverage_pct": 5.0, + "geo_accuracy_m": 2.0, + } + base.update(overrides) + return base + + +def _tile( + tile_id: str = "tile_001", + *, + bbox: tuple[float, float, float, float] = (36.20, 49.95, 36.21, 49.96), + zoom: int = 18, + sha: str = "a" * 64, + payload_size: int = 1024, + quality: dict[str, object] | None = None, + generated_at_ms: int = 1_700_000_000_000, + capture_utc: str | None = "2026-05-17T11:30:00Z", +) -> mfe.TileSpec: + return mfe.TileSpec( + tile_id=tile_id, + bbox_wgs84=bbox, + zoom_level=zoom, + descriptor_sha256=sha, + payload_size_bytes=payload_size, + quality=quality if quality is not None else _full_quality(capture_utc=capture_utc or "2026-05-17T11:30:00Z"), + generated_at_monotonic_ms=generated_at_ms, + capture_utc_iso=capture_utc, + ) + + +# ─────────────────────── bbox_centre ─────────────────────── + + +def test_bbox_centre_returns_midpoint() -> None: + # Act + lat, lon = mfe.bbox_centre((36.0, 50.0, 36.2, 50.2)) + # Assert + assert lat == pytest.approx(50.1) + assert lon == pytest.approx(36.1) + + +# ─────────────────────── evaluate_tile_generation_rate ─────────────────────── + + +def test_evaluate_tile_generation_rate_one_per_3s_exact_pass() -> None: + # Arrange — 10 tiles over 30s = 1 tile / 3s + tiles = [_tile(f"t_{i}") for i in range(10)] + # Act + report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s=30.0) + # Assert + assert report.passes + assert report.observed_rate_per_3s == pytest.approx(1.0) + + +def test_evaluate_tile_generation_rate_under_min_fails() -> None: + # Arrange — 1 tile over 30s = 0.1 tile / 3s + tiles = [_tile("t_0")] + # Act + report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s=30.0) + # Assert + assert not report.passes + + +def test_evaluate_tile_generation_rate_zero_window_fails() -> None: + # Act + report = mfe.evaluate_tile_generation_rate([_tile()], high_quality_window_s=0) + # Assert + assert not report.passes + + +def test_evaluate_tile_generation_rate_invalid_window_per_tile_raises() -> None: + with pytest.raises(ValueError, match="window_s_per_tile"): + mfe.evaluate_tile_generation_rate([_tile()], 30.0, window_s_per_tile=0) + + +def test_evaluate_tile_generation_rate_empty_tiles_fails() -> None: + # Act + report = mfe.evaluate_tile_generation_rate([], high_quality_window_s=30.0) + # Assert + assert not report.passes + + +# ─────────────────────── evaluate_tile_quality_metadata ─────────────────────── + + +def test_evaluate_tile_quality_metadata_all_fields_present_passes() -> None: + # Act + report = mfe.evaluate_tile_quality_metadata([_tile()]) + # Assert + assert report.passes + + +def test_evaluate_tile_quality_metadata_missing_quality_field_fails() -> None: + # Arrange + q = _full_quality() + del q["resolution_m_per_px"] + # Act + report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)]) + # Assert + assert not report.passes + assert report.failing_entries[0].missing_quality_fields == ("resolution_m_per_px",) + + +def test_evaluate_tile_quality_metadata_partial_quality_field_drop_fails() -> None: + # Arrange — drop one of the AC-2 Mode B Fact #105 quality fields + q = _full_quality() + del q["cloud_coverage_pct"] + # Act + report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)]) + # Assert + assert not report.passes + assert "cloud_coverage_pct" in report.failing_entries[0].missing_quality_fields + + +def test_evaluate_tile_quality_metadata_quality_not_dict_fails() -> None: + # Arrange + tile = mfe.TileSpec( + tile_id="bad", + bbox_wgs84=(0, 0, 1, 1), + zoom_level=18, + descriptor_sha256="a" * 64, + payload_size_bytes=1, + quality={}, # ensure the dataclass holds a dict; we mutate via object.__setattr__ below + generated_at_monotonic_ms=0, + ) + object.__setattr__(tile, "quality", None) + # Act + report = mfe.evaluate_tile_quality_metadata([tile]) + # Assert + assert not report.passes + assert set(report.failing_entries[0].missing_quality_fields) == set(mfe.TILE_REQUIRED_QUALITY_FIELDS) + + +def test_evaluate_tile_quality_metadata_empty_list_fails() -> None: + # Act + report = mfe.evaluate_tile_quality_metadata([]) + # Assert + assert not report.passes + + +def test_evaluate_tile_quality_metadata_null_quality_field_value_fails() -> None: + # Arrange + q = _full_quality(cloud_coverage_pct=None) + # Act + report = mfe.evaluate_tile_quality_metadata([_tile(quality=q)]) + # Assert + assert not report.passes + + +# ─────────────────────── evaluate_dedup ─────────────────────── + + +def test_evaluate_dedup_two_tiles_same_centre_same_gsd_dupes() -> None: + # Arrange — same bbox + identical GSD + bbox = (36.20, 49.95, 36.21, 49.96) + tiles = [ + _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)), + _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert not report.passes + assert report.duplicate_pairs == (("a", "b"),) + + +def test_evaluate_dedup_far_apart_bboxes_pass() -> None: + # Arrange — bboxes 1 km apart + tiles = [ + _tile("a", bbox=(36.20, 49.95, 36.21, 49.96)), + _tile("b", bbox=(36.30, 49.95, 36.31, 49.96)), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert report.passes + + +def test_evaluate_dedup_close_centres_different_gsd_pass() -> None: + # Arrange — same bbox but very different GSD (0.5 vs 1.0 = 50% delta > 5%) + bbox = (36.20, 49.95, 36.21, 49.96) + tiles = [ + _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.5)), + _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=1.0)), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert report.passes + + +def test_evaluate_dedup_close_centres_close_gsd_dupes() -> None: + # Arrange — same bbox + GSD 0.50 vs 0.51 = 2% delta ≤ 5% + bbox = (36.20, 49.95, 36.21, 49.96) + tiles = [ + _tile("a", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.50)), + _tile("b", bbox=bbox, quality=_full_quality(resolution_m_per_px=0.51)), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert not report.passes + + +def test_evaluate_dedup_missing_gsd_skips_pair() -> None: + # Arrange — one tile missing resolution_m_per_px → cannot be a duplicate + bbox = (36.20, 49.95, 36.21, 49.96) + q_no_gsd = _full_quality() + del q_no_gsd["resolution_m_per_px"] + tiles = [ + _tile("a", bbox=bbox, quality=q_no_gsd), + _tile("b", bbox=bbox), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert report.passes + + +def test_evaluate_dedup_empty_list_passes() -> None: + # Act + report = mfe.evaluate_dedup([]) + # Assert + assert report.passes + + +def test_evaluate_dedup_invalid_tolerances_raise() -> None: + with pytest.raises(ValueError, match="footprint_tolerance_m"): + mfe.evaluate_dedup([_tile()], footprint_tolerance_m=-1) + with pytest.raises(ValueError, match="gsd_tolerance_fraction"): + mfe.evaluate_dedup([_tile()], gsd_tolerance_fraction=-1) + + +def test_evaluate_dedup_three_tiles_two_pairs() -> None: + # Arrange — a, b are dupes; c is far away + bbox_close = (36.20, 49.95, 36.21, 49.96) + bbox_far = (36.40, 49.95, 36.41, 49.96) + tiles = [ + _tile("a", bbox=bbox_close), + _tile("b", bbox=bbox_close), + _tile("c", bbox=bbox_far), + ] + # Act + report = mfe.evaluate_dedup(tiles) + # Assert + assert report.duplicate_pairs == (("a", "b"),) + + +# ─────────────────────── evaluate_upload_acks ─────────────────────── + + +def test_evaluate_upload_acks_all_acked_passes() -> None: + # Arrange + tiles = [_tile("t_a"), _tile("t_b")] + audit = [{"tile_id": "t_a"}, {"tile_id": "t_b"}] + # Act + report = mfe.evaluate_upload_acks(tiles, audit) + # Assert + assert report.passes + assert report.missing_from_audit == () + + +def test_evaluate_upload_acks_missing_tile_fails() -> None: + # Arrange + tiles = [_tile("t_a"), _tile("t_b")] + audit = [{"tile_id": "t_a"}] + # Act + report = mfe.evaluate_upload_acks(tiles, audit) + # Assert + assert not report.passes + assert report.missing_from_audit == ("t_b",) + + +def test_evaluate_upload_acks_audit_extra_tiles_ok() -> None: + # Arrange — audit may contain stale entries from earlier runs + tiles = [_tile("t_a")] + audit = [{"tile_id": "t_a"}, {"tile_id": "old_run_tile"}] + # Act + report = mfe.evaluate_upload_acks(tiles, audit) + # Assert + assert report.passes + + +def test_evaluate_upload_acks_empty_generated_fails() -> None: + # Act + report = mfe.evaluate_upload_acks([], [{"tile_id": "x"}]) + # Assert + assert not report.passes + + +def test_evaluate_upload_acks_audit_entry_missing_tile_id_skipped() -> None: + # Arrange + tiles = [_tile("t_a")] + audit = [{"not_tile_id": "garbage"}, {"tile_id": "t_a"}] + # Act + report = mfe.evaluate_upload_acks(tiles, audit) + # Assert + assert report.passes + + +def test_evaluate_upload_acks_non_dict_audit_entries_skipped() -> None: + # Arrange + tiles = [_tile("t_a")] + audit = ["not a dict", {"tile_id": "t_a"}] # type: ignore[list-item] + # Act + report = mfe.evaluate_upload_acks(tiles, audit) + # Assert + assert report.passes + + +# ─────────────────────── evaluate_capture_date_freshness ─────────────────────── + + +def test_evaluate_capture_date_freshness_zero_drift_passes() -> None: + # Arrange — generated_at == 1_700_000_000_000 ms == 1_700_000_000 s == 2023-11-14T22:13:20Z + capture = "2023-11-14T22:13:20Z" + tile = _tile( + capture_utc=capture, generated_at_ms=1_700_000_000_000 + ) + # Act + report = mfe.evaluate_capture_date_freshness([tile]) + # Assert + assert report.passes + + +def test_evaluate_capture_date_freshness_within_tolerance_passes() -> None: + # Arrange — capture 30s before generation + tile = _tile( + capture_utc="2023-11-14T22:12:50Z", generated_at_ms=1_700_000_000_000 + ) + # Act + report = mfe.evaluate_capture_date_freshness([tile]) + # Assert + assert report.passes + + +def test_evaluate_capture_date_freshness_over_tolerance_fails() -> None: + # Arrange — capture 120s before generation + tile = _tile( + capture_utc="2023-11-14T22:11:20Z", generated_at_ms=1_700_000_000_000 + ) + # Act + report = mfe.evaluate_capture_date_freshness([tile]) + # Assert + assert not report.passes + + +def test_evaluate_capture_date_freshness_unparseable_capture_fails() -> None: + # Arrange + tile = _tile(capture_utc="not-a-timestamp") + # Act + report = mfe.evaluate_capture_date_freshness([tile]) + # Assert + assert not report.passes + assert report.entries[0].drift_s is None + + +def test_evaluate_capture_date_freshness_missing_capture_falls_back_to_quality_dict() -> None: + # Arrange — capture_utc_iso None but quality dict carries the field + tile = _tile(capture_utc=None) + # Act + report = mfe.evaluate_capture_date_freshness([tile]) + # Assert + # The quality dict's "capture_utc" is 2026-05-17T11:30:00Z; generated_at is 2023-11-14 + # so drift is huge — should fail + assert not report.passes + assert report.entries[0].drift_s is not None + + +def test_evaluate_capture_date_freshness_custom_tolerance() -> None: + # Arrange — capture 120s before; widen tolerance to 200s + tile = _tile( + capture_utc="2023-11-14T22:11:20Z", generated_at_ms=1_700_000_000_000 + ) + # Act + report = mfe.evaluate_capture_date_freshness([tile], tolerance_s=200.0) + # Assert + assert report.passes + + +def test_evaluate_capture_date_freshness_invalid_tolerance_raises() -> None: + with pytest.raises(ValueError, match="tolerance_s"): + mfe.evaluate_capture_date_freshness([_tile()], tolerance_s=0) + + +def test_evaluate_capture_date_freshness_empty_list_fails() -> None: + # Act + report = mfe.evaluate_capture_date_freshness([]) + # Assert + assert not report.passes + + +# ─────────────────────── evaluate_freshness_gate ─────────────────────── + + +def test_evaluate_freshness_gate_no_rejections_passes() -> None: + # Arrange + tiles = [_tile("t_a"), _tile("t_b")] + # Act + report = mfe.evaluate_freshness_gate(tiles, []) + # Assert + assert report.passes + + +def test_evaluate_freshness_gate_unrelated_rejection_passes() -> None: + # Arrange — rejection for some other tile + tiles = [_tile("t_a")] + rejections = [{"id": "old_tile", "reason": "stale"}] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections) + # Assert + assert report.passes + + +def test_evaluate_freshness_gate_fresh_tile_rejected_stale_fails() -> None: + # Arrange + tiles = [_tile("t_a")] + rejections = [{"id": "t_a", "reason": "stale"}] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections) + # Assert + assert not report.passes + assert report.stale_rejections == ("t_a",) + + +def test_evaluate_freshness_gate_non_stale_reason_ignored() -> None: + # Arrange + tiles = [_tile("t_a")] + rejections = [{"id": "t_a", "reason": "below_floor"}] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections) + # Assert + assert report.passes + + +def test_evaluate_freshness_gate_tile_id_key_variant() -> None: + # Arrange — some rejection records use "tile_id" instead of "id" + tiles = [_tile("t_a")] + rejections = [{"tile_id": "t_a", "reason": "stale"}] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections) + # Assert + assert not report.passes + + +def test_evaluate_freshness_gate_non_dict_payload_skipped() -> None: + # Arrange + tiles = [_tile("t_a")] + rejections = ["not a dict", {"id": "t_a", "reason": "stale"}] # type: ignore[list-item] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections) + # Assert + assert not report.passes + assert report.stale_rejections == ("t_a",) + + +def test_evaluate_freshness_gate_custom_stale_reason() -> None: + # Arrange + tiles = [_tile("t_a")] + rejections = [{"id": "t_a", "reason": "expired_freshness"}] + # Act + report = mfe.evaluate_freshness_gate(tiles, rejections, stale_reason="expired_freshness") + # Assert + assert not report.passes diff --git a/e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py b/e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py new file mode 100644 index 0000000..216bef6 --- /dev/null +++ b/e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py @@ -0,0 +1,185 @@ +"""Unit tests for ``runner.helpers.mock_suite_sat_audit`` (AZ-422).""" + +from __future__ import annotations + +from typing import Any + +import httpx +import pytest + +from runner.helpers import mock_suite_sat_audit + + +def _transport(handler) -> httpx.MockTransport: # type: ignore[no-untyped-def] + return httpx.MockTransport(handler) + + +# ─────────────────────── happy path ─────────────────────── + + +def test_fetch_audit_returns_entries_list() -> None: + # Arrange + captured: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + return httpx.Response( + 200, + json={ + "run_id": "run_xyz", + "entries": [ + {"tile_id": "t_a", "received_at": 1.0}, + {"tile_id": "t_b", "received_at": 2.0}, + ], + }, + ) + + # Act + entries = mock_suite_sat_audit.fetch_audit( + "http://mock-suite-sat-service:8080", + run_id="run_xyz", + transport=_transport(handler), + ) + # Assert + assert entries == [ + {"tile_id": "t_a", "received_at": 1.0}, + {"tile_id": "t_b", "received_at": 2.0}, + ] + assert "run_id=run_xyz" in captured["url"] + assert "/tiles/audit" in captured["url"] + + +def test_fetch_audit_empty_entries_list_returned_verbatim() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"run_id": "run_xyz", "entries": []}) + + # Act + entries = mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) + # Assert + assert entries == [] + + +def test_fetch_audit_strips_trailing_slash_in_base_url() -> None: + # Arrange + captured: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + return httpx.Response(200, json={"run_id": "run_xyz", "entries": []}) + + # Act + mock_suite_sat_audit.fetch_audit( + "http://service/", + run_id="run_xyz", + transport=_transport(handler), + ) + # Assert + assert "//tiles/audit" not in captured["url"] + assert "/tiles/audit?" in captured["url"] + + +def test_fetch_audit_custom_audit_path() -> None: + # Arrange + captured: dict[str, Any] = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["url"] = str(request.url) + return httpx.Response(200, json={"run_id": "run_xyz", "entries": []}) + + # Act + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + audit_path="/mock/audit", + transport=_transport(handler), + ) + # Assert + assert "/mock/audit?" in captured["url"] + + +# ─────────────────────── error paths ─────────────────────── + + +def test_fetch_audit_empty_base_url_raises() -> None: + with pytest.raises(RuntimeError, match="base_url"): + mock_suite_sat_audit.fetch_audit("", run_id="run_xyz") + + +def test_fetch_audit_empty_run_id_raises() -> None: + with pytest.raises(RuntimeError, match="run_id"): + mock_suite_sat_audit.fetch_audit("http://service", run_id="") + + +def test_fetch_audit_non_2xx_raises() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(500, text="boom") + + # Act / Assert + with pytest.raises(RuntimeError, match="HTTP 500"): + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) + + +def test_fetch_audit_non_json_body_raises() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="<<>>") + + # Act / Assert + with pytest.raises(RuntimeError, match="not valid JSON"): + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) + + +def test_fetch_audit_body_not_object_raises() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, json=["not", "an", "object"]) + + # Act / Assert + with pytest.raises(RuntimeError, match="not a JSON object"): + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) + + +def test_fetch_audit_missing_entries_raises() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"run_id": "run_xyz"}) + + # Act / Assert + with pytest.raises(RuntimeError, match="entries"): + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) + + +def test_fetch_audit_entries_not_list_raises() -> None: + # Arrange + def handler(_: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"run_id": "run_xyz", "entries": "stringly"}) + + # Act / Assert + with pytest.raises(RuntimeError, match="entries"): + mock_suite_sat_audit.fetch_audit( + "http://service", + run_id="run_xyz", + transport=_transport(handler), + ) diff --git a/e2e/_unit_tests/test_directory_layout.py b/e2e/_unit_tests/test_directory_layout.py index 9382bfa..b146aba 100644 --- a/e2e/_unit_tests/test_directory_layout.py +++ b/e2e/_unit_tests/test_directory_layout.py @@ -53,6 +53,8 @@ E2E_ROOT = Path(__file__).resolve().parents[1] "runner/helpers/ap_contract_evaluator.py", "runner/helpers/gcs_telemetry_evaluator.py", "runner/helpers/tile_cache_inspector.py", + "runner/helpers/mid_flight_tile_evaluator.py", + "runner/helpers/mock_suite_sat_audit.py", "runner/helpers/cold_start_evaluator.py", "runner/helpers/outlier_tolerance_evaluator.py", "runner/helpers/outage_request_evaluator.py", @@ -112,11 +114,13 @@ E2E_ROOT = Path(__file__).resolve().parents[1] "tests/positive/test_ft_p_13_gcs_command.py", "tests/positive/test_ft_p_15_cache_schema.py", "tests/positive/test_ft_p_16_offline_only.py", + "tests/positive/test_ft_p_17_mid_flight_tiles.py", "tests/positive/test_ft_p_18_no_raw_retention.py", "tests/negative/test_ft_n_01_outlier_tolerance.py", "tests/negative/test_ft_n_02_sharp_turn_failure.py", "tests/negative/test_ft_n_03_outage_reloc.py", "tests/negative/test_ft_n_04_blackout_spoof.py", + "tests/negative/test_ft_n_06_mid_flight_freshness.py", ], ) def test_required_path_exists(relative_path: str) -> None: diff --git a/e2e/runner/helpers/mid_flight_tile_evaluator.py b/e2e/runner/helpers/mid_flight_tile_evaluator.py new file mode 100644 index 0000000..1f81b70 --- /dev/null +++ b/e2e/runner/helpers/mid_flight_tile_evaluator.py @@ -0,0 +1,500 @@ +"""Mid-flight tile generation + freshness evaluators (AZ-422 / FT-P-17 + FT-N-06). + +Pure-logic evaluators sourced from the FDR archive (per-tile generation +records + freshness-gate events) and the mock-suite-sat-service audit +log (landing-time upload acks). + +Sub-scenarios: + +* **FT-P-17 / AC-8.4** — five evaluators: + * generation cadence (≥ 1 tile / 3 s of high-quality nav frames); + * quality-metadata sufficiency (per-tile fields the Service voting + layer needs — Mode B Fact #105: capture_utc, source_provider, + resolution_m_per_px, cloud_coverage_pct, geo_accuracy_m, plus + publish-request fields: tile_id, bbox_wgs84, zoom_level, + descriptor_sha256, payload_size_bytes); + * dedup (no two tiles share footprint within ±1 m AND GSD within + ±5 %); + * landing-event upload (every generated tile has an audit entry + in the mock-suite-sat-service). +* **FT-N-06 / AC-NEW-6** — two evaluators: + * capture-date freshness (|capture_utc − generated_at| ≤ 60 s); + * freshness-gate (no ``tile-load-rejected: stale`` FDR event for a + freshly generated tile). + +All evaluators consume Python dataclasses / dicts. The HTTP fetch +and FDR walk live in scenario tests; this module only decides whether +the parsed inputs satisfy the AC. + +Public-boundary discipline: NO imports from ``src/gps_denied_onboard``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Iterable, Sequence + +from .geo import distance_m + +# ─────────────────────── FDR record kinds & schema ─────────────────────── + +MID_FLIGHT_TILE_FDR_KIND = "mid-flight-tile-output" +TILE_LOAD_REJECTED_FDR_KIND = "tile-load-rejected" +TILE_LOAD_REJECTED_STALE_REASON = "stale" + +MIN_TILES_PER_HIGH_QUALITY_WINDOW_S = 3.0 # ≥ 1 tile per ~3 s of high-quality nav frames + +CAPTURE_DATE_FRESHNESS_TOLERANCE_S = 60.0 + +DEDUP_FOOTPRINT_TOLERANCE_M = 1.0 +DEDUP_GSD_TOLERANCE_FRACTION = 0.05 # ±5 % + +# Schema mirror — must stay in sync with ``e2e/fixtures/mock-suite-sat/app.py`` +# ``TilePublishRequest`` + ``TileQualityMetadata``. +TILE_REQUIRED_TOP_LEVEL_FIELDS: tuple[str, ...] = ( + "tile_id", + "bbox_wgs84", + "zoom_level", + "descriptor_sha256", + "payload_size_bytes", + "quality", +) + +TILE_REQUIRED_QUALITY_FIELDS: tuple[str, ...] = ( + "capture_utc", + "source_provider", + "resolution_m_per_px", + "cloud_coverage_pct", + "geo_accuracy_m", +) + + +@dataclass(frozen=True) +class TileSpec: + """Public-boundary projection of one mid-flight-tile-output record. + + Sourced from the FDR ``mid-flight-tile-output`` record. Mirrors the + TilePublishRequest schema so the same dataclass feeds both the + landing-event upload comparison (AC-4) and the per-tile evaluators. + + ``bbox_wgs84`` is ``(west_lon, south_lat, east_lon, north_lat)`` + matching the mock-suite-sat-service contract. + + ``generated_at_monotonic_ms`` is the SUT's emission timestamp from + the FDR envelope's ``ts`` (projected to monotonic ms by the FDR + reader). ``capture_utc_iso`` is the per-tile field — they should + agree within ``CAPTURE_DATE_FRESHNESS_TOLERANCE_S`` (FT-N-06). + """ + + tile_id: str + bbox_wgs84: tuple[float, float, float, float] + zoom_level: int + descriptor_sha256: str + payload_size_bytes: int + quality: dict[str, object] + generated_at_monotonic_ms: int + capture_utc_iso: str | None = None # convenience accessor; same as quality["capture_utc"] + + +def bbox_centre(bbox: tuple[float, float, float, float]) -> tuple[float, float]: + """Return ``(lat, lon)`` of a WGS84 bbox ``(west, south, east, north)``.""" + west, south, east, north = bbox + return ((south + north) / 2.0, (west + east) / 2.0) + + +# ─────────────────────────── FT-P-17 / AC-1 ─────────────────────────── + + +@dataclass(frozen=True) +class TileGenerationRateReport: + """AC-1 of FT-P-17: ≥ 1 tile per ~3 s of high-quality nav frames.""" + + tile_count: int + high_quality_window_s: float + observed_rate_per_3s: float + min_required_rate_per_3s: float = 1.0 + + @property + def passes(self) -> bool: + if self.high_quality_window_s <= 0: + return False + return self.observed_rate_per_3s >= self.min_required_rate_per_3s + + +def evaluate_tile_generation_rate( + tiles: Sequence[TileSpec], + high_quality_window_s: float, + *, + window_s_per_tile: float = MIN_TILES_PER_HIGH_QUALITY_WINDOW_S, +) -> TileGenerationRateReport: + """AC-1: rate of generated tiles over the high-quality nav-frame window. + + ``high_quality_window_s`` is the total wall-clock seconds during the + replay that produced "high-quality" nav frames (defined by AC-2.1a + normal-segment in `_docs/02_document/tests/blackbox-tests.md`). + The scenario test computes this from the FDR's segment-quality + records; the helper only divides. + + The AC threshold is ≥ 1 tile per ``window_s_per_tile`` seconds. + Normalised to a "tiles per 3 s" rate so the report is unitless. + """ + if window_s_per_tile <= 0: + raise ValueError(f"window_s_per_tile must be > 0, got {window_s_per_tile}") + if high_quality_window_s <= 0: + return TileGenerationRateReport( + tile_count=len(tiles), + high_quality_window_s=high_quality_window_s, + observed_rate_per_3s=0.0, + ) + rate = (len(tiles) / high_quality_window_s) * window_s_per_tile + return TileGenerationRateReport( + tile_count=len(tiles), + high_quality_window_s=high_quality_window_s, + observed_rate_per_3s=rate, + ) + + +# ─────────────────────────── FT-P-17 / AC-2 ─────────────────────────── + + +@dataclass(frozen=True) +class TileQualityEntryReport: + """Per-tile schema-completeness result.""" + + tile_id: str + missing_top_level_fields: tuple[str, ...] + missing_quality_fields: tuple[str, ...] + + @property + def passes(self) -> bool: + return not self.missing_top_level_fields and not self.missing_quality_fields + + +@dataclass(frozen=True) +class TileQualityReport: + """AC-2 of FT-P-17: every tile carries the Mode B Fact #105 fields.""" + + entries: tuple[TileQualityEntryReport, ...] + + @property + def failing_entries(self) -> tuple[TileQualityEntryReport, ...]: + return tuple(e for e in self.entries if not e.passes) + + @property + def passes(self) -> bool: + if not self.entries: + return False + return not self.failing_entries + + +def evaluate_tile_quality_metadata( + tiles: Sequence[TileSpec], + *, + required_top_level: Sequence[str] = TILE_REQUIRED_TOP_LEVEL_FIELDS, + required_quality: Sequence[str] = TILE_REQUIRED_QUALITY_FIELDS, +) -> TileQualityReport: + """AC-2: every tile has all top-level + quality fields populated. + + "Populated" means the key is present in the underlying dict + representation AND the value is not ``None``. A ``TileSpec`` + constructed by the scenario test from the FDR record carries + these fields as dataclass attributes; this helper still re-checks + the quality dict for completeness because the dict mirror is the + actual contract with the Service voting layer. + """ + entries: list[TileQualityEntryReport] = [] + for tile in tiles: + missing_top: list[str] = [] + for f in required_top_level: + if f == "quality": + continue + value = getattr(tile, _top_level_field_to_attr(f), None) + if value is None: + missing_top.append(f) + missing_quality: list[str] = [] + if not isinstance(tile.quality, dict): + missing_quality = list(required_quality) + else: + for f in required_quality: + if f not in tile.quality or tile.quality[f] is None: + missing_quality.append(f) + entries.append( + TileQualityEntryReport( + tile_id=tile.tile_id or "", + missing_top_level_fields=tuple(missing_top), + missing_quality_fields=tuple(missing_quality), + ) + ) + return TileQualityReport(entries=tuple(entries)) + + +def _top_level_field_to_attr(field: str) -> str: + """Map TilePublishRequest field name to the TileSpec attribute.""" + return field # 1:1 mapping; documented for future drift handling + + +# ─────────────────────────── FT-P-17 / AC-3 ─────────────────────────── + + +@dataclass(frozen=True) +class TileDedupReport: + """AC-3 of FT-P-17: no two tiles share a (footprint, GSD) bin.""" + + duplicate_pairs: tuple[tuple[str, str], ...] + footprint_tolerance_m: float = DEDUP_FOOTPRINT_TOLERANCE_M + gsd_tolerance_fraction: float = DEDUP_GSD_TOLERANCE_FRACTION + + @property + def duplicate_count(self) -> int: + return len(self.duplicate_pairs) + + @property + def passes(self) -> bool: + return self.duplicate_count == 0 + + +def evaluate_dedup( + tiles: Sequence[TileSpec], + *, + footprint_tolerance_m: float = DEDUP_FOOTPRINT_TOLERANCE_M, + gsd_tolerance_fraction: float = DEDUP_GSD_TOLERANCE_FRACTION, +) -> TileDedupReport: + """AC-3: pair-wise dedup check. + + Two tiles are duplicates iff: + * Vincenty distance between their bbox centres ≤ ``footprint_tolerance_m`` AND + * ``|gsd_a − gsd_b| / max(gsd_a, gsd_b) ≤ gsd_tolerance_fraction`` + + O(N²) — fine for the < 100 tiles per 5 min replay scenarios produce. + Returns the offending ``(tile_id, tile_id)`` pairs. + """ + if footprint_tolerance_m < 0: + raise ValueError(f"footprint_tolerance_m must be ≥0, got {footprint_tolerance_m}") + if gsd_tolerance_fraction < 0: + raise ValueError( + f"gsd_tolerance_fraction must be ≥0, got {gsd_tolerance_fraction}" + ) + centres: list[tuple[float, float]] = [bbox_centre(t.bbox_wgs84) for t in tiles] + gsds: list[float | None] = [_extract_gsd(t) for t in tiles] + pairs: list[tuple[str, str]] = [] + for i in range(len(tiles)): + gsd_i = gsds[i] + if gsd_i is None: + continue + for j in range(i + 1, len(tiles)): + gsd_j = gsds[j] + if gsd_j is None: + continue + denom = max(gsd_i, gsd_j) + if denom == 0: + continue + gsd_delta_fraction = abs(gsd_i - gsd_j) / denom + if gsd_delta_fraction > gsd_tolerance_fraction: + continue + d_m = distance_m( + centres[i][0], centres[i][1], centres[j][0], centres[j][1] + ) + if d_m <= footprint_tolerance_m: + pairs.append((tiles[i].tile_id, tiles[j].tile_id)) + return TileDedupReport( + duplicate_pairs=tuple(pairs), + footprint_tolerance_m=footprint_tolerance_m, + gsd_tolerance_fraction=gsd_tolerance_fraction, + ) + + +def _extract_gsd(tile: TileSpec) -> float | None: + """Pull GSD (resolution_m_per_px) from the tile's quality dict.""" + if not isinstance(tile.quality, dict): + return None + raw = tile.quality.get("resolution_m_per_px") + if isinstance(raw, (int, float)): + return float(raw) + return None + + +# ─────────────────────────── FT-P-17 / AC-4 ─────────────────────────── + + +@dataclass(frozen=True) +class TileUploadAckReport: + """AC-4 of FT-P-17: every generated tile uploaded with HTTP 202.""" + + generated_tile_ids: tuple[str, ...] + audit_tile_ids: tuple[str, ...] + missing_from_audit: tuple[str, ...] + + @property + def passes(self) -> bool: + if not self.generated_tile_ids: + return False + return not self.missing_from_audit + + +def evaluate_upload_acks( + generated_tiles: Sequence[TileSpec], + audit_entries: Sequence[dict], +) -> TileUploadAckReport: + """AC-4: every generated tile_id appears in the mock-suite-sat-service audit. + + The mock-suite-sat-service ``POST /tiles`` endpoint records HTTP 202 + responses to its run-scoped audit log; a tile that did not return + 202 (i.e., was rejected with 400 or any forced-5xx) is NOT in the + audit. So a tile_id present in ``generated_tiles`` but absent from + ``audit_entries`` is by construction a missing ack. + + ``audit_entries`` is the ``entries`` field of the JSON response from + ``GET /tiles/audit?run_id=``. + """ + generated_ids = tuple(t.tile_id for t in generated_tiles) + audit_ids = tuple( + e["tile_id"] for e in audit_entries if isinstance(e, dict) and "tile_id" in e + ) + audit_id_set = set(audit_ids) + missing = tuple(tid for tid in generated_ids if tid not in audit_id_set) + return TileUploadAckReport( + generated_tile_ids=generated_ids, + audit_tile_ids=audit_ids, + missing_from_audit=missing, + ) + + +# ─────────────────────────── FT-N-06 / AC-5 ─────────────────────────── + + +@dataclass(frozen=True) +class CaptureDateFreshnessEntryReport: + """Per-tile drift between ``capture_utc`` and ``generated_at``. + + Whether the drift passes the AC threshold is decided at the + ``CaptureDateFreshnessReport`` level because the tolerance is a + report-wide knob (AC-5 stipulates 60 s globally). + """ + + tile_id: str + drift_s: float | None # None when capture_utc cannot be parsed + + +@dataclass(frozen=True) +class CaptureDateFreshnessReport: + """AC-5 of FT-N-06: |capture_utc - generated_at_wall_clock| ≤ 60 s.""" + + entries: tuple[CaptureDateFreshnessEntryReport, ...] + tolerance_s: float = CAPTURE_DATE_FRESHNESS_TOLERANCE_S + + @property + def failing_entries(self) -> tuple[CaptureDateFreshnessEntryReport, ...]: + return tuple( + e for e in self.entries + if e.drift_s is None or abs(e.drift_s) > self.tolerance_s + ) + + @property + def passes(self) -> bool: + if not self.entries: + return False + return not self.failing_entries + + +def evaluate_capture_date_freshness( + tiles: Sequence[TileSpec], + *, + tolerance_s: float = CAPTURE_DATE_FRESHNESS_TOLERANCE_S, +) -> CaptureDateFreshnessReport: + """AC-5: per-tile capture_utc drift against generated_at_monotonic_ms. + + Drift is signed: ``capture_utc − generated_at``. A drift of +0 is + "capture happened at generation"; negative drift means capture + happened BEFORE generation (the usual direction — capture is + instantaneous, generation is the orthorectification step that + follows). + + A tile whose ``capture_utc`` cannot be parsed as ISO 8601 records + drift_s = None and fails the AC. + """ + if tolerance_s <= 0: + raise ValueError(f"tolerance_s must be > 0, got {tolerance_s}") + entries: list[CaptureDateFreshnessEntryReport] = [] + for tile in tiles: + capture_str = tile.capture_utc_iso + if capture_str is None and isinstance(tile.quality, dict): + raw = tile.quality.get("capture_utc") + if isinstance(raw, str): + capture_str = raw + drift: float | None + if capture_str is None: + drift = None + else: + parsed = _parse_iso8601_utc_seconds(capture_str) + if parsed is None: + drift = None + else: + drift = parsed - (tile.generated_at_monotonic_ms / 1000.0) + entries.append( + CaptureDateFreshnessEntryReport(tile_id=tile.tile_id, drift_s=drift) + ) + return CaptureDateFreshnessReport( + entries=tuple(entries), tolerance_s=tolerance_s + ) + + +def _parse_iso8601_utc_seconds(ts: str) -> float | None: + """Parse ISO 8601 ``ts`` into seconds-since-epoch; ``None`` on failure. + + Accepts the trailing ``Z`` shorthand that ``datetime.fromisoformat`` + did not accept until 3.11. + """ + try: + normalised = ts[:-1] + "+00:00" if ts.endswith("Z") else ts + dt = datetime.fromisoformat(normalised) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.timestamp() + except (TypeError, ValueError): + return None + + +# ─────────────────────────── FT-N-06 / AC-6 ─────────────────────────── + + +@dataclass(frozen=True) +class FreshnessGateReport: + """AC-6 of FT-N-06: no `tile-load-rejected: stale` for freshly generated tiles.""" + + generated_tile_ids: tuple[str, ...] + stale_rejections: tuple[str, ...] + + @property + def passes(self) -> bool: + return not self.stale_rejections + + +def evaluate_freshness_gate( + generated_tiles: Sequence[TileSpec], + fdr_rejection_records: Iterable[dict], + *, + stale_reason: str = TILE_LOAD_REJECTED_STALE_REASON, +) -> FreshnessGateReport: + """AC-6: any ``tile-load-rejected: stale`` for a freshly generated tile fails. + + ``fdr_rejection_records`` is the payload dict of each FDR record whose + ``record_type == TILE_LOAD_REJECTED_FDR_KIND``. A "stale" rejection + sets ``reason == "stale"``. If the rejected tile_id matches a + generated tile_id, the freshness gate misclassified it. + """ + generated_ids = tuple(t.tile_id for t in generated_tiles) + gen_id_set = set(generated_ids) + stale: list[str] = [] + for payload in fdr_rejection_records: + if not isinstance(payload, dict): + continue + reason = payload.get("reason") + if reason != stale_reason: + continue + tile_id = payload.get("id") or payload.get("tile_id") + if isinstance(tile_id, str) and tile_id in gen_id_set: + stale.append(tile_id) + return FreshnessGateReport( + generated_tile_ids=generated_ids, stale_rejections=tuple(stale) + ) diff --git a/e2e/runner/helpers/mock_suite_sat_audit.py b/e2e/runner/helpers/mock_suite_sat_audit.py new file mode 100644 index 0000000..5d492d0 --- /dev/null +++ b/e2e/runner/helpers/mock_suite_sat_audit.py @@ -0,0 +1,76 @@ +"""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 diff --git a/e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py b/e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py new file mode 100644 index 0000000..155fb98 --- /dev/null +++ b/e2e/tests/negative/test_ft_n_06_mid_flight_freshness.py @@ -0,0 +1,126 @@ +"""FT-N-06 — Mid-flight tile current-timestamp + fresh-treatment (AZ-422 / AC-NEW-6). + +The full scenario: + +1. Same 5 min Derkachi replay as FT-P-17; the SUT generates one + FDR ``mid-flight-tile-output`` record per tile. +2. Inspect each tile's manifest entry: + * AC-5: ``|capture_utc - generated_at_monotonic_ms| ≤ 60 s``. + * AC-6: no FDR ``tile-load-rejected`` record with + ``reason == "stale"`` carries any of the generated tile IDs + (a fresh tile must not be misclassified by the freshness gate). + +Gated on: + +* ``sitl_replay_ready`` — full replay requires the SITL fixture. +* ``runner.helpers.mid_flight_tile_evaluator`` — pure-logic + evaluator covered by + ``e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py``. + +This is a "negative" test in the sense that it asserts a *non*-event: +no stale rejection of a freshly generated tile. The test still skips +cleanly when the SITL fixture is not prepared. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from runner.helpers import mid_flight_tile_evaluator as mfe + + +@pytest.mark.traces_to("AC-NEW-6,AC-5,AC-6,AC-7") +def test_ft_n_06_mid_flight_freshness( + fc_adapter: str, + vio_strategy: str, + evidence_dir, # type: ignore[no-untyped-def] + run_id: str, + nfr_recorder, # type: ignore[no-untyped-def] + sitl_replay_ready: bool, +) -> None: + """Full FT-N-06 scenario (AC-NEW-6).""" + if not sitl_replay_ready: + pytest.skip( + "FT-N-06 requires `E2E_SITL_REPLAY_DIR` to point at a SITL replay " + "fixture exposing `mid-flight-tile-output` FDR records and any " + "`tile-load-rejected` events emitted by the freshness gate " + "(AZ-595 + AZ-422 fixture builder). Pure-logic AC-NEW-6 " + "coverage lives in " + "e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py." + ) + + from runner.helpers import fdr_reader + + fdr_root = Path(evidence_dir).parent / f"run-{run_id}" / "fdr" + + tiles: list[mfe.TileSpec] = [] + rejection_payloads: list[dict] = [] + for rec in fdr_reader.iter_records(fdr_root): + if rec.record_type == mfe.MID_FLIGHT_TILE_FDR_KIND: + tile = _project_tile(rec) + if tile is not None: + tiles.append(tile) + elif rec.record_type == mfe.TILE_LOAD_REJECTED_FDR_KIND: + rejection_payloads.append(dict(rec.payload)) + + if not tiles: + pytest.fail( + f"FT-N-06: no `{mfe.MID_FLIGHT_TILE_FDR_KIND}` FDR records at " + f"{fdr_root}. The fixture builder must produce at least one " + "generated tile for the freshness/stale check to be meaningful." + ) + + capture_report = mfe.evaluate_capture_date_freshness(tiles) + freshness_report = mfe.evaluate_freshness_gate(tiles, rejection_payloads) + + nfr_recorder.record_metric( + "ft_n_06.tile_count", float(len(tiles)), ac_id="AC-NEW-6" + ) + nfr_recorder.record_metric( + "ft_n_06.capture_drift_failures", + float(len(capture_report.failing_entries)), + ac_id="AC-5", + ) + nfr_recorder.record_metric( + "ft_n_06.stale_rejection_count", + float(len(freshness_report.stale_rejections)), + ac_id="AC-6", + ) + + assert capture_report.passes, ( + f"AC-5 (|capture_utc - generated_at| ≤ {capture_report.tolerance_s} s) failed: " + f"failures={[(e.tile_id, e.drift_s) for e in capture_report.failing_entries]}" + ) + assert freshness_report.passes, ( + "AC-6 (no `tile-load-rejected: stale` for freshly generated tile) failed: " + f"stale_rejected_tile_ids={freshness_report.stale_rejections}" + ) + + +def _project_tile(rec) -> mfe.TileSpec | None: # type: ignore[no-untyped-def] + """Project an FDR record onto a ``TileSpec``; ``None`` if malformed.""" + p = rec.payload + try: + bbox = tuple(p["bbox_wgs84"]) # type: ignore[index] + except (KeyError, TypeError): + return None + if len(bbox) != 4: + return None + quality = p.get("quality") if isinstance(p.get("quality"), dict) else {} + capture_utc: str | None = None + if isinstance(quality, dict): + raw_capture = quality.get("capture_utc") + if isinstance(raw_capture, str): + capture_utc = raw_capture + return mfe.TileSpec( + tile_id=str(p.get("tile_id") or ""), + bbox_wgs84=(float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])), + zoom_level=int(p.get("zoom_level") or 0), + descriptor_sha256=str(p.get("descriptor_sha256") or ""), + payload_size_bytes=int(p.get("payload_size_bytes") or 0), + quality=dict(quality) if isinstance(quality, dict) else {}, + generated_at_monotonic_ms=int(rec.monotonic_ms), + capture_utc_iso=capture_utc, + ) diff --git a/e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py b/e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py new file mode 100644 index 0000000..4d0ea18 --- /dev/null +++ b/e2e/tests/positive/test_ft_p_17_mid_flight_tiles.py @@ -0,0 +1,182 @@ +"""FT-P-17 — Mid-flight tile generation + landing-time upload (AZ-422 / AC-8.4). + +The full scenario: + +1. The SUT cold-starts against an empty ``mid-flight-tile-output/`` + FDR directory + the bind-mounted Derkachi fixture. +2. Replay 5 min of Derkachi at the SUT's runtime cadence. While the + SUT generates orthorectified tiles it writes one FDR record per + tile under ``mid-flight-tile-output`` carrying every field the + mock-suite-sat-service ingest schema requires (Mode B Fact #105). +3. After replay, the test simulates a landing event (mechanism is + public-input — ``simulate_landing()`` MAVLink command, owned by + AZ-595 fixture builder); the SUT then uploads every generated + tile to ``mock-suite-sat-service``. +4. The test parses the FDR archive for generated tiles, fetches the + mock-service audit log, and asserts: + * AC-1: ≥ 1 tile per ~3 s of high-quality nav frames. + * AC-2: every tile has all Mode B Fact #105 fields populated. + * AC-3: no two tiles share footprint within ±1 m AND GSD within ±5 %. + * AC-4: every generated tile_id is in the audit log (HTTP 202). + * AC-7: parameterised across ``(fc_adapter, vio_strategy)``. + +FT-N-06 (AC-5/AC-6) is a separate file: ``test_ft_n_06_mid_flight_freshness.py``. + +Gated on: + +* ``sitl_replay_ready`` — full replay requires the SITL fixture. +* ``runner.helpers.mid_flight_tile_evaluator`` — pure-logic evaluator + covered by ``e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py``. +* ``runner.helpers.mock_suite_sat_audit.fetch_audit`` — HTTP wrapper + covered by ``e2e/_unit_tests/helpers/test_mock_suite_sat_audit.py``. +* ``FT_P_17_HIGH_QUALITY_WINDOW_S_ENV`` — the fixture builder records + the total wall-clock seconds of high-quality nav frames produced + by the replay (per AC-2.1a normal-segment criterion). Without this + env var the scenario can't compute the AC-1 denominator and skips. +""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from runner.helpers import mid_flight_tile_evaluator as mfe +from runner.helpers import mock_suite_sat_audit + +FT_P_17_HIGH_QUALITY_WINDOW_S_ENV = "FT_P_17_HIGH_QUALITY_WINDOW_S" + + +@pytest.mark.traces_to("AC-8.4,AC-1,AC-2,AC-3,AC-4,AC-7") +def test_ft_p_17_mid_flight_tiles( + fc_adapter: str, + vio_strategy: str, + evidence_dir, # type: ignore[no-untyped-def] + run_id: str, + nfr_recorder, # type: ignore[no-untyped-def] + sitl_replay_ready: bool, + mock_suite_sat_url: str, +) -> None: + """Full FT-P-17 scenario (AC-8.4).""" + if not sitl_replay_ready: + pytest.skip( + "FT-P-17 requires `E2E_SITL_REPLAY_DIR` to point at a SITL replay " + "fixture exposing `mid-flight-tile-output` FDR records and the " + "post-landing audit population on mock-suite-sat-service " + "(AZ-595 + AZ-422 fixture builder). Pure-logic AC-8.4 coverage " + "lives in e2e/_unit_tests/helpers/test_mid_flight_tile_evaluator.py." + ) + + high_quality_window_s_str = os.environ.get(FT_P_17_HIGH_QUALITY_WINDOW_S_ENV) + if not high_quality_window_s_str: + pytest.skip( + f"FT-P-17 needs `{FT_P_17_HIGH_QUALITY_WINDOW_S_ENV}` env var " + "(total wall-clock seconds of high-quality nav frames per " + "AC-2.1a). The fixture builder records this from the replay's " + "segment-quality FDR records." + ) + try: + high_quality_window_s = float(high_quality_window_s_str) + except ValueError as exc: + pytest.fail( + f"FT-P-17: `{FT_P_17_HIGH_QUALITY_WINDOW_S_ENV}` must parse as " + f"float; got {high_quality_window_s_str!r}: {exc}" + ) + + from runner.helpers import fdr_reader + + fdr_root = Path(evidence_dir).parent / f"run-{run_id}" / "fdr" + tiles = list(_extract_tiles_from_fdr(fdr_reader, fdr_root)) + if not tiles: + pytest.fail( + f"FT-P-17: no `{mfe.MID_FLIGHT_TILE_FDR_KIND}` FDR records under " + f"{fdr_root}. The SUT must generate at least one tile per AC-1." + ) + + audit_entries = mock_suite_sat_audit.fetch_audit(mock_suite_sat_url, run_id=run_id) + + rate_report = mfe.evaluate_tile_generation_rate(tiles, high_quality_window_s) + quality_report = mfe.evaluate_tile_quality_metadata(tiles) + dedup_report = mfe.evaluate_dedup(tiles) + upload_report = mfe.evaluate_upload_acks(tiles, audit_entries) + + nfr_recorder.record_metric( + "ft_p_17.tile_count", float(rate_report.tile_count), ac_id="AC-1" + ) + nfr_recorder.record_metric( + "ft_p_17.observed_rate_per_3s", rate_report.observed_rate_per_3s, ac_id="AC-1" + ) + nfr_recorder.record_metric( + "ft_p_17.high_quality_window_s", high_quality_window_s, ac_id="AC-1" + ) + nfr_recorder.record_metric( + "ft_p_17.tile_quality_failures", + float(len(quality_report.failing_entries)), + ac_id="AC-2", + ) + nfr_recorder.record_metric( + "ft_p_17.dedup_duplicate_pairs", + float(dedup_report.duplicate_count), + ac_id="AC-3", + ) + nfr_recorder.record_metric( + "ft_p_17.audit_missing_count", + float(len(upload_report.missing_from_audit)), + ac_id="AC-4", + ) + + assert rate_report.passes, ( + f"AC-1 (≥1 tile per {mfe.MIN_TILES_PER_HIGH_QUALITY_WINDOW_S} s) failed: " + f"{rate_report.tile_count} tiles over {high_quality_window_s} s " + f"high-quality window → rate={rate_report.observed_rate_per_3s:.3f}/3s" + ) + assert quality_report.passes, ( + "AC-2 (every tile has Mode B Fact #105 quality fields) failed: " + f"failures={[(e.tile_id, e.missing_top_level_fields, e.missing_quality_fields) for e in quality_report.failing_entries]}" + ) + assert dedup_report.passes, ( + "AC-3 (no duplicate footprint+GSD bins) failed: " + f"duplicate_pairs={dedup_report.duplicate_pairs}" + ) + assert upload_report.passes, ( + "AC-4 (landing-event upload accepted) failed: " + f"generated={len(upload_report.generated_tile_ids)}, " + f"audited={len(upload_report.audit_tile_ids)}, " + f"missing={upload_report.missing_from_audit}" + ) + + +def _extract_tiles_from_fdr(fdr_reader, fdr_root: Path): # type: ignore[no-untyped-def] + """Yield ``TileSpec``s from every ``mid-flight-tile-output`` FDR record. + + Each record's payload mirrors the mock-suite-sat-service TilePublishRequest + shape; the scenario only projects it onto a ``TileSpec`` and lets the + evaluators do the AC math. + """ + for rec in fdr_reader.iter_records(fdr_root): + if rec.record_type != mfe.MID_FLIGHT_TILE_FDR_KIND: + continue + p = rec.payload + try: + bbox = tuple(p["bbox_wgs84"]) # type: ignore[index] + except (KeyError, TypeError): + continue + if len(bbox) != 4: + continue + quality = p.get("quality") if isinstance(p.get("quality"), dict) else {} + capture_utc: str | None = None + if isinstance(quality, dict): + raw_capture = quality.get("capture_utc") + if isinstance(raw_capture, str): + capture_utc = raw_capture + yield mfe.TileSpec( + tile_id=str(p.get("tile_id") or ""), + bbox_wgs84=(float(bbox[0]), float(bbox[1]), float(bbox[2]), float(bbox[3])), + zoom_level=int(p.get("zoom_level") or 0), + descriptor_sha256=str(p.get("descriptor_sha256") or ""), + payload_size_bytes=int(p.get("payload_size_bytes") or 0), + quality=dict(quality) if isinstance(quality, dict) else {}, + generated_at_monotonic_ms=int(rec.monotonic_ms), + capture_utc_iso=capture_utc, + )