mirror of
https://github.com/azaion/gps-denied-onboard.git
synced 2026-06-22 01:41:13 +00:00
[AZ-526] Consolidate _iso_ts_from_clock into helpers/iso_timestamps
Closes cumulative review 46-48 F1 (Medium) + F3 (Low). Adds iso_ts_from_clock(clock) alongside iso_ts_now() in the Layer-1 helper; migrates four duplicate definitions in c2_vpr (net_vlad, ultra_vpr, _faiss_bridge) and c12_operator_orchestrator (operator_reloc_service). Output format flipped +00:00 -> Z to align with iso_ts_now() and the canonical FDR _TS fixture (FDR schema test passes unmodified). 18 helper AC tests + 186 sibling tests pass; ruff clean. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -1,14 +1,15 @@
|
||||
"""AC tests for AZ-508: ISO-timestamp helper consolidation.
|
||||
"""AC tests for AZ-508 + AZ-526: ISO-timestamp helper consolidation.
|
||||
|
||||
Verifies the `iso_timestamps` helper exposed at
|
||||
`gps_denied_onboard.helpers.iso_timestamps.iso_ts_now` — the single
|
||||
Layer-1 source for FDR record envelope timestamps that replaced the
|
||||
duplicated `_iso_ts_now` one-liners in c6_tile_cache and c7_inference.
|
||||
Verifies the `iso_timestamps` helper module exposed at
|
||||
`gps_denied_onboard.helpers.iso_timestamps` — the single Layer-1 source
|
||||
for FDR record envelope timestamps. Two surfaces:
|
||||
|
||||
Output contract (matches the canonical FDR `_TS` fixture in
|
||||
`tests/unit/test_az272_fdr_record_schema.py`):
|
||||
- ``iso_ts_now()`` (AZ-508) — wall-clock variant, microsecond precision.
|
||||
- ``iso_ts_from_clock(clock)`` (AZ-526) — Clock-injected variant,
|
||||
nanosecond precision.
|
||||
|
||||
YYYY-MM-DDTHH:MM:SS.ffffffZ (UTC, microsecond precision, ``Z`` suffix)
|
||||
Both produce RFC 3339 UTC with the canonical ``Z`` suffix, matching the
|
||||
FDR ``_TS`` fixture in ``tests/unit/test_az272_fdr_record_schema.py``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -21,12 +22,18 @@ from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from gps_denied_onboard.helpers import iso_ts_now
|
||||
from gps_denied_onboard.helpers import iso_ts_from_clock, iso_ts_now
|
||||
from gps_denied_onboard.helpers.iso_timestamps import (
|
||||
iso_ts_from_clock as iso_ts_from_clock_direct,
|
||||
)
|
||||
from gps_denied_onboard.helpers.iso_timestamps import iso_ts_now as iso_ts_now_direct
|
||||
|
||||
_TS_REGEX: re.Pattern[str] = re.compile(
|
||||
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{6}Z$"
|
||||
)
|
||||
_TS_NS_REGEX: re.Pattern[str] = re.compile(
|
||||
r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{9}Z$"
|
||||
)
|
||||
|
||||
_REPO_ROOT: Path = Path(__file__).resolve().parents[2]
|
||||
_HELPER_PATH: Path = (
|
||||
@@ -38,6 +45,25 @@ _C6_DIR: Path = (
|
||||
_C7_DIR: Path = (
|
||||
_REPO_ROOT / "src" / "gps_denied_onboard" / "components" / "c7_inference"
|
||||
)
|
||||
_C2_DIR: Path = (
|
||||
_REPO_ROOT / "src" / "gps_denied_onboard" / "components" / "c2_vpr"
|
||||
)
|
||||
_C12_DIR: Path = (
|
||||
_REPO_ROOT / "src"
|
||||
/ "gps_denied_onboard"
|
||||
/ "components"
|
||||
/ "c12_operator_orchestrator"
|
||||
)
|
||||
|
||||
|
||||
class _StubClock:
|
||||
"""Minimal :class:`Clock`-shaped stub for AC-3 nanosecond verification."""
|
||||
|
||||
def __init__(self, ns: int) -> None:
|
||||
self._ns = ns
|
||||
|
||||
def time_ns(self) -> int:
|
||||
return self._ns
|
||||
|
||||
|
||||
def test_ac1_import_and_call_returns_str() -> None:
|
||||
@@ -85,9 +111,9 @@ def test_ac3_two_successive_calls_are_non_decreasing() -> None:
|
||||
|
||||
|
||||
def test_ac4_no_other_iso_ts_now_definition_exists_in_src() -> None:
|
||||
"""AC-4: a `def _iso_ts_now` / `def iso_ts_now` MUST exist only inside
|
||||
`helpers/iso_timestamps.py`. Any other definition under `src/` means a
|
||||
consumer slipped a copy back in.
|
||||
"""AC-4 (AZ-508): a `def _iso_ts_now` / `def iso_ts_now` MUST exist
|
||||
only inside `helpers/iso_timestamps.py`. Any other definition under
|
||||
`src/` means a consumer slipped a copy back in.
|
||||
"""
|
||||
# Arrange
|
||||
src_root = _REPO_ROOT / "src"
|
||||
@@ -114,6 +140,40 @@ def test_ac4_no_other_iso_ts_now_definition_exists_in_src() -> None:
|
||||
)
|
||||
|
||||
|
||||
def test_ac4_az526_no_module_level_iso_ts_from_clock_outside_helper() -> None:
|
||||
"""AC-4 (AZ-526): a module-level `def iso_ts_from_clock` /
|
||||
`def _iso_ts_from_clock` MUST exist only inside
|
||||
`helpers/iso_timestamps.py`. The two instance-method delegations
|
||||
(`_faiss_bridge.FaissBridge._iso_ts_from_clock`,
|
||||
`operator_reloc_service.OperatorReLocService._iso_ts_from_clock`)
|
||||
live inside a `ClassDef`, so we only scan top-level function defs.
|
||||
"""
|
||||
# Arrange
|
||||
src_root = _REPO_ROOT / "src"
|
||||
offenders: list[tuple[Path, str]] = []
|
||||
|
||||
# Act
|
||||
for path in src_root.rglob("*.py"):
|
||||
if path == _HELPER_PATH:
|
||||
continue
|
||||
try:
|
||||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||||
except SyntaxError:
|
||||
continue
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.FunctionDef) and node.name in {
|
||||
"iso_ts_from_clock",
|
||||
"_iso_ts_from_clock",
|
||||
}:
|
||||
offenders.append((path.relative_to(_REPO_ROOT), node.name))
|
||||
|
||||
# Assert
|
||||
assert offenders == [], (
|
||||
"Found stray module-level `iso_ts_from_clock` definitions outside "
|
||||
f"the helper: {offenders}"
|
||||
)
|
||||
|
||||
|
||||
def test_ac4_c6_and_c7_callers_import_from_helpers() -> None:
|
||||
"""The 3 migrated call-sites must import from `helpers.iso_timestamps`
|
||||
(directly or via the `helpers` package facade) so future hygiene
|
||||
@@ -142,11 +202,15 @@ def test_ac4_c6_and_c7_callers_import_from_helpers() -> None:
|
||||
)
|
||||
|
||||
|
||||
def test_ac6_helper_uses_stdlib_only() -> None:
|
||||
"""AC-6: no third-party imports inside the helper module."""
|
||||
def test_ac6_helper_uses_stdlib_or_layer1_clock_only() -> None:
|
||||
"""AC-6: the helper module's imports MUST be stdlib OR the Layer-1
|
||||
`gps_denied_onboard.clock` package (needed by `iso_ts_from_clock`
|
||||
for its `Clock` type annotation, imported under `TYPE_CHECKING`).
|
||||
"""
|
||||
# Arrange
|
||||
tree = ast.parse(_HELPER_PATH.read_text(encoding="utf-8"))
|
||||
allowed_stdlib = {"datetime", "__future__"}
|
||||
allowed_stdlib = {"datetime", "__future__", "typing"}
|
||||
allowed_layer1 = {"gps_denied_onboard.clock"}
|
||||
|
||||
# Act
|
||||
imports: list[str] = []
|
||||
@@ -160,15 +224,16 @@ def test_ac6_helper_uses_stdlib_only() -> None:
|
||||
# Assert
|
||||
for name in imports:
|
||||
top = name.split(".")[0]
|
||||
assert top in allowed_stdlib, (
|
||||
assert top in allowed_stdlib or name in allowed_layer1, (
|
||||
f"helpers/iso_timestamps.py imports `{name}`; only stdlib "
|
||||
f"({sorted(allowed_stdlib)}) is allowed by AC-6"
|
||||
f"({sorted(allowed_stdlib)}) or {sorted(allowed_layer1)} is "
|
||||
"allowed by AC-6"
|
||||
)
|
||||
|
||||
|
||||
def test_helper_is_layer_1_no_component_imports() -> None:
|
||||
"""Layer-1 discipline: the helper MUST NOT import from any component.
|
||||
(Constraint § Layer-1 discipline in the AZ-508 task spec.)
|
||||
(Constraint § Layer-1 discipline in the AZ-508 / AZ-526 task specs.)
|
||||
"""
|
||||
# Arrange
|
||||
text = _HELPER_PATH.read_text(encoding="utf-8")
|
||||
@@ -180,12 +245,145 @@ def test_helper_is_layer_1_no_component_imports() -> None:
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("expected_field", ["iso_ts_now"])
|
||||
@pytest.mark.parametrize(
|
||||
"expected_field", ["iso_ts_now", "iso_ts_from_clock"]
|
||||
)
|
||||
def test_helper_public_surface_is_minimal(expected_field: str) -> None:
|
||||
"""Defensive: only ``iso_ts_now`` is re-exported from the module."""
|
||||
"""Defensive: only the two consolidated helpers are re-exported."""
|
||||
# Arrange
|
||||
import gps_denied_onboard.helpers.iso_timestamps as module
|
||||
|
||||
# Assert
|
||||
assert expected_field in module.__all__
|
||||
assert module.__all__ == ["iso_ts_now"]
|
||||
assert module.__all__ == ["iso_ts_from_clock", "iso_ts_now"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# AZ-526 — `iso_ts_from_clock(clock)` AC tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_az526_ac1_import_and_call_returns_str() -> None:
|
||||
# Arrange
|
||||
clock = _StubClock(ns=1_700_000_000_000_000_000)
|
||||
|
||||
# Act
|
||||
value = iso_ts_from_clock(clock)
|
||||
|
||||
# Assert
|
||||
assert isinstance(value, str)
|
||||
assert value, "iso_ts_from_clock() returned an empty string"
|
||||
assert iso_ts_from_clock is iso_ts_from_clock_direct
|
||||
|
||||
|
||||
def test_az526_ac2_format_matches_canonical_ns_regex() -> None:
|
||||
# Arrange
|
||||
clock = _StubClock(ns=1_734_567_890_123_456_789)
|
||||
|
||||
# Act
|
||||
value = iso_ts_from_clock(clock)
|
||||
|
||||
# Assert
|
||||
assert _TS_NS_REGEX.fullmatch(value), (
|
||||
f"{value!r} does not match the canonical clock-driven format "
|
||||
f"YYYY-MM-DDTHH:MM:SS.fffffffffZ"
|
||||
)
|
||||
|
||||
|
||||
def test_az526_ac2_fromisoformat_truncated_roundtrip_yields_utc() -> None:
|
||||
# Arrange
|
||||
clock = _StubClock(ns=1_700_000_000_000_000_000)
|
||||
value = iso_ts_from_clock(clock)
|
||||
# stdlib fromisoformat (3.11+) accepts up to 6 fractional digits;
|
||||
# truncate the trailing 3 ns digits and the Z, then append `+00:00`.
|
||||
iso_with_offset = value[:26] + "+00:00"
|
||||
|
||||
# Act
|
||||
parsed = datetime.fromisoformat(iso_with_offset)
|
||||
|
||||
# Assert
|
||||
assert parsed.tzinfo is not None
|
||||
assert parsed.utcoffset() == timezone.utc.utcoffset(parsed)
|
||||
|
||||
|
||||
def test_az526_ac3_nanosecond_fraction_preserved_verbatim() -> None:
|
||||
"""The AC-3 fixture from the AZ-526 task spec."""
|
||||
# Arrange
|
||||
clock = _StubClock(ns=1_234_567_890_123_456_789)
|
||||
|
||||
# Act
|
||||
value = iso_ts_from_clock(clock)
|
||||
|
||||
# Assert
|
||||
assert value.endswith(".123456789Z"), (
|
||||
f"expected ns suffix `.123456789Z`, got {value!r}"
|
||||
)
|
||||
expected_prefix = (
|
||||
datetime.fromtimestamp(1_234_567_890, tz=timezone.utc)
|
||||
.strftime("%Y-%m-%dT%H:%M:%S")
|
||||
)
|
||||
assert value == f"{expected_prefix}.123456789Z"
|
||||
|
||||
|
||||
def test_az526_ac3_sub_second_distinguishability() -> None:
|
||||
"""Two clock instances 1 ns apart MUST yield distinct outputs."""
|
||||
# Arrange
|
||||
base_ns = 1_700_000_000_000_000_000
|
||||
|
||||
# Act
|
||||
a = iso_ts_from_clock(_StubClock(ns=base_ns))
|
||||
b = iso_ts_from_clock(_StubClock(ns=base_ns + 1))
|
||||
|
||||
# Assert
|
||||
assert a != b, "nanosecond precision lost (1-ns advance produced same output)"
|
||||
assert b > a, "lexicographic ordering broken at the ns boundary"
|
||||
|
||||
|
||||
def test_az526_ac4_all_four_call_sites_import_from_helper() -> None:
|
||||
"""AC-4: the four call sites import `iso_ts_from_clock` from the
|
||||
helper module. Module-level callers import as `_iso_ts_from_clock`
|
||||
(preserving their local symbol); instance-method callers import the
|
||||
canonical name and delegate.
|
||||
"""
|
||||
# Arrange
|
||||
callers = [
|
||||
_C2_DIR / "net_vlad.py",
|
||||
_C2_DIR / "ultra_vpr.py",
|
||||
_C2_DIR / "_faiss_bridge.py",
|
||||
_C12_DIR / "operator_reloc_service.py",
|
||||
]
|
||||
expected_token = (
|
||||
"from gps_denied_onboard.helpers.iso_timestamps import "
|
||||
)
|
||||
|
||||
# Act / Assert
|
||||
for caller in callers:
|
||||
text = caller.read_text(encoding="utf-8")
|
||||
assert expected_token in text and "iso_ts_from_clock" in text, (
|
||||
f"{caller.relative_to(_REPO_ROOT)} does not import "
|
||||
f"`iso_ts_from_clock` from the helper"
|
||||
)
|
||||
|
||||
|
||||
def test_az526_ac4_no_module_level_local_iso_ts_from_clock_body_in_callers() -> None:
|
||||
"""AC-4 follow-on: the four migrated callers must not contain the
|
||||
legacy multi-line body (`divmod(ns, 1_000_000_000)` + `+00:00`
|
||||
suffix). Catches a future contributor who deletes the import but
|
||||
forgets the helper exists.
|
||||
"""
|
||||
# Arrange
|
||||
callers = [
|
||||
_C2_DIR / "net_vlad.py",
|
||||
_C2_DIR / "ultra_vpr.py",
|
||||
_C2_DIR / "_faiss_bridge.py",
|
||||
_C12_DIR / "operator_reloc_service.py",
|
||||
]
|
||||
forbidden_format_marker = "+00:00\""
|
||||
|
||||
# Act / Assert
|
||||
for caller in callers:
|
||||
text = caller.read_text(encoding="utf-8")
|
||||
assert forbidden_format_marker not in text, (
|
||||
f"{caller.relative_to(_REPO_ROOT)} still emits a `+00:00`-suffix "
|
||||
"timestamp literal — the legacy ts body wasn't fully migrated"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user