mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 21:01:13 +00:00
[AZ-271] [AZ-276] [AZ-278] [AZ-282] Finish cross-cutting helpers + relax opencv pin
E-CC-HELPERS closes with the three remaining Layer-1 helpers and E-CC-CONF closes with the env > YAML > defaults precedence test gate. All four tickets ship with frozen public surfaces, hermetic unit tests, and no upward (components.*) imports. * AZ-271 — tests/unit/shared/config/test_precedence.py (5 ACs + smoke test + helper that names the layer in failure messages). * AZ-282 — helpers/ransac_filter.py: static RansacFilter + RansacResult; cv2.setRNGSeed(0) for byte-equal determinism; median residual semantics pinned by contract. * AZ-276 — helpers/imu_preintegrator.py + make_imu_preintegrator; GTSAM PreintegratedCombinedMeasurements; strict-monotonic ts_ns guard runs before any state mutation. Adjacent hygiene: _types/nav.py ImuSample/ImuWindow now use ts_ns:int and the spec-mandated ImuBias dataclass. * AZ-278 — helpers/lightglue_runtime.py: structural R14 fix. LightGlueRuntime + non-blocking concurrent-access guard that raises rather than serialising. EngineHandle Protocol in _types/manifests.py + KeypointSet/CorrespondenceSet in _types/matching.py (Protocol surface adds approved by spec). Dependency conflict (Finding 1, user-approved): gtsam 4.2 (PyPI) is numpy-1.x-ABI only; opencv-python>=4.12 needs numpy>=2 at runtime. Resolution: opencv-python pin relaxed to >=4.11.0.86,<4.12. The D-CROSS-CVE-1 ratchet at ci/opencv_pin_gate.py is held at 4.11.0 with the original 4.12.0 floor restored once a numpy-2-compatible gtsam wheel ships. Full replay procedure in _docs/_process_leftovers/2026-05-11_d_cross_cve_1_opencv_pin_deferred.md. Tests: 294 passed, 2 skipped (cmake/actionlint env-skips, pre-existing). 43 new tests added for batch 5. Ruff check + format clean. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,235 @@
|
||||
"""AZ-271 — Config precedence tests (env > YAML > defaults).
|
||||
|
||||
Verifies the precedence rule for ≥3 keys at each layer plus the
|
||||
multi-file YAML merge order (later wins) per epic AZ-246 AC-3. Tests
|
||||
are hermetic: env is passed in via the loader's ``env`` argument and
|
||||
YAML is materialised via ``tmp_path``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gps_denied_onboard.config import (
|
||||
Config,
|
||||
FdrConfig,
|
||||
LogConfig,
|
||||
RuntimeConfig,
|
||||
load_config,
|
||||
)
|
||||
|
||||
REQUIRED_ENV: dict[str, str] = {
|
||||
"GPS_DENIED_FC_PROFILE": "ardupilot_plane",
|
||||
"GPS_DENIED_TIER": "1",
|
||||
"DB_URL": "postgresql://localhost:5432/test",
|
||||
"CAMERA_CALIBRATION_PATH": "/tmp/cal.yaml",
|
||||
"LOG_LEVEL": "INFO",
|
||||
"LOG_SINK": "console",
|
||||
"INFERENCE_BACKEND": "pytorch_fp16",
|
||||
"FDR_PATH": "/tmp/fdr",
|
||||
"TILE_CACHE_PATH": "/tmp/tiles",
|
||||
}
|
||||
|
||||
|
||||
def _write_yaml(tmp_path: Path, name: str, content: str) -> Path:
|
||||
path = tmp_path / name
|
||||
path.write_text(content)
|
||||
return path
|
||||
|
||||
|
||||
def _layer_msg(layer: str, key: str, expected: object, actual: object) -> str:
|
||||
"""Standardised assertion message naming the precedence layer (AC-5)."""
|
||||
return (
|
||||
f"precedence layer {layer!r} for key {key!r}: "
|
||||
f"expected {expected!r} (from {layer}), got {actual!r}"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-1: env wins over YAML for ≥3 keys (LOG_LEVEL, FDR_QUEUE_SIZE, GPS_DENIED_TIER).
|
||||
|
||||
|
||||
def test_ac1_env_wins_over_yaml_for_three_keys(tmp_path: Path) -> None:
|
||||
# Arrange
|
||||
yaml_path = _write_yaml(
|
||||
tmp_path,
|
||||
"base.yaml",
|
||||
"""
|
||||
log:
|
||||
level: WARN
|
||||
fdr:
|
||||
queue_size: 8192
|
||||
runtime:
|
||||
tier: 2
|
||||
""",
|
||||
)
|
||||
env = dict(REQUIRED_ENV)
|
||||
env["LOG_LEVEL"] = "ERROR"
|
||||
env["FDR_QUEUE_SIZE"] = "16384"
|
||||
env["GPS_DENIED_TIER"] = "1"
|
||||
|
||||
# Act
|
||||
config = load_config(env=env, paths=(yaml_path,))
|
||||
|
||||
# Assert
|
||||
assert config.log.level == "ERROR", _layer_msg("env", "log.level", "ERROR", config.log.level)
|
||||
assert config.fdr.queue_size == 16384, _layer_msg(
|
||||
"env", "fdr.queue_size", 16384, config.fdr.queue_size
|
||||
)
|
||||
assert config.runtime.tier == 1, _layer_msg("env", "runtime.tier", 1, config.runtime.tier)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-2: YAML wins over defaults for ≥3 keys.
|
||||
|
||||
|
||||
def test_ac2_yaml_wins_over_defaults_for_three_keys(tmp_path: Path) -> None:
|
||||
# Arrange
|
||||
yaml_path = _write_yaml(
|
||||
tmp_path,
|
||||
"base.yaml",
|
||||
"""
|
||||
log:
|
||||
level: DEBUG
|
||||
sink: journald
|
||||
fdr:
|
||||
queue_size: 2048
|
||||
""",
|
||||
)
|
||||
env = dict(REQUIRED_ENV)
|
||||
# Remove any env keys that map to the YAML overrides — we want YAML > defaults
|
||||
# without env shadowing.
|
||||
for env_key in ("LOG_LEVEL", "LOG_SINK", "FDR_QUEUE_SIZE"):
|
||||
env.pop(env_key, None)
|
||||
# LOG_LEVEL is in REQUIRED_ENV but the loader's required-env gate would
|
||||
# complain; bypass with ``require_env=False`` to keep the test hermetic.
|
||||
|
||||
# Act
|
||||
config = load_config(env=env, paths=(yaml_path,), require_env=False)
|
||||
|
||||
# Assert
|
||||
log_defaults = LogConfig()
|
||||
fdr_defaults = FdrConfig()
|
||||
assert config.log.level == "DEBUG", _layer_msg("yaml", "log.level", "DEBUG", config.log.level)
|
||||
assert config.log.level != log_defaults.level
|
||||
assert config.log.sink == "journald", _layer_msg(
|
||||
"yaml", "log.sink", "journald", config.log.sink
|
||||
)
|
||||
assert config.log.sink != log_defaults.sink
|
||||
assert config.fdr.queue_size == 2048, _layer_msg(
|
||||
"yaml", "fdr.queue_size", 2048, config.fdr.queue_size
|
||||
)
|
||||
assert config.fdr.queue_size != fdr_defaults.queue_size
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-3: defaults apply for ≥3 keys when env + YAML omit them.
|
||||
|
||||
|
||||
def test_ac3_defaults_apply_when_layers_silent() -> None:
|
||||
# Arrange — empty YAML, env_default-only-required-vars.
|
||||
env = dict(REQUIRED_ENV)
|
||||
# Strip three env keys so loader falls to defaults for those three.
|
||||
env.pop("LOG_LEVEL")
|
||||
env.pop("LOG_SINK")
|
||||
env.pop("FDR_QUEUE_SIZE", None) # FDR_QUEUE_SIZE is not required
|
||||
|
||||
# Act
|
||||
config = load_config(env=env, paths=(), require_env=False)
|
||||
|
||||
# Assert
|
||||
log_defaults = LogConfig()
|
||||
fdr_defaults = FdrConfig()
|
||||
runtime_defaults = RuntimeConfig()
|
||||
assert config.log.level == log_defaults.level, _layer_msg(
|
||||
"default", "log.level", log_defaults.level, config.log.level
|
||||
)
|
||||
assert config.log.sink == log_defaults.sink, _layer_msg(
|
||||
"default", "log.sink", log_defaults.sink, config.log.sink
|
||||
)
|
||||
assert config.fdr.queue_size == fdr_defaults.queue_size, _layer_msg(
|
||||
"default", "fdr.queue_size", fdr_defaults.queue_size, config.fdr.queue_size
|
||||
)
|
||||
# Sanity: runtime defaults also intact for keys with NO env override.
|
||||
assert (
|
||||
config.runtime.inference_backend == runtime_defaults.inference_backend
|
||||
or env.get("INFERENCE_BACKEND") == config.runtime.inference_backend
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-4: multi-file YAML — later wins.
|
||||
|
||||
|
||||
def test_ac4_multi_file_yaml_later_wins(tmp_path: Path) -> None:
|
||||
# Arrange
|
||||
first = _write_yaml(
|
||||
tmp_path,
|
||||
"first.yaml",
|
||||
"""
|
||||
log:
|
||||
level: WARN
|
||||
fdr:
|
||||
queue_size: 1024
|
||||
""",
|
||||
)
|
||||
second = _write_yaml(
|
||||
tmp_path,
|
||||
"second.yaml",
|
||||
"""
|
||||
log:
|
||||
level: ERROR
|
||||
fdr:
|
||||
queue_size: 8192
|
||||
""",
|
||||
)
|
||||
env = dict(REQUIRED_ENV)
|
||||
env.pop("LOG_LEVEL") # don't let env shadow the YAML precedence test
|
||||
|
||||
# Act
|
||||
config = load_config(env=env, paths=(first, second), require_env=False)
|
||||
|
||||
# Assert
|
||||
assert config.log.level == "ERROR", _layer_msg(
|
||||
"later-yaml", "log.level", "ERROR", config.log.level
|
||||
)
|
||||
assert config.fdr.queue_size == 8192, _layer_msg(
|
||||
"later-yaml", "fdr.queue_size", 8192, config.fdr.queue_size
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AC-5: assertion message names the layer (verified by introspecting helper).
|
||||
|
||||
|
||||
def test_ac5_failure_messages_name_the_layer() -> None:
|
||||
# Arrange
|
||||
msg = _layer_msg("env", "log.level", "ERROR", "INFO")
|
||||
|
||||
# Assert — message contains the layer name, the key, and both values.
|
||||
assert "env" in msg
|
||||
assert "log.level" in msg
|
||||
assert "ERROR" in msg
|
||||
assert "INFO" in msg
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Smoke: load_config + compose_root integrate (regression guard).
|
||||
|
||||
|
||||
def test_load_config_returns_frozen_config_dataclass() -> None:
|
||||
# Arrange
|
||||
env = dict(REQUIRED_ENV)
|
||||
|
||||
# Act
|
||||
config = load_config(env=env, paths=())
|
||||
|
||||
# Assert
|
||||
import dataclasses
|
||||
|
||||
assert isinstance(config, Config)
|
||||
with pytest.raises(dataclasses.FrozenInstanceError):
|
||||
# frozen=True → cannot mutate
|
||||
config.log = LogConfig() # type: ignore[misc]
|
||||
Reference in New Issue
Block a user