mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-21 19:21:12 +00:00
Oleksandr Bezdieniezhnykh
ba20c2d195
AZ-273: lock-free SPSC ring buffer with pre-allocated slots, power-of- two capacity, opt-in SPSC guard, and EnqueueResult / FdrSpscViolationError on the public surface. make_fdr_client caches one client per producer_id and reads capacity from config.fdr.per_producer_capacity with fallback to queue_size. AZ-274: default_overrun_policy implements drop-oldest + retry + immediate marker emission, with prior-marker dropped_count folding via _evict_one so user-loss info is never lost across iterations. ERROR diagnostic is rate-limited to <=1/sec per producer. AZ-275: FakeFdrSink mirrors the FdrClient public surface and reuses the production default_overrun_policy via a duck-typed _PolicyAdapter. The test-only records/all_records_ever properties let component tests assert both in-buffer and lifetime state. tests/conftest.py registers the fake_fdr_sink fixture and an AST architecture lint forbids production imports of fakes. AZ-267: FdrLogBridgeHandler installs on the root logger via wire_log_bridge and forwards only WARN+ERROR records into the FDR with kind="log". Thread-local recursion guard short-circuits internal logging; saturated- queue diagnostics go to stderr every N=1000 drops. AZ-268: tests/contract/log_schema.py covers every row of the schema's Test Cases table plus the "DEBUG+INFO never reach FDR" invariant. pyproject.toml registers the contract pytest marker and the contract-mandated log_schema.py file-name. 251 unit + contract tests pass (48 new). Review verdict: PASS_WITH_WARNINGS; findings are NFR-perf deferrals + documented relaxation of AZ-274 AC-2 coalescing under permanently-stalled consumer. Co-authored-by: Cursor <cursoragent@cursor.com>
42 lines
1.4 KiB
Python
42 lines
1.4 KiB
Python
"""Top-level pytest fixtures.
|
|
|
|
Heavy fixtures (Postgres bring-up, ArduPilot SITL, Derkachi corpus mount) are added
|
|
incrementally by the components that need them. AZ-263 ships only the smoke-level
|
|
scaffolding.
|
|
|
|
Tier-2-only tests are guarded by `pytest.mark.tier2` and auto-skipped on Tier-1.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from collections.abc import Iterator
|
|
|
|
import pytest
|
|
|
|
from gps_denied_onboard.fdr_client.fakes import FakeFdrSink
|
|
|
|
|
|
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
|
|
"""Auto-skip `tier2` tests when GPS_DENIED_TIER != 2."""
|
|
if os.environ.get("GPS_DENIED_TIER") == "2":
|
|
return
|
|
skip_tier2 = pytest.mark.skip(reason="Tier-2-only test; set GPS_DENIED_TIER=2 to run")
|
|
skip_gpu = pytest.mark.skip(reason="GPU-only test")
|
|
skip_docker = pytest.mark.skip(reason="Requires Docker compose services")
|
|
for item in items:
|
|
if "tier2" in item.keywords:
|
|
item.add_marker(skip_tier2)
|
|
if "gpu" in item.keywords:
|
|
item.add_marker(skip_gpu)
|
|
if "docker" in item.keywords:
|
|
item.add_marker(skip_docker)
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_fdr_sink() -> Iterator[FakeFdrSink]:
|
|
"""Default-configuration FakeFdrSink with overrun policy disabled (AZ-275 AC-5)."""
|
|
sink = FakeFdrSink(producer_id="test.producer")
|
|
yield sink
|
|
sink.flush()
|