azaion/gps-denied-onboard
1
0
Fork 0
mirror of https://github.com/azaion/gps-denied-onboard.git synced 2026-06-21 07:01:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

[AZ-303] C6 storage interfaces: Protocols + DTOs + factories

Browse Source 
Freezes the c6_tile_cache Public API per
_docs/02_document/contracts/c6_tile_cache/{tile_store,tile_metadata_store,
descriptor_index}.md v1.0.0:

- Three runtime_checkable Protocols (TileStore 4-method, TileMetadataStore
  9-method, DescriptorIndex 5-method) in components/c6_tile_cache/interface.py.
- Frozen DTOs + enums (TileId, TileMetadata, TileMetadataPersistent,
  TileQualityMetadata, Bbox, SectorBoundary, HnswParams, IndexMetadata,
  TileSource, FreshnessLabel, VotingStatus, SectorClassification) in
  components/c6_tile_cache/_types.py. Constructor-time validation rejects
  out-of-range zoom_level / lat / lon and inverted Bbox.
- TilePixelHandle ABC for read-only mmap access (Invariant I-4).
- TileCacheError family (6 subtypes) + IndexBuildError (deliberately
  outside the family) in components/c6_tile_cache/errors.py.
- C6TileCacheConfig per-component config block, registered on package
  import; validates known runtime labels at construction time.
- Composition-root factories build_tile_store / build_tile_metadata_store /
  build_descriptor_index in runtime_root/storage_factory.py, with lazy
  concrete-impl imports gated by BUILD_FAISS_INDEX (AC-5 / Risk 2:
  no module-level FAISS import when the flag is OFF).
- RuntimeNotAvailableError defined in runtime_root/errors.py to be shared
  with AZ-297 (composition-time error, distinct from per-component
  runtime errors).

51 conformance tests cover all 10 ACs + NFR-perf-factory (p99 build_*
under 50 ms across 1000 calls) + NFR-reliability-error-family. AC-9
introspects each contract file's Shape table and asserts method
parity against the runtime Protocol.

Retired the AZ-263 scaffolding SectorClassification (dataclass) and
TileQualityMetadata from _types/tile.py since their canonical home is
now c6_tile_cache._types; Tile and TileRecord remain in _types/tile.py
until c3_matcher (AZ-344) and c11_tile_manager (AZ-316/319) retire
their interface stubs.

Full unit-test sweep: 791 passed, 2 pre-existing environment skips.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Oleksandr Bezdieniezhnykh
2026-05-12 04:21:44 +03:00
parent 48281db9e9
commit f925af9de3
12 changed files with 1539 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
_docs/02_tasks/todo/AZ-303_c6_storage_interfaces.md → _docs/02_tasks/done/AZ-303_c6_storage_interfaces.md
View File
src/gps_denied_onboard/_types/tile.py
+25 -24
View File
@@ -1,4 +1,18 @@
"""C6 tile-cache DTOs."""
"""Legacy C6 tile-cache scaffolding DTOs.
AZ-303 froze the canonical C6 contract; ``TileQualityMetadata`` and
``SectorClassification`` (as a dataclass) moved into
``gps_denied_onboard.components.c6_tile_cache._types`` (the former
matches 1:1; the latter became a ``str, Enum``).
The two remaining types here (``Tile``, ``TileRecord``) are still
referenced by the C3 / C11 scaffolding Protocols
(``components/c3_matcher/interface.py``,
``components/c11_tile_manager/interface.py``). Their own component
tasks (AZ-344 for C3, AZ-316 / AZ-319 for C11) will replace those
Protocol stubs with the AZ-303 DTOs; this file disappears in the
same migration step.
"""
from __future__ import annotations
@@ -9,7 +23,11 @@ from typing import Any
@dataclass(frozen=True)
class Tile:
"""A single satellite tile (image body + metadata)."""
"""Scaffolding satellite-tile DTO used by the C3 matcher Protocol stub.
Superseded by :class:`gps_denied_onboard.components.c6_tile_cache.TileMetadata`;
held here only until AZ-344 retires the C3 scaffolding.
"""
tile_id: str
zoom_level: int
@@ -20,20 +38,13 @@ class Tile:
image_path: str
@dataclass(frozen=True)
class TileQualityMetadata:
"""Quality metadata attached to an onboard-ingested tile (D-PROJ-2 ingest contract)."""
estimator_label: str
covariance_2x2: tuple[tuple[float, float], tuple[float, float]]
last_anchor_age_ms: int
mre_px: float
imu_bias_norm: float
@dataclass(frozen=True)
class TileRecord:
"""Postgres row for a tile (mirrors satellite-provider's canonical columns + additive)."""
"""Scaffolding Postgres row used by the C11 manager Protocol stub.
Superseded by :class:`gps_denied_onboard.components.c6_tile_cache.TileMetadata`;
held here only until AZ-316 / AZ-319 retire the C11 scaffolding.
"""
tile_id: str
zoom_level: int
@@ -46,14 +57,4 @@ class TileRecord:
flight_id: str | None = None
companion_id: str | None = None
capture_timestamp: datetime | None = None
quality: TileQualityMetadata | None = None
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class SectorClassification:
"""Operator-set classification of a geographic sector (urban / forest / agriculture / …)."""
sector_id: str
classification: str
freshness_threshold_days: int
src/gps_denied_onboard/components/c6_tile_cache/__init__.py
+71 -6
View File
@@ -1,21 +1,86 @@
"""C6 Tile Cache & Vector Index component — Public API."""
"""C6 Tile Cache & Vector Index — Public API (AZ-303).
from gps_denied_onboard._types.tile import (
Per ``tile_store.md`` / ``tile_metadata_store.md`` / ``descriptor_index.md``
v1.0.0, the public surface consists of:
- Three Protocols: :class:`TileStore`, :class:`TileMetadataStore`,
:class:`DescriptorIndex`.
- DTOs: :class:`TileId`, :class:`TileMetadata`, :class:`TileMetadataPersistent`,
:class:`TileQualityMetadata`, :class:`Bbox`, :class:`SectorBoundary`,
:class:`HnswParams`, :class:`IndexMetadata`.
- Enums: :class:`TileSource`, :class:`FreshnessLabel`, :class:`VotingStatus`,
:class:`SectorClassification`.
- ABC: :class:`TilePixelHandle`.
- Errors: the :class:`TileCacheError` family + :class:`IndexBuildError`.
- Config block: :class:`C6TileCacheConfig` (registered on import).
Concrete impls (``PostgresFilesystemStore``, ``FaissDescriptorIndex``)
live in sibling modules and are imported lazily by the composition root
factories at ``runtime_root.storage_factory`` — Risk-2 mitigation: this
``__init__.py`` MUST NOT import any concrete impl module.
"""
from gps_denied_onboard.components.c6_tile_cache._tile_pixel_handle import (
TilePixelHandle,
)
from gps_denied_onboard.components.c6_tile_cache._types import (
Bbox,
FreshnessLabel,
HnswParams,
IndexMetadata,
SectorBoundary,
SectorClassification,
Tile,
TileId,
TileMetadata,
TileMetadataPersistent,
TileQualityMetadata,
TileRecord,
TileSource,
VotingStatus,
)
from gps_denied_onboard.components.c6_tile_cache.config import C6TileCacheConfig
from gps_denied_onboard.components.c6_tile_cache.errors import (
ContentHashMismatchError,
FreshnessRejectionError,
IndexBuildError,
IndexUnavailableError,
TileCacheError,
TileFsError,
TileMetadataError,
TileNotFoundError,
)
from gps_denied_onboard.components.c6_tile_cache.interface import (
DescriptorIndex,
TileMetadataStore,
TileStore,
)
from gps_denied_onboard.config.schema import register_component_block
register_component_block("c6_tile_cache", C6TileCacheConfig)
__all__ = [
"Bbox",
"C6TileCacheConfig",
"ContentHashMismatchError",
"DescriptorIndex",
"FreshnessLabel",
"FreshnessRejectionError",
"HnswParams",
"IndexBuildError",
"IndexMetadata",
"IndexUnavailableError",
"SectorBoundary",
"SectorClassification",
"Tile",
"TileCacheError",
"TileFsError",
"TileId",
"TileMetadata",
"TileMetadataError",
"TileMetadataPersistent",
"TileMetadataStore",
"TileNotFoundError",
"TilePixelHandle",
"TileQualityMetadata",
"TileRecord",
"TileSource",
"TileStore",
"VotingStatus",
]
src/gps_denied_onboard/components/c6_tile_cache/_tile_pixel_handle.py
+51
View File
@@ -0,0 +1,51 @@
"""``TilePixelHandle`` ABC for read-only mmap-backed JPEG access.
Concrete impls (e.g., ``PostgresFilesystemStore``'s internal
``_FilesystemTilePixelHandle``) subclass and provide an
``__enter__`` that returns a read-only :class:`memoryview` over the
mmap. Consumers use the handle in a ``with`` block; the underlying
mmap is unmapped on ``__exit__``.
Invariant I-4 of ``tile_store.md`` makes the read-only guarantee a
contract obligation: a writer that mutates through the memoryview
is a Critical Reliability finding at code-review time.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from pathlib import Path
from types import TracebackType
class TilePixelHandle(ABC):
"""Opaque read-only view of a tile's JPEG bytes.
Lifetime is bounded by the caller's ``with`` block. The handle
MUST NOT outlive its ``__exit__``; consumers MUST NOT cache the
:class:`memoryview` past the block.
"""
@property
@abstractmethod
def filesystem_path(self) -> Path:
"""Absolute path to the JPEG file backing this handle.
Used only by C12 operator tooling (post-flight inspection)
and the C11 ``TileUploader`` post-landing copy. In-flight
consumers MUST NOT open a second handle to the same path;
they MUST use this :class:`TilePixelHandle`.
"""
@abstractmethod
def __enter__(self) -> memoryview:
"""Return a read-only :class:`memoryview` over the JPEG bytes."""
@abstractmethod
def __exit__(
self,
exc_type: type[BaseException] | None,
exc_val: BaseException | None,
exc_tb: TracebackType | None,
) -> None:
"""Release the mmap. Always called by the ``with`` block."""
src/gps_denied_onboard/components/c6_tile_cache/_types.py
+241
View File
@@ -0,0 +1,241 @@
"""C6 tile cache DTOs + enums (AZ-303).
This module is the single source of truth for the data shapes the three
``c6_tile_cache`` Protocols (``TileStore``, ``TileMetadataStore``,
``DescriptorIndex``) exchange with consumers. Every DTO is
``@dataclass(frozen=True)`` per PEP 557; every enum is ``str, Enum``
so JSON / Postgres / FDR serialisation is trivial.
The contract files at ``_docs/02_document/contracts/c6_tile_cache/``
are the authoritative human-readable shape — this module mirrors them
1:1.
"""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from pathlib import Path
__all__ = [
"Bbox",
"FreshnessLabel",
"HnswParams",
"IndexMetadata",
"SectorBoundary",
"SectorClassification",
"TileId",
"TileMetadata",
"TileMetadataPersistent",
"TileQualityMetadata",
"TileSource",
"VotingStatus",
]
_MAX_ZOOM_LEVEL = 21 # satellite-provider legal range upper bound
class TileSource(str, Enum):
"""Where a tile originated.
``GOOGLEMAPS``: pre-flight download via ``satellite-provider``.
``ONBOARD_INGEST``: produced mid-flight by C5 orthorectification
(F4 path) and uploaded post-landing by C11.
"""
GOOGLEMAPS = "googlemaps"
ONBOARD_INGEST = "onboard_ingest"
class FreshnessLabel(str, Enum):
"""C6 freshness gate output.
See ``tile_metadata_store.md`` Invariants I-2 / I-3 for the
transitions; this enum only exposes the four legal states.
"""
FRESH = "fresh"
STALE_ACTIVE_CONFLICT = "stale_active_conflict"
STALE_REAR = "stale_rear"
DOWNGRADED = "downgraded"
class VotingStatus(str, Enum):
"""C6 voting state for an onboard-ingested tile.
Forward-only transitions per Invariant I-8 of
``tile_metadata_store.md``: ``PENDING → TRUSTED``,
``PENDING → REJECTED``, ``TRUSTED → REJECTED``. The
impl (NOT this task) enforces the transition table.
"""
PENDING = "pending"
TRUSTED = "trusted"
REJECTED = "rejected"
class SectorClassification(str, Enum):
"""Operator-set classification of a geographic sector.
Drives the C6 freshness gate. Set pre-flight by C12 against the
``sector_boundaries`` table; the metadata store reads sector rows
at insert-time only.
"""
ACTIVE_CONFLICT = "active_conflict"
STABLE_REAR = "stable_rear"
@dataclass(frozen=True)
class TileId:
"""Spatial identity of a Web-Mercator tile.
``(zoom_level, lat, lon)`` is the composite identity. ``(x, y)``
integer tile coordinates are derived by the same function
``satellite-provider`` uses (Invariant I-1 of ``tile_store.md``);
this DTO carries the WGS84 form because it is the consumer-facing
shape.
"""
zoom_level: int
lat: float
lon: float
def __post_init__(self) -> None:
if not 0 <= self.zoom_level <= _MAX_ZOOM_LEVEL:
raise ValueError(
f"TileId.zoom_level must be in [0, {_MAX_ZOOM_LEVEL}]; "
f"got {self.zoom_level}"
)
if not -90.0 <= self.lat <= 90.0:
raise ValueError(
f"TileId.lat must be in [-90.0, 90.0]; got {self.lat}"
)
if not -180.0 <= self.lon <= 180.0:
raise ValueError(
f"TileId.lon must be in [-180.0, 180.0]; got {self.lon}"
)
@dataclass(frozen=True)
class Bbox:
"""Axis-aligned WGS84 bounding box, inclusive on min, exclusive on max."""
min_lat: float
min_lon: float
max_lat: float
max_lon: float
def __post_init__(self) -> None:
if self.min_lat >= self.max_lat:
raise ValueError(
f"Bbox.min_lat ({self.min_lat}) must be < max_lat "
f"({self.max_lat})"
)
if self.min_lon >= self.max_lon:
raise ValueError(
f"Bbox.min_lon ({self.min_lon}) must be < max_lon "
f"({self.max_lon})"
)
@dataclass(frozen=True)
class TileQualityMetadata:
"""Quality metadata attached to an onboard-ingested tile.
Set at write-time by C5; consumed by C6's freshness gate (insert
path) and by C11's voting-status updater (post-landing). Field
semantics are documented in the D-PROJ-2 ingest contract.
"""
estimator_label: str
covariance_2x2: tuple[tuple[float, float], tuple[float, float]]
last_anchor_age_ms: int
mre_px: float
imu_bias_norm: float
@dataclass(frozen=True)
class TileMetadata:
"""Per-row tile metadata as understood by both ``TileStore`` and
``TileMetadataStore``.
``flight_id`` / ``companion_id`` / ``quality_metadata`` are set
only for :attr:`TileSource.ONBOARD_INGEST`; for
:attr:`TileSource.GOOGLEMAPS` they are ``None``. ``voting_status``
defaults to :attr:`VotingStatus.PENDING` for onboard ingest and
:attr:`VotingStatus.TRUSTED` for googlemaps (the impl applies the
default; this DTO does not).
"""
tile_id: TileId
tile_size_meters: float
tile_size_pixels: int
capture_timestamp: datetime
source: TileSource
content_sha256_hex: str
freshness_label: FreshnessLabel
flight_id: str | None
companion_id: str | None
quality_metadata: TileQualityMetadata | None
voting_status: VotingStatus
@dataclass(frozen=True)
class TileMetadataPersistent:
"""In-process view of LRU + disk-budget bookkeeping.
Returned ONLY by ``TileMetadataStore.lru_candidates`` and
constructed inside the impl when responding to
``record_lru_access`` / ``total_disk_bytes``. Consumers reading
by ``tile_id`` / by ``bbox`` get the plain :class:`TileMetadata`.
"""
metadata: TileMetadata
accessed_at: datetime
uploaded_at: datetime | None
disk_bytes: int
@dataclass(frozen=True)
class SectorBoundary:
"""A single operator-defined sector polygon (axis-aligned bbox today)."""
bbox: Bbox
classification: SectorClassification
@dataclass(frozen=True)
class HnswParams:
"""HNSW build hyperparameters.
Defaults track the FAISS-team baseline (``HNSW32`` + M=32 +
efConstruction=200 + efSearch=64); overriding any value is a
research-time concern not exercised in production.
"""
m: int = 32
ef_construction: int = 200
ef_search: int = 64
metric: str = "L2"
@dataclass(frozen=True)
class IndexMetadata:
"""Sidecar metadata block written next to the ``.index`` file.
Read by ``DescriptorIndex.index_metadata()`` at runtime; produced
by C10's pre-flight ``rebuild_from_descriptors`` together with the
AZ-280 ``.sha256`` sidecar.
"""
descriptor_dim: int
n_vectors: int
backbone_label: str
backbone_sha256_hex: str
built_at: datetime
hnsw_params: HnswParams
sidecar_sha256_hex: str
file_path: Path
src/gps_denied_onboard/components/c6_tile_cache/config.py
+85
View File
@@ -0,0 +1,85 @@
"""C6 tile cache config block (AZ-303).
Registered into ``config.components['c6_tile_cache']`` by the package
``__init__.py``. The composition-root factories
(``build_tile_store`` / ``build_tile_metadata_store`` /
``build_descriptor_index``) read this block to select the runtime and
locate the persistent root.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Final
from gps_denied_onboard.config.schema import ConfigError
__all__ = [
"C6TileCacheConfig",
"KNOWN_DESCRIPTOR_INDEX_RUNTIMES",
"KNOWN_METADATA_RUNTIMES",
"KNOWN_TILE_STORE_RUNTIMES",
]
KNOWN_TILE_STORE_RUNTIMES: Final[frozenset[str]] = frozenset({"postgres_filesystem"})
KNOWN_METADATA_RUNTIMES: Final[frozenset[str]] = frozenset({"postgres_filesystem"})
KNOWN_DESCRIPTOR_INDEX_RUNTIMES: Final[frozenset[str]] = frozenset({"faiss_hnsw"})
@dataclass(frozen=True)
class C6TileCacheConfig:
"""Per-component config for C6 tile cache.
``store_runtime`` / ``metadata_runtime`` are currently both
``"postgres_filesystem"`` (single concrete impl); the field
exists so a future SQLite Tier-0 dev runtime can be added
without rippling the contract.
``descriptor_index_runtime`` selects the vector index strategy
(today only ``"faiss_hnsw"``; gated by ``BUILD_FAISS_INDEX``).
``root_dir`` is the filesystem root for tile JPEGs +
``.index`` sidecars; layout is byte-identical to
``satellite-provider`` (Invariant I-1 of ``tile_store.md``).
``postgres_dsn`` is the PostgreSQL connection string the
metadata-store impl uses (e.g. ``postgresql://...``); empty
string means "use the runtime ``db_url`` from ``RuntimeConfig``"
(the impl will resolve at build time).
``lru_eviction_threshold_bytes`` is the disk-budget high-water
mark the cache-budget enforcer (separate task) uses to trigger
LRU eviction; default 10 GiB per the C6 description.
"""
store_runtime: str = "postgres_filesystem"
metadata_runtime: str = "postgres_filesystem"
descriptor_index_runtime: str = "faiss_hnsw"
root_dir: str = "/var/lib/gps-denied/tiles"
postgres_dsn: str = ""
lru_eviction_threshold_bytes: int = 10 * 1024**3
def __post_init__(self) -> None:
if self.store_runtime not in KNOWN_TILE_STORE_RUNTIMES:
raise ConfigError(
f"C6TileCacheConfig.store_runtime={self.store_runtime!r} not in "
f"{sorted(KNOWN_TILE_STORE_RUNTIMES)}"
)
if self.metadata_runtime not in KNOWN_METADATA_RUNTIMES:
raise ConfigError(
f"C6TileCacheConfig.metadata_runtime={self.metadata_runtime!r} not in "
f"{sorted(KNOWN_METADATA_RUNTIMES)}"
)
if self.descriptor_index_runtime not in KNOWN_DESCRIPTOR_INDEX_RUNTIMES:
raise ConfigError(
f"C6TileCacheConfig.descriptor_index_runtime="
f"{self.descriptor_index_runtime!r} not in "
f"{sorted(KNOWN_DESCRIPTOR_INDEX_RUNTIMES)}"
)
if not self.root_dir:
raise ConfigError("C6TileCacheConfig.root_dir must be non-empty")
if self.lru_eviction_threshold_bytes <= 0:
raise ConfigError(
f"C6TileCacheConfig.lru_eviction_threshold_bytes must be > 0; "
f"got {self.lru_eviction_threshold_bytes}"
)
src/gps_denied_onboard/components/c6_tile_cache/errors.py
+96
View File
@@ -0,0 +1,96 @@
"""C6 tile cache runtime error taxonomy (AZ-303).
All read-side errors raised by ``TileStore`` / ``TileMetadataStore`` /
``DescriptorIndex`` are subclasses of :class:`TileCacheError`. Consumers
catch the family; implementations rewrap third-party exceptions
(psycopg, FAISS C++, OS errors) into one of these types.
:class:`IndexBuildError` is intentionally NOT in the :class:`TileCacheError`
family — it is raised only by the offline ``rebuild_from_descriptors``
path (C10 pre-flight provisioning) and has different fault semantics
than the in-flight read envelope.
"""
from __future__ import annotations
__all__ = [
"ContentHashMismatchError",
"FreshnessRejectionError",
"IndexBuildError",
"IndexUnavailableError",
"TileCacheError",
"TileFsError",
"TileMetadataError",
"TileNotFoundError",
]
class TileCacheError(Exception):
"""Base class for the C6 read-side error family.
Consumers MUST be able to catch the family with a single
``except TileCacheError:`` and recover or surface to FDR.
"""
class TileNotFoundError(TileCacheError):
"""A tile lookup did not find the requested ``TileId`` on disk.
Distinct from :class:`TileMetadataError`: the metadata row is also
absent, so the cache is in a consistent (just-empty) state.
"""
class TileFsError(TileCacheError):
"""An OS-level filesystem error escaped from a Protocol method.
Always wraps the originating ``OSError`` via ``__cause__``. The
impl MUST rewrap; raw ``OSError`` MUST NOT escape the Protocol.
"""
class TileMetadataError(TileCacheError):
"""The metadata store violated a consistency invariant.
Includes: row present but JPEG missing (or vice-versa), duplicate
composite-key insert, backward voting-status transition. See
``tile_metadata_store.md`` Invariants I-1, I-8 for the full list.
"""
class ContentHashMismatchError(TileCacheError):
"""``sha256(tile_blob) != metadata.content_sha256_hex`` on ``write_tile``.
Bound to the cache-poisoning safety budget (D-C10-3 / AC-NEW-7);
the impl MUST NOT silently retry — surface to the caller (and FDR)
so the operator can investigate the source feed.
"""
class FreshnessRejectionError(TileCacheError):
"""The C6 freshness gate rejected an insert.
Raised when the tile's ``(lat, lon)`` falls in an ``ACTIVE_CONFLICT``
sector AND ``capture_timestamp < now() - active_conflict_max_age``.
See ``tile_metadata_store.md`` Invariant I-2.
"""
class IndexUnavailableError(TileCacheError):
"""The descriptor index could not satisfy a read.
Includes: mmap handle invalid, ``.index`` file missing, sidecar
``.sha256`` mismatched, query vector dimension does not match
``descriptor_dim()``. See ``descriptor_index.md`` Invariants
I-3 / I-6.
"""
class IndexBuildError(Exception):
"""Offline index build failed.
Intentionally NOT a :class:`TileCacheError` subclass — the build
path is the C10 pre-flight provisioning envelope, distinct from
the in-flight read envelope. C2 catches :class:`TileCacheError`;
C10 catches :class:`IndexBuildError`.
"""
src/gps_denied_onboard/components/c6_tile_cache/interface.py
+188 -15
View File
@@ -1,32 +1,205 @@
"""C6 `TileStore` + `DescriptorIndex` Protocols.
"""C6 tile cache Protocols (AZ-303).
Concrete impl: `PostgresFilesystemStore` (Postgres mirror + filesystem mmap +
FAISS HNSW). See `_docs/02_document/components/08_c6_tile_cache/`.
Three ``runtime_checkable`` ``typing.Protocol`` declarations:
- :class:`TileStore` — filesystem-resident JPEG body I/O.
- :class:`TileMetadataStore` — Postgres-backed spatial + LRU + voting bookkeeping.
- :class:`DescriptorIndex` — per-flight FAISS HNSW vector index.
The contract files at ``_docs/02_document/contracts/c6_tile_cache/``
are the authoritative human-readable shape; this module mirrors them
1:1. Concrete impls live in sibling modules
(``postgres_filesystem_store``, ``faiss_descriptor_index``); the
composition root selects between them via
``runtime_root.storage_factory``.
"""
from __future__ import annotations
from collections.abc import Iterable
from typing import Protocol
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Protocol, runtime_checkable
from gps_denied_onboard._types.tile import Tile, TileRecord
from gps_denied_onboard.components.c6_tile_cache._tile_pixel_handle import (
TilePixelHandle,
)
from gps_denied_onboard.components.c6_tile_cache._types import (
Bbox,
HnswParams,
IndexMetadata,
TileId,
TileMetadata,
TileMetadataPersistent,
TileSource,
VotingStatus,
)
if TYPE_CHECKING:
import numpy as np
__all__ = [
"DescriptorIndex",
"TileMetadataStore",
"TileStore",
]
@runtime_checkable
class TileStore(Protocol):
"""Tile metadata + body store (mirrors satellite-provider; cached locally)."""
"""JPEG body store. See ``tile_store.md`` v1.0.0.
def get(self, tile_id: str) -> Tile | None: ...
The four-method surface is intentionally minimal: read, write,
exists, delete. Spatial queries live on :class:`TileMetadataStore`;
descriptor lookups live on :class:`DescriptorIndex`.
"""
def query_by_lat_lon(
self, lat: float, lon: float, zoom: int, radius_m: float
) -> Iterable[TileRecord]: ...
def read_tile_pixels(self, tile_id: TileId) -> TilePixelHandle:
"""Return a read-only mmap handle to the tile's JPEG bytes.
def put(self, record: TileRecord) -> None: ...
Raises :class:`TileNotFoundError` if the tile is absent;
:class:`TileMetadataError` if the row exists but the JPEG file
is missing (or vice-versa — Invariant I-8).
"""
...
def write_tile(self, tile_blob: bytes, metadata: TileMetadata) -> None:
"""Atomically persist a JPEG body + its sidecar + metadata row.
Raises :class:`ContentHashMismatchError` if
``sha256(tile_blob) != metadata.content_sha256_hex``;
:class:`FreshnessRejectionError` if the C6 freshness gate
rejects the insert; :class:`TileFsError` /
:class:`TileMetadataError` on infrastructure failure.
"""
...
def tile_exists(self, tile_id: TileId) -> bool:
"""Cheap existence check (page-cache lookup). Does not load bytes."""
...
def delete_tile(self, tile_id: TileId) -> bool:
"""Remove the JPEG + sidecar; return ``True`` if a file was removed.
Returns ``False`` (no exception) if the tile was already
missing — the LRU evictor relies on this no-error path
(Invariant I-6).
"""
...
@runtime_checkable
class TileMetadataStore(Protocol):
"""Postgres-backed metadata + spatial index + LRU bookkeeping.
See ``tile_metadata_store.md`` v1.0.0 for the nine-method surface
and the I-1..I-9 invariants.
"""
def query_by_bbox(
self,
bbox: Bbox,
zoom: int,
*,
voting_filter: VotingStatus | None = None,
source_filter: TileSource | None = None,
) -> list[TileMetadata]:
"""Spatial query, optionally filtered by voting status / source."""
...
def insert_metadata(self, metadata: TileMetadata) -> None:
"""Insert a row; freshness gate runs atomically (Invariant I-7).
Raises :class:`FreshnessRejectionError` in
``ACTIVE_CONFLICT`` sectors when the tile is too old;
:class:`TileMetadataError` on composite-key collision.
"""
...
def update_voting_status(
self, tile_id: TileId, status: VotingStatus
) -> None:
"""Forward-only state transition. Backward transitions raise."""
...
def mark_uploaded(self, tile_id: TileId, uploaded_at: datetime) -> None:
"""Stamp ``uploaded_at`` so the row leaves ``pending_uploads``."""
...
def pending_uploads(self) -> list[TileMetadata]:
"""Rows where ``source == ONBOARD_INGEST`` and ``uploaded_at IS NULL``.
Single source of truth for C11 ``TileUploader`` (Invariant I-9);
the uploader MUST NOT scan the filesystem.
"""
...
def record_lru_access(
self, tile_id: TileId, accessed_at: datetime
) -> None:
"""Update ``accessed_at = max(current, supplied)`` (Invariant I-4)."""
...
def lru_candidates(
self, *, max_count: int
) -> list[TileMetadataPersistent]:
"""Return up to ``max_count`` oldest-``accessed_at``-first rows."""
...
def total_disk_bytes(self) -> int:
"""``SUM(disk_bytes) WHERE voting_status != REJECTED`` (Invariant I-5)."""
...
def get_by_id(self, tile_id: TileId) -> TileMetadata | None:
"""Point lookup; returns ``None`` if absent (NOT raises)."""
...
@runtime_checkable
class DescriptorIndex(Protocol):
"""Vector index over tile descriptors (FAISS HNSW concrete impl)."""
"""Per-flight FAISS HNSW vector index.
def add(self, tile_id: str, descriptor) -> None: ...
See ``descriptor_index.md`` v1.0.0 for the five-method surface
and the I-1..I-8 invariants. The single in-flight consumer is C2
VPR; the offline producer is C10 ``CacheProvisioner``.
"""
def search(self, descriptor, top_k: int) -> Iterable[tuple[str, float]]: ...
def search_topk(
self, query: "np.ndarray", k: int
) -> list[tuple[TileId, float]]:
"""Top-K nearest neighbour search.
``query`` must be ``(descriptor_dim,)`` ``float32`` C-contiguous;
dimension mismatch raises :class:`IndexUnavailableError`
(Invariant I-3). Returns ≤ k results, distance-ascending.
"""
...
def descriptor_dim(self) -> int:
"""Indexed-vector dimension; fixed at build time (Invariant I-3)."""
...
def mmap_handle(self) -> Path:
"""Absolute path to the ``.index`` file.
Used by C12 operator post-flight inspection tooling; in-flight
callers MUST go through ``search_topk``.
"""
...
def rebuild_from_descriptors(
self,
descriptors: "np.ndarray",
tile_ids: list[TileId],
hnsw_params: HnswParams,
) -> None:
"""Offline full-rebuild (C10 pre-flight; Invariant I-5 atomic).
Raises :class:`IndexBuildError` on dtype/shape mismatch or
underlying FAISS failure; :class:`TileFsError` on disk write
failure.
"""
...
def index_metadata(self) -> IndexMetadata:
"""Sidecar metadata block (Invariant I-6 sidecar coherence)."""
...
src/gps_denied_onboard/runtime_root/errors.py
+24
View File
@@ -0,0 +1,24 @@
"""Composition-root errors shared across runtime factories.
These are raised at composition time (``build_*`` factory entry) and
NOT during the running flight. Components own their per-runtime error
families; this module owns the cross-component selection error.
"""
from __future__ import annotations
class RuntimeNotAvailableError(RuntimeError):
"""Raised when ``build_*`` is asked for a runtime whose compile-time
``BUILD_*`` flag is OFF.
Used by:
- ``runtime_root.storage_factory.build_descriptor_index`` (AZ-303 / E-C6,
``BUILD_FAISS_INDEX`` gate)
- ``runtime_root.inference_factory.build_inference_runtime`` (AZ-297 / E-C7,
``BUILD_TENSORRT_RUNTIME`` / ``BUILD_ONNX_TRT_EP_RUNTIME`` /
``BUILD_PYTORCH_FP16_RUNTIME`` gates)
The message MUST name the requested runtime label so the operator can
correlate against ``.env``'s ``BUILD_*`` matrix without guessing.
"""
src/gps_denied_onboard/runtime_root/storage_factory.py
+150
View File
@@ -0,0 +1,150 @@
"""C6 storage composition-root factories (AZ-303).
Three factories selected by ``config.components['c6_tile_cache']``:
- :func:`build_tile_store` — JPEG body store.
- :func:`build_tile_metadata_store` — Postgres metadata + LRU + voting.
- :func:`build_descriptor_index` — FAISS HNSW vector index.
The factories lazily import concrete impl modules. A ``BUILD_*`` flag
that is OFF MUST NOT cause the concrete module to be imported — that
is the AC-5 / Risk-2 invariant. We honour it by reading the flag
from :mod:`os.environ` BEFORE the ``import`` statement.
The concrete impls (``postgres_filesystem_store``, ``faiss_descriptor_index``)
are produced by separate downstream tasks (AZ-305 / AZ-306). Until they
land, requesting any runtime raises :class:`RuntimeNotAvailableError`
with a message that names the missing impl module — the failure is
explicit, not silent.
"""
from __future__ import annotations
import os
from typing import TYPE_CHECKING
from gps_denied_onboard.runtime_root.errors import RuntimeNotAvailableError
if TYPE_CHECKING:
from gps_denied_onboard.components.c6_tile_cache import (
C6TileCacheConfig,
DescriptorIndex,
TileMetadataStore,
TileStore,
)
from gps_denied_onboard.config.schema import Config
__all__ = [
"build_descriptor_index",
"build_tile_metadata_store",
"build_tile_store",
]
def _is_build_flag_on(flag_name: str) -> bool:
"""Read a compile-time ``BUILD_*`` flag from the environment.
``ON`` / ``1`` / ``true`` / ``yes`` (case-insensitive) → ``True``.
Anything else (including unset) → ``False``. Defaults to OFF to
keep test envs honest about which runtimes they advertise.
"""
raw = os.environ.get(flag_name, "")
return raw.strip().lower() in {"on", "1", "true", "yes"}
def _c6_config(config: "Config") -> "C6TileCacheConfig":
"""Pull the registered C6 config block.
``c6_tile_cache.__init__`` registers it on import; if the package
has not been imported, :class:`KeyError` here is the right
failure mode — silent fallback would mask a missing import.
"""
return config.components["c6_tile_cache"]
def build_tile_store(config: "Config") -> "TileStore":
"""Construct the :class:`TileStore` impl selected by config.
Today only ``"postgres_filesystem"`` is wired; the runtime label
is validated at config-load time so unknown labels never reach
here. Concrete impl is produced by AZ-305.
"""
block = _c6_config(config)
runtime = block.store_runtime
if runtime == "postgres_filesystem":
try:
from gps_denied_onboard.components.c6_tile_cache.postgres_filesystem_store import ( # noqa: PLC0415
PostgresFilesystemStore,
)
except ModuleNotFoundError as exc:
raise RuntimeNotAvailableError(
f"TileStore runtime {runtime!r} is configured but its "
"concrete impl module "
"'c6_tile_cache.postgres_filesystem_store' has not been "
"built into this binary yet (AZ-305 pending)."
) from exc
return PostgresFilesystemStore(config)
raise RuntimeNotAvailableError(
f"TileStore runtime {runtime!r} is not buildable in this binary."
)
def build_tile_metadata_store(config: "Config") -> "TileMetadataStore":
"""Construct the :class:`TileMetadataStore` impl selected by config.
Today the same ``PostgresFilesystemStore`` class implements both
:class:`TileStore` and :class:`TileMetadataStore` (single-row
transactional store across the two surfaces). The factories
return the same instance shape but stay separate so a future
SQLite Tier-0 variant can swap one without the other.
"""
block = _c6_config(config)
runtime = block.metadata_runtime
if runtime == "postgres_filesystem":
try:
from gps_denied_onboard.components.c6_tile_cache.postgres_filesystem_store import ( # noqa: PLC0415
PostgresFilesystemStore,
)
except ModuleNotFoundError as exc:
raise RuntimeNotAvailableError(
f"TileMetadataStore runtime {runtime!r} is configured "
"but its concrete impl module "
"'c6_tile_cache.postgres_filesystem_store' has not been "
"built into this binary yet (AZ-305 pending)."
) from exc
return PostgresFilesystemStore(config)
raise RuntimeNotAvailableError(
f"TileMetadataStore runtime {runtime!r} is not buildable in this binary."
)
def build_descriptor_index(config: "Config") -> "DescriptorIndex":
"""Construct the :class:`DescriptorIndex` impl selected by config.
Gated by ``BUILD_FAISS_INDEX``: if the flag is OFF, the concrete
module is NOT imported (sys.modules invariant; AC-5) and
:class:`RuntimeNotAvailableError` is raised at composition time.
"""
block = _c6_config(config)
runtime = block.descriptor_index_runtime
if runtime == "faiss_hnsw":
if not _is_build_flag_on("BUILD_FAISS_INDEX"):
raise RuntimeNotAvailableError(
f"DescriptorIndex runtime {runtime!r} requires "
"BUILD_FAISS_INDEX=ON in this binary; the flag is OFF."
)
try:
from gps_denied_onboard.components.c6_tile_cache.faiss_descriptor_index import ( # noqa: PLC0415
FaissDescriptorIndex,
)
except ModuleNotFoundError as exc:
raise RuntimeNotAvailableError(
f"DescriptorIndex runtime {runtime!r} is configured but "
"its concrete impl module "
"'c6_tile_cache.faiss_descriptor_index' has not been "
"built into this binary yet (AZ-306 pending)."
) from exc
return FaissDescriptorIndex(config)
raise RuntimeNotAvailableError(
f"DescriptorIndex runtime {runtime!r} is not buildable in this binary."
)
tests/unit/c6_tile_cache/test_protocol_conformance.py
+608
View File
@@ -0,0 +1,608 @@
"""AZ-303 — C6 storage Protocol + DTO + error + factory conformance.
Covers all 10 ACs of AZ-303 (see ``_docs/02_tasks/todo/AZ-303_...``).
The factory ACs (AC-4 / AC-5) substitute lazy-importable fake impl
modules at ``sys.modules`` boundaries so the test never touches FAISS
or psycopg2.
"""
from __future__ import annotations
import dataclasses
import re
import sys
import types
from datetime import datetime, timezone
from pathlib import Path
import pytest
from gps_denied_onboard.components.c6_tile_cache import (
Bbox,
C6TileCacheConfig,
ContentHashMismatchError,
DescriptorIndex,
FreshnessLabel,
FreshnessRejectionError,
HnswParams,
IndexBuildError,
IndexMetadata,
IndexUnavailableError,
SectorBoundary,
SectorClassification,
TileCacheError,
TileFsError,
TileId,
TileMetadata,
TileMetadataError,
TileMetadataPersistent,
TileMetadataStore,
TileNotFoundError,
TilePixelHandle,
TileQualityMetadata,
TileSource,
TileStore,
VotingStatus,
)
from gps_denied_onboard.components.c6_tile_cache.config import (
KNOWN_DESCRIPTOR_INDEX_RUNTIMES,
)
from gps_denied_onboard.config.schema import Config, ConfigError
from gps_denied_onboard.runtime_root.errors import RuntimeNotAvailableError
from gps_denied_onboard.runtime_root.storage_factory import (
build_descriptor_index,
build_tile_metadata_store,
build_tile_store,
)
_CONTRACT_DIR = Path(__file__).resolve().parents[3] / (
"_docs/02_document/contracts/c6_tile_cache"
)
_FAKE_IMPL_MODULE = "gps_denied_onboard.components.c6_tile_cache.faiss_descriptor_index"
_FAKE_STORE_MODULE = (
"gps_denied_onboard.components.c6_tile_cache.postgres_filesystem_store"
)
def _valid_tile_id(zoom: int = 18, lat: float = 49.94, lon: float = 36.31) -> TileId:
return TileId(zoom_level=zoom, lat=lat, lon=lon)
def _valid_tile_metadata() -> TileMetadata:
return TileMetadata(
tile_id=_valid_tile_id(),
tile_size_meters=256.0,
tile_size_pixels=256,
capture_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc),
source=TileSource.GOOGLEMAPS,
content_sha256_hex="a" * 64,
freshness_label=FreshnessLabel.FRESH,
flight_id=None,
companion_id=None,
quality_metadata=None,
voting_status=VotingStatus.TRUSTED,
)
def _config_with_c6(overrides: dict[str, object] | None = None) -> Config:
overrides = overrides or {}
block = C6TileCacheConfig(**overrides) # type: ignore[arg-type]
return Config.with_blocks(c6_tile_cache=block)
# ----------------------------------------------------------------------
# AC-1: three Protocols are conformance-checkable.
class _FullTileStore:
def read_tile_pixels(self, tile_id):
raise NotImplementedError
def write_tile(self, tile_blob, metadata):
raise NotImplementedError
def tile_exists(self, tile_id):
raise NotImplementedError
def delete_tile(self, tile_id):
raise NotImplementedError
class _PartialTileStore:
def read_tile_pixels(self, tile_id):
raise NotImplementedError
def write_tile(self, tile_blob, metadata):
raise NotImplementedError
def tile_exists(self, tile_id):
raise NotImplementedError
class _FullTileMetadataStore:
def query_by_bbox(self, bbox, zoom, *, voting_filter=None, source_filter=None):
raise NotImplementedError
def insert_metadata(self, metadata):
raise NotImplementedError
def update_voting_status(self, tile_id, status):
raise NotImplementedError
def mark_uploaded(self, tile_id, uploaded_at):
raise NotImplementedError
def pending_uploads(self):
raise NotImplementedError
def record_lru_access(self, tile_id, accessed_at):
raise NotImplementedError
def lru_candidates(self, *, max_count):
raise NotImplementedError
def total_disk_bytes(self):
raise NotImplementedError
def get_by_id(self, tile_id):
raise NotImplementedError
class _PartialTileMetadataStore:
def query_by_bbox(self, bbox, zoom, *, voting_filter=None, source_filter=None):
raise NotImplementedError
class _FullDescriptorIndex:
def search_topk(self, query, k):
raise NotImplementedError
def descriptor_dim(self):
raise NotImplementedError
def mmap_handle(self):
raise NotImplementedError
def rebuild_from_descriptors(self, descriptors, tile_ids, hnsw_params):
raise NotImplementedError
def index_metadata(self):
raise NotImplementedError
class _PartialDescriptorIndex:
def search_topk(self, query, k):
raise NotImplementedError
def descriptor_dim(self):
raise NotImplementedError
def test_ac1_tile_store_conformance_full() -> None:
assert isinstance(_FullTileStore(), TileStore)
def test_ac1_tile_store_conformance_partial_missing_delete() -> None:
assert not isinstance(_PartialTileStore(), TileStore)
def test_ac1_tile_metadata_store_conformance_full() -> None:
assert isinstance(_FullTileMetadataStore(), TileMetadataStore)
def test_ac1_tile_metadata_store_conformance_partial() -> None:
assert not isinstance(_PartialTileMetadataStore(), TileMetadataStore)
def test_ac1_descriptor_index_conformance_full() -> None:
assert isinstance(_FullDescriptorIndex(), DescriptorIndex)
def test_ac1_descriptor_index_conformance_partial_missing_metadata() -> None:
assert not isinstance(_PartialDescriptorIndex(), DescriptorIndex)
# ----------------------------------------------------------------------
# AC-2: frozen DTOs reject mutation.
@pytest.mark.parametrize(
"dto, field_name, new_value",
[
(_valid_tile_id(), "lat", 0.0),
(_valid_tile_metadata(), "tile_size_meters", 9999.0),
(Bbox(min_lat=0.0, min_lon=0.0, max_lat=1.0, max_lon=1.0), "min_lat", 5.0),
(HnswParams(), "m", 64),
(
TileQualityMetadata(
estimator_label="satellite_anchored",
covariance_2x2=((0.1, 0.0), (0.0, 0.1)),
last_anchor_age_ms=100,
mre_px=0.5,
imu_bias_norm=0.01,
),
"mre_px",
1.0,
),
],
)
def test_ac2_frozen_dtos_reject_mutation(dto, field_name: str, new_value) -> None:
original_value = getattr(dto, field_name)
with pytest.raises(dataclasses.FrozenInstanceError):
setattr(dto, field_name, new_value)
assert getattr(dto, field_name) == original_value
# ----------------------------------------------------------------------
# AC-3: error hierarchy catchable as a single family.
@pytest.mark.parametrize(
"exc_factory",
[
TileNotFoundError,
TileFsError,
TileMetadataError,
ContentHashMismatchError,
FreshnessRejectionError,
IndexUnavailableError,
],
)
def test_ac3_all_runtime_errors_caught_as_family(exc_factory) -> None:
with pytest.raises(TileCacheError):
raise exc_factory("boom")
def test_ac3_unrelated_exception_not_caught_as_family() -> None:
with pytest.raises(ValueError):
try:
raise ValueError("not us")
except TileCacheError:
pytest.fail("ValueError must not be caught as TileCacheError")
def test_ac3_index_build_error_outside_family() -> None:
with pytest.raises(IndexBuildError):
try:
raise IndexBuildError("offline only")
except TileCacheError:
pytest.fail("IndexBuildError must NOT be in the TileCacheError family")
# ----------------------------------------------------------------------
# AC-4 + AC-5: factory honours config + BUILD flag gate.
@pytest.fixture
def faiss_module_cleanup():
"""Ensure no residual fake FAISS impl module leaks between tests."""
sys.modules.pop(_FAKE_IMPL_MODULE, None)
yield
sys.modules.pop(_FAKE_IMPL_MODULE, None)
@pytest.fixture
def store_module_cleanup():
sys.modules.pop(_FAKE_STORE_MODULE, None)
yield
sys.modules.pop(_FAKE_STORE_MODULE, None)
def _install_fake_faiss_impl_module() -> type:
"""Install a fake ``faiss_descriptor_index`` module in ``sys.modules``.
The fake's ``FaissDescriptorIndex`` class structurally satisfies the
:class:`DescriptorIndex` Protocol. We attach the class via
``types.ModuleType`` so the factory's lazy import succeeds.
"""
class _FakeFaissDescriptorIndex(_FullDescriptorIndex):
def __init__(self, config: Config) -> None:
self.config = config
fake_module = types.ModuleType(_FAKE_IMPL_MODULE)
fake_module.FaissDescriptorIndex = _FakeFaissDescriptorIndex # type: ignore[attr-defined]
sys.modules[_FAKE_IMPL_MODULE] = fake_module
return _FakeFaissDescriptorIndex
def _install_fake_postgres_store_module() -> type:
class _FakePostgresFilesystemStore(_FullTileStore, _FullTileMetadataStore):
def __init__(self, config: Config) -> None:
self.config = config
fake_module = types.ModuleType(_FAKE_STORE_MODULE)
fake_module.PostgresFilesystemStore = _FakePostgresFilesystemStore # type: ignore[attr-defined]
sys.modules[_FAKE_STORE_MODULE] = fake_module
return _FakePostgresFilesystemStore
def test_ac4_build_descriptor_index_returns_protocol_impl(
monkeypatch, faiss_module_cleanup
) -> None:
monkeypatch.setenv("BUILD_FAISS_INDEX", "ON")
fake_cls = _install_fake_faiss_impl_module()
config = _config_with_c6()
handle = build_descriptor_index(config)
assert isinstance(handle, fake_cls)
assert isinstance(handle, DescriptorIndex)
def test_ac5_build_descriptor_index_flag_off_raises_no_import(
monkeypatch, faiss_module_cleanup
) -> None:
monkeypatch.delenv("BUILD_FAISS_INDEX", raising=False)
config = _config_with_c6()
with pytest.raises(RuntimeNotAvailableError) as exc_info:
build_descriptor_index(config)
assert "faiss_hnsw" in str(exc_info.value)
assert _FAKE_IMPL_MODULE not in sys.modules
def test_ac4_build_tile_store_returns_protocol_impl(store_module_cleanup) -> None:
fake_cls = _install_fake_postgres_store_module()
config = _config_with_c6()
store = build_tile_store(config)
assert isinstance(store, fake_cls)
assert isinstance(store, TileStore)
def test_ac4_build_tile_metadata_store_returns_protocol_impl(
store_module_cleanup,
) -> None:
fake_cls = _install_fake_postgres_store_module()
config = _config_with_c6()
md = build_tile_metadata_store(config)
assert isinstance(md, fake_cls)
assert isinstance(md, TileMetadataStore)
def test_ac5_tile_store_runtime_module_missing_raises(store_module_cleanup) -> None:
config = _config_with_c6()
with pytest.raises(RuntimeNotAvailableError) as exc_info:
build_tile_store(config)
assert "postgres_filesystem" in str(exc_info.value)
# ----------------------------------------------------------------------
# AC-6: unknown runtime label rejected at config load.
def test_ac6_unknown_descriptor_index_runtime_rejected() -> None:
with pytest.raises(ConfigError) as exc_info:
C6TileCacheConfig(descriptor_index_runtime="scann")
msg = str(exc_info.value)
assert "scann" in msg
for valid in KNOWN_DESCRIPTOR_INDEX_RUNTIMES:
assert valid in msg
def test_ac6_unknown_store_runtime_rejected() -> None:
with pytest.raises(ConfigError):
C6TileCacheConfig(store_runtime="sqlite_filesystem")
def test_ac6_unknown_metadata_runtime_rejected() -> None:
with pytest.raises(ConfigError):
C6TileCacheConfig(metadata_runtime="sqlite_filesystem")
# ----------------------------------------------------------------------
# AC-7: constructor-time validation rejects bad input.
@pytest.mark.parametrize(
"kwargs, offending_field",
[
({"zoom_level": 22, "lat": 0.0, "lon": 0.0}, "zoom_level"),
({"zoom_level": -1, "lat": 0.0, "lon": 0.0}, "zoom_level"),
({"zoom_level": 18, "lat": 100.0, "lon": 0.0}, "lat"),
({"zoom_level": 18, "lat": 0.0, "lon": -200.0}, "lon"),
],
)
def test_ac7_tile_id_rejects_bad_input(
kwargs: dict[str, float], offending_field: str
) -> None:
with pytest.raises(ValueError) as exc_info:
TileId(**kwargs) # type: ignore[arg-type]
assert offending_field in str(exc_info.value)
@pytest.mark.parametrize(
"kwargs",
[
{"min_lat": 10.0, "min_lon": 0.0, "max_lat": 5.0, "max_lon": 10.0},
{"min_lat": 0.0, "min_lon": 10.0, "max_lat": 5.0, "max_lon": 5.0},
{"min_lat": 5.0, "min_lon": 5.0, "max_lat": 5.0, "max_lon": 10.0},
],
)
def test_ac7_bbox_rejects_inverted_or_degenerate(kwargs: dict[str, float]) -> None:
with pytest.raises(ValueError):
Bbox(**kwargs) # type: ignore[arg-type]
# ----------------------------------------------------------------------
# AC-8: TilePixelHandle is read-only by contract.
class _BytesTilePixelHandle(TilePixelHandle):
"""In-memory fake used by tests; mirrors the mmap impl's read-only contract."""
def __init__(self, blob: bytes, path: Path) -> None:
self._blob = blob
self._path = path
self._view: memoryview | None = None
@property
def filesystem_path(self) -> Path:
return self._path
def __enter__(self) -> memoryview:
self._view = memoryview(self._blob)
return self._view
def __exit__(self, exc_type, exc_val, exc_tb) -> None:
if self._view is not None:
self._view.release()
self._view = None
def test_ac8_tile_pixel_handle_returns_read_only_memoryview(tmp_path: Path) -> None:
blob = b"\xff\xd8\xff" + b"\x00" * 100 # JPEG SOI + filler
handle_path = tmp_path / "fake.jpg"
handle = _BytesTilePixelHandle(blob, handle_path)
with handle as memview:
with pytest.raises(TypeError):
memview[0] = 0xFF # type: ignore[index]
assert bytes(memview[:3]) == b"\xff\xd8\xff"
# ----------------------------------------------------------------------
# AC-9: contract files match Protocol shapes.
_METHOD_TABLE_RE = re.compile(r"^\|\s*`(?P<name>[a-z_][a-z0-9_]*)`\s*\|", re.MULTILINE)
def _methods_from_contract(contract_file: Path) -> set[str]:
"""Pull every backtick-quoted method name from the ``## Shape`` table."""
text = contract_file.read_text(encoding="utf-8")
shape_start = text.index("## Shape")
next_section = text.find("\n## ", shape_start + len("## Shape"))
shape_section = text[shape_start:next_section] if next_section != -1 else text[shape_start:]
return {m.group("name") for m in _METHOD_TABLE_RE.finditer(shape_section)}
def _protocol_methods(proto: type) -> set[str]:
"""Reflect over a Protocol's method names."""
return {
name
for name in dir(proto)
if not name.startswith("_") and callable(getattr(proto, name))
}
@pytest.mark.parametrize(
"contract_filename, proto",
[
("tile_store.md", TileStore),
("tile_metadata_store.md", TileMetadataStore),
("descriptor_index.md", DescriptorIndex),
],
)
def test_ac9_contract_methods_match_protocol(
contract_filename: str, proto: type
) -> None:
contract_path = _CONTRACT_DIR / contract_filename
contract_methods = _methods_from_contract(contract_path)
protocol_methods = _protocol_methods(proto)
missing_in_protocol = contract_methods - protocol_methods
missing_in_contract = protocol_methods - contract_methods
assert not missing_in_protocol, (
f"{contract_filename}: methods declared in contract but missing from "
f"Protocol: {sorted(missing_in_protocol)}"
)
assert not missing_in_contract, (
f"{contract_filename}: methods present on Protocol but missing from "
f"contract: {sorted(missing_in_contract)}"
)
# ----------------------------------------------------------------------
# AC-10: VotingStatus surface.
def test_ac10_voting_status_has_documented_states_only() -> None:
assert {v.value for v in VotingStatus} == {"pending", "trusted", "rejected"}
assert VotingStatus.PENDING.value == "pending"
assert VotingStatus.TRUSTED.value == "trusted"
assert VotingStatus.REJECTED.value == "rejected"
# ----------------------------------------------------------------------
# NFR-reliability-error-family + smoke surface tests.
@pytest.mark.parametrize(
"exc_type",
[
TileNotFoundError,
TileFsError,
TileMetadataError,
ContentHashMismatchError,
FreshnessRejectionError,
IndexUnavailableError,
],
)
def test_nfr_reliability_all_runtime_errors_subclass_family(exc_type) -> None:
assert issubclass(exc_type, TileCacheError)
def test_nfr_reliability_index_build_error_not_in_family() -> None:
assert not issubclass(IndexBuildError, TileCacheError)
def test_sector_classification_enum_surface() -> None:
assert {v.value for v in SectorClassification} == {
"active_conflict",
"stable_rear",
}
def test_sector_boundary_dto_constructs_and_freezes() -> None:
bbox = Bbox(min_lat=0.0, min_lon=0.0, max_lat=1.0, max_lon=1.0)
sector = SectorBoundary(bbox=bbox, classification=SectorClassification.ACTIVE_CONFLICT)
with pytest.raises(dataclasses.FrozenInstanceError):
sector.classification = SectorClassification.STABLE_REAR # type: ignore[misc]
def test_tile_metadata_persistent_dto_constructs_and_freezes() -> None:
md = _valid_tile_metadata()
persistent = TileMetadataPersistent(
metadata=md,
accessed_at=datetime(2026, 1, 2, tzinfo=timezone.utc),
uploaded_at=None,
disk_bytes=12345,
)
with pytest.raises(dataclasses.FrozenInstanceError):
persistent.disk_bytes = 0 # type: ignore[misc]
def test_nfr_perf_build_factories_under_50ms_p99(
monkeypatch, faiss_module_cleanup, store_module_cleanup
) -> None:
"""Factory triple p99 ≤ 50 ms across 1000 calls (NFR-perf-factory)."""
import time
monkeypatch.setenv("BUILD_FAISS_INDEX", "ON")
_install_fake_faiss_impl_module()
_install_fake_postgres_store_module()
config = _config_with_c6()
durations_ms: list[float] = []
for _ in range(1000):
for factory in (build_tile_store, build_tile_metadata_store, build_descriptor_index):
t0 = time.perf_counter()
factory(config)
durations_ms.append((time.perf_counter() - t0) * 1000.0)
durations_ms.sort()
p99 = durations_ms[int(0.99 * len(durations_ms))]
assert p99 <= 50.0, f"build_*() p99={p99:.3f} ms exceeds 50 ms NFR"
def test_index_metadata_dto_constructs_and_freezes(tmp_path: Path) -> None:
md = IndexMetadata(
descriptor_dim=512,
n_vectors=10_000,
backbone_label="ultra_vpr_v0",
backbone_sha256_hex="b" * 64,
built_at=datetime(2026, 1, 3, tzinfo=timezone.utc),
hnsw_params=HnswParams(),
sidecar_sha256_hex="c" * 64,
file_path=tmp_path / "tiles.index",
)
with pytest.raises(dataclasses.FrozenInstanceError):
md.n_vectors = 0 # type: ignore[misc]
tests/unit/c6_tile_cache/test_smoke.py
-18
View File
@@ -1,18 +0,0 @@
"""C6 TileCache smoke test — AC-9."""
def test_interface_importable() -> None:
# Assert
from gps_denied_onboard.components.c6_tile_cache import (
DescriptorIndex,
SectorClassification,
Tile,
TileQualityMetadata,
TileRecord,
TileStore,
)
assert TileStore is not None
assert DescriptorIndex is not None
for cls in (Tile, TileQualityMetadata, TileRecord, SectorClassification):
assert cls is not None