tests/test_datagen_cache.py

"""Synthetic tests for fpgm.datagen.cache.StageCache.

No GPU, no network: everything here is a tmp_path directory tree and hand-built
fingerprints.
"""

from __future__ import annotations

import json
from pathlib import Path

import numpy as np

from fpgm.datagen.cache import StageCache


class TestStageDir:
    def test_stage_dir_is_created(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        stage_dir = cache.stage_dir("dense_depth")
        assert stage_dir.is_dir()
        assert stage_dir == tmp_path / "dense_depth"


class TestFreshness:
    def test_missing_meta_is_not_fresh(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        assert cache.is_fresh("s3", {"a": 1}) is False

    def test_matching_fingerprint_is_fresh(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        fp = {"config_hash": "abc123", "code_version": "s3.v1"}
        cache.write_meta("s3", fp, payload={"n_frames": 100}, limitations=["sparse"])
        assert cache.is_fresh("s3", fp) is True

    def test_changed_fingerprint_invalidates(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s3", {"config_hash": "abc123"}, payload={}, limitations=[])
        assert cache.is_fresh("s3", {"config_hash": "def456"}) is False

    def test_added_fingerprint_key_invalidates(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s3", {"config_hash": "abc123"}, payload={}, limitations=[])
        assert cache.is_fresh("s3", {"config_hash": "abc123", "extra": 1}) is False

    def test_fingerprint_with_numpy_and_path_values_round_trips(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        fp = {
            "threshold": np.float32(0.6),
            "shape": np.array([1, 2, 3]),
            "input_path": tmp_path / "input.h5",
            "nested": {"a": np.int64(7)},
        }
        cache.write_meta("s3", fp, payload={}, limitations=[])
        assert cache.is_fresh("s3", fp) is True

    def test_fingerprint_key_order_does_not_matter(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s3", {"a": 1, "b": 2}, payload={}, limitations=[])
        assert cache.is_fresh("s3", {"b": 2, "a": 1}) is True

    def test_different_stage_names_are_independent(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s2", {"x": 1}, payload={}, limitations=[])
        assert cache.is_fresh("s3", {"x": 1}) is False


class TestCorruptMeta:
    def test_corrupt_json_is_not_fresh_and_does_not_raise(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        stage_dir = cache.stage_dir("s3")
        (stage_dir / "meta.json").write_text("{not valid json")

        assert cache.is_fresh("s3", {"a": 1}) is False
        assert cache.read_meta("s3") is None

    def test_meta_json_that_is_not_an_object_is_not_fresh(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        stage_dir = cache.stage_dir("s3")
        (stage_dir / "meta.json").write_text(json.dumps([1, 2, 3]))

        assert cache.is_fresh("s3", {"a": 1}) is False
        assert cache.read_meta("s3") is None

    def test_empty_file_is_not_fresh(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        stage_dir = cache.stage_dir("s3")
        (stage_dir / "meta.json").write_text("")

        assert cache.is_fresh("s3", {"a": 1}) is False


class TestAtomicity:
    def test_partial_write_is_not_considered_fresh(self, tmp_path: Path):
        """Simulates a crash mid-write: a truncated file sitting at meta.json's
        final path must never be read as a valid, fresh cache entry."""
        cache = StageCache(tmp_path)
        stage_dir = cache.stage_dir("s3")
        full_meta = json.dumps(
            {"fingerprint": {"a": 1}, "payload": {"big": "x" * 10_000}, "limitations": []}
        )
        # A crash mid-write would leave only a byte prefix at the final path.
        truncated = full_meta[: len(full_meta) // 2]
        (stage_dir / "meta.json").write_text(truncated)

        assert cache.is_fresh("s3", {"a": 1}) is False
        assert cache.read_meta("s3") is None

    def test_no_temp_files_left_behind_after_write(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s3", {"a": 1}, payload={}, limitations=[])
        stage_dir = cache.stage_dir("s3")
        entries = list(stage_dir.iterdir())
        assert entries == [stage_dir / "meta.json"]

    def test_rewrite_replaces_previous_meta_atomically(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta("s3", {"a": 1}, payload={"v": 1}, limitations=[])
        cache.write_meta("s3", {"a": 2}, payload={"v": 2}, limitations=["updated"])

        meta = cache.read_meta("s3")
        assert meta["fingerprint"] == {"a": 2}
        assert meta["payload"] == {"v": 2}
        assert meta["limitations"] == ["updated"]


class TestPayloadAndLimitations:
    def test_payload_and_limitations_round_trip(self, tmp_path: Path):
        cache = StageCache(tmp_path)
        cache.write_meta(
            "s7",
            fingerprint={"code_version": "s7.v1"},
            payload={"n_flagged": 3, "pass_rate": 0.65},
            limitations=["Event labels are heuristic flags, never training targets."],
        )
        meta = cache.read_meta("s7")
        assert meta["payload"] == {"n_flagged": 3, "pass_rate": 0.65}
        assert meta["limitations"] == [
            "Event labels are heuristic flags, never training targets."
        ]