gps-denied-onboard/tests/unit/test_az508_iso_timestamps.py

"""AC tests for AZ-508 + AZ-526: ISO-timestamp helper consolidation.

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:

- ``iso_ts_now()`` (AZ-508) — wall-clock variant, microsecond precision.
- ``iso_ts_from_clock(clock)`` (AZ-526) — Clock-injected variant,
  nanosecond precision.

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

import ast
import re
import time
from datetime import datetime, timezone
from pathlib import Path

import pytest

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 = (
    _REPO_ROOT / "src" / "gps_denied_onboard" / "helpers" / "iso_timestamps.py"
)
_C6_DIR: Path = (
    _REPO_ROOT / "src" / "gps_denied_onboard" / "components" / "c6_tile_cache"
)
_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:
    # Act
    value = iso_ts_now()
    # Assert
    assert isinstance(value, str)
    assert value, "iso_ts_now() returned an empty string"
    # Both the package-level and module-level imports must resolve to the
    # same callable so consumers can reach it either way.
    assert iso_ts_now is iso_ts_now_direct


def test_ac2_format_matches_canonical_regex() -> None:
    # Act
    value = iso_ts_now()
    # Assert
    assert _TS_REGEX.fullmatch(value), (
        f"{value!r} does not match the canonical FDR ts format "
        f"YYYY-MM-DDTHH:MM:SS.ffffffZ"
    )


def test_ac2_fromisoformat_roundtrip_yields_utc_aware_datetime() -> None:
    # Arrange
    value = iso_ts_now()
    iso_with_offset = value.replace("Z", "+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_ac3_two_successive_calls_are_non_decreasing() -> None:
    # Arrange / Act
    a = iso_ts_now()
    time.sleep(0.000_005)
    b = iso_ts_now()

    # Assert (lexicographic comparison is correct for the fixed-width format)
    assert b >= a, f"expected {b!r} >= {a!r}"


def test_ac4_no_other_iso_ts_now_definition_exists_in_src() -> None:
    """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"
    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 ast.walk(tree):
            if isinstance(node, ast.FunctionDef) and node.name in {
                "iso_ts_now",
                "_iso_ts_now",
            }:
                offenders.append((path.relative_to(_REPO_ROOT), node.name))

    # Assert
    assert offenders == [], (
        f"Found stray `iso_ts_now` definitions outside the helper: {offenders}"
    )


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
    cycles can rely on the single source of truth.
    """
    # Arrange
    callers = [
        _C6_DIR / "cache_budget_enforcer.py",
        _C6_DIR / "postgres_filesystem_store.py",
        _C6_DIR / "freshness_gate.py",
        _C7_DIR / "onnx_trt_ep_runtime.py",
        _C7_DIR / "thermal_publisher.py",
    ]
    expected_token = "gps_denied_onboard.helpers.iso_timestamps"

    # Act / Assert
    for caller in callers:
        text = caller.read_text(encoding="utf-8")
        assert expected_token in text, (
            f"{caller.relative_to(_REPO_ROOT)} does not import "
            f"`iso_ts_now` from `{expected_token}`"
        )
        assert "def _iso_ts_now" not in text, (
            f"{caller.relative_to(_REPO_ROOT)} still defines a local "
            "_iso_ts_now (consolidation incomplete)"
        )


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__", "typing"}
    allowed_layer1 = {"gps_denied_onboard.clock"}

    # Act
    imports: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            imports.extend(alias.name for alias in node.names)
        elif isinstance(node, ast.ImportFrom):
            if node.module is not None:
                imports.append(node.module)

    # Assert
    for name in imports:
        top = name.split(".")[0]
        assert top in allowed_stdlib or name in allowed_layer1, (
            f"helpers/iso_timestamps.py imports `{name}`; only stdlib "
            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 / AZ-526 task specs.)
    """
    # Arrange
    text = _HELPER_PATH.read_text(encoding="utf-8")

    # Assert
    assert "gps_denied_onboard.components" not in text, (
        "helpers/iso_timestamps.py imports from a component — Layer-1 "
        "discipline violated"
    )


@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 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_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"
        )