Files
gps-denied-onboard/tests/unit/test_az280_sha256_sidecar.py
Oleksandr Bezdieniezhnykh 8e71f6c002 [AZ-266] [AZ-269] [AZ-277] [AZ-280] Cross-cutting log/config + SE3/SHA256 helpers
AZ-266: schema-compliant JSON logging entrypoint, level normalisation,
handler-topology guard, format-error fallback (log_record_schema v1.0.0).
AZ-269: env > YAML > defaults config loader, frozen Config dataclass,
missing-var fail-fast with pointer to .env.example, component-block registry.
AZ-277: GTSAM-backed SE3Utils (matrix<->SE3 + exp/log/adjoint) with strict
orthogonality, dtype, and bottom-row contract enforcement.
AZ-280: atomicwrites-backed write_atomic + independent verify +
order-deterministic aggregate_hash; sidecar format strictness.
pyproject.toml pins gtsam>=4.2,<5.0 and atomicwrites>=1.4,<2.0
(named-backend deps per the AZ-277 / AZ-280 contracts).
139 unit tests pass (44 new). Review verdict: PASS_WITH_WARNINGS;
findings are perf-NFR + journald deferrals, no blocking issues.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-11 01:33:42 +03:00

215 lines
6.8 KiB
Python

"""AC tests for AZ-280: Sha256Sidecar helper.
Verifies the `sha256_sidecar` contract v1.0.0 — atomic write, independent
verification, sidecar format strictness, aggregate determinism, and the
no-upward-imports invariant.
"""
from __future__ import annotations
import ast
import hashlib
from pathlib import Path
import pytest
from gps_denied_onboard.helpers import (
SIDECAR_SUFFIX,
Sha256Sidecar,
Sha256SidecarError,
)
def _payload(seed: int, size: int = 1 << 20) -> bytes:
"""1 MiB random-ish payload derived from a seed for reproducibility."""
import random
rng = random.Random(seed)
return bytes(rng.randint(0, 255) for _ in range(size))
def _sidecar(path: Path) -> Path:
return Path(str(path) + SIDECAR_SUFFIX)
def test_ac1_roundtrip_write_and_verify(tmp_path: Path) -> None:
# Arrange
payload = _payload(seed=1)
target = tmp_path / "artifact.bin"
# Act
written_digest = Sha256Sidecar.write_atomic_and_sidecar(target, payload)
# Assert
assert target.exists()
assert _sidecar(target).exists()
sidecar_text = _sidecar(target).read_text()
expected_digest = hashlib.sha256(payload).hexdigest()
assert sidecar_text == expected_digest
assert written_digest == expected_digest
assert Sha256Sidecar.verify(target) is True
def test_ac2_atomicity_no_partial_file_on_fault(tmp_path: Path, monkeypatch) -> None:
# Arrange
payload = b"hello"
target = tmp_path / "atomic.bin"
# Sabotage `os.replace` via the atomicwrites module so the rename step fails.
import atomicwrites
real_replace = atomicwrites.replace_atomic
def _failing_replace(src: str, dst: str) -> None:
raise OSError("simulated rename failure")
monkeypatch.setattr(atomicwrites, "replace_atomic", _failing_replace)
# Act + Assert — the write should raise; no half file should appear at the target.
with pytest.raises(Sha256SidecarError):
Sha256Sidecar.write_atomic(target, payload)
assert not target.exists(), "no partial file may exist at the target name on fault"
# Also confirm no orphaned `.tmp` siblings remain at the target name.
siblings = [p for p in tmp_path.iterdir() if p.name.startswith("atomic.bin")]
assert all(p == target for p in siblings if p.exists())
# Cleanup
monkeypatch.setattr(atomicwrites, "replace_atomic", real_replace)
def test_ac3_independent_verification_rejects_swapped_payload(tmp_path: Path) -> None:
# Arrange
payload = b"original payload"
target = tmp_path / "swap.bin"
Sha256Sidecar.write_atomic_and_sidecar(target, payload)
# Act — replace the artifact bytes out-of-band without updating the sidecar.
target.write_bytes(b"completely different bytes that are longer")
# Assert
assert Sha256Sidecar.verify(target) is False
def test_ac4_missing_sidecar_raises(tmp_path: Path) -> None:
# Arrange
payload = b"data"
target = tmp_path / "missing_sidecar.bin"
Sha256Sidecar.write_atomic_and_sidecar(target, payload)
_sidecar(target).unlink()
# Act + Assert
with pytest.raises(Sha256SidecarError) as excinfo:
Sha256Sidecar.verify(target)
assert "sidecar" in str(excinfo.value).lower()
def test_ac5_malformed_sidecar_rejected(tmp_path: Path) -> None:
# Arrange
payload = b"data"
target = tmp_path / "bad_sidecar.bin"
Sha256Sidecar.write_atomic_and_sidecar(target, payload)
_sidecar(target).write_text("not a hex digest at all")
# Act + Assert
with pytest.raises(Sha256SidecarError):
Sha256Sidecar.verify(target)
def test_ac6_aggregate_hash_order_deterministic(tmp_path: Path) -> None:
# Arrange
paths = []
for i in range(3):
p = tmp_path / f"file_{i}.bin"
Sha256Sidecar.write_atomic(p, f"payload-{i}".encode())
paths.append(p)
# Act
a = Sha256Sidecar.aggregate_hash(paths)
b = Sha256Sidecar.aggregate_hash(list(reversed(paths)))
c = Sha256Sidecar.aggregate_hash([paths[1], paths[0], paths[2]])
# Assert — same set, three permutations -> identical aggregate
assert a == b == c
def test_ac7_aggregate_hash_missing_path_raises(tmp_path: Path) -> None:
# Arrange
real = tmp_path / "real.bin"
Sha256Sidecar.write_atomic(real, b"data")
ghost = tmp_path / "ghost.bin" # never created
# Act + Assert
with pytest.raises(Sha256SidecarError) as excinfo:
Sha256Sidecar.aggregate_hash([real, ghost])
assert "ghost.bin" in str(excinfo.value)
def test_ac8_sidecar_format_strictness(tmp_path: Path) -> None:
# Arrange
payload = b"hello world"
target = tmp_path / "strict.bin"
Sha256Sidecar.write_atomic_and_sidecar(target, payload)
# Act
raw_bytes = _sidecar(target).read_bytes()
text = _sidecar(target).read_text()
# Assert — exactly 64 hex chars, no newline, no JSON wrapper, no whitespace.
assert len(raw_bytes) == 64, f"sidecar must be 64 bytes; got {len(raw_bytes)}"
assert not raw_bytes.endswith(b"\n"), "sidecar must NOT have a trailing newline"
assert text == text.strip(), "sidecar must contain no whitespace"
int(text, 16) # raises if not pure hex
assert text == hashlib.sha256(payload).hexdigest()
def test_ac9_no_upward_imports_to_components() -> None:
# Arrange
module_path = (
Path(__file__).resolve().parents[2]
/ "src"
/ "gps_denied_onboard"
/ "helpers"
/ "sha256_sidecar.py"
)
tree = ast.parse(module_path.read_text())
bad_imports: list[str] = []
# Act
for node in ast.walk(tree):
if isinstance(node, ast.ImportFrom):
module = node.module or ""
if module.startswith("gps_denied_onboard.components"):
bad_imports.append(module)
elif isinstance(node, ast.Import):
for alias in node.names:
if alias.name.startswith("gps_denied_onboard.components"):
bad_imports.append(alias.name)
# Assert
assert not bad_imports, (
f"helpers.sha256_sidecar must not import from components.*; found {bad_imports}"
)
def test_verify_on_missing_path_returns_false(tmp_path: Path) -> None:
"""Contract: `verify` distinguishes 'missing path' (False) from 'missing sidecar' (raise)."""
# Act + Assert
assert Sha256Sidecar.verify(tmp_path / "never_created.bin") is False
def test_aggregate_includes_all_file_digests(tmp_path: Path) -> None:
"""The aggregate must be sensitive to file content, not only path names."""
# Arrange
a = tmp_path / "a.bin"
b = tmp_path / "b.bin"
Sha256Sidecar.write_atomic(a, b"alpha")
Sha256Sidecar.write_atomic(b, b"beta")
digest_initial = Sha256Sidecar.aggregate_hash([a, b])
# Act — mutate b's content and recompute
Sha256Sidecar.write_atomic(b, b"gamma")
digest_after = Sha256Sidecar.aggregate_hash([a, b])
# Assert
assert digest_initial != digest_after