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

"""AC tests for AZ-508: 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.

Output contract (matches the canonical FDR `_TS` fixture in
`tests/unit/test_az272_fdr_record_schema.py`):

    YYYY-MM-DDTHH:MM:SS.ffffffZ  (UTC, microsecond precision, ``Z`` suffix)
"""

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_now
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$"
)

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


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: 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_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_only() -> None:
    """AC-6: no third-party imports inside the helper module."""
    # Arrange
    tree = ast.parse(_HELPER_PATH.read_text(encoding="utf-8"))
    allowed_stdlib = {"datetime", "__future__"}

    # 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, (
            f"helpers/iso_timestamps.py imports `{name}`; only stdlib "
            f"({sorted(allowed_stdlib)}) 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.)
    """
    # 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"])
def test_helper_public_surface_is_minimal(expected_field: str) -> None:
    """Defensive: only ``iso_ts_now`` is re-exported from the module."""
    # Arrange
    import gps_denied_onboard.helpers.iso_timestamps as module

    # Assert
    assert expected_field in module.__all__
    assert module.__all__ == ["iso_ts_now"]