DataOps-platform/tests/test_p3_wp14_fresh_verifier.py

from __future__ import annotations

import subprocess
import sys
import hashlib
from datetime import datetime, timezone
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))


def test_fresh_verifier_runs_generator_before_reading_unsigned_cache(monkeypatch) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    calls: list[str] = []
    monkeypatch.setattr(verifier, "_run_generator", lambda: calls.append("generator") or "fresh-nonce")
    monkeypatch.setattr(verifier, "_verify_cache", lambda _now, **_kwargs: calls.append("verify"))

    assert verifier.main() == 0
    assert calls == ["generator", "verify"]


def test_tampered_cache_is_rebuilt_before_fresh_verification(monkeypatch) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    cache = {"digest": "tampered"}

    def rebuild() -> str:
        cache["digest"] = "fresh"
        return "fresh-nonce"

    def verify(_now, **_kwargs) -> None:
        assert cache["digest"] == "fresh"

    monkeypatch.setattr(verifier, "_run_generator", rebuild)
    monkeypatch.setattr(verifier, "_verify_cache", verify)
    assert verifier.main() == 0


def test_fresh_verifier_rejects_expired_cache_without_a_new_generator_run() -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    with pytest.raises(ValueError, match="expired"):
        verifier._assert_current_expiry(
            {"expires_at": "2000-01-01T00:00:00Z"}, datetime.now(timezone.utc)
        )


def test_fresh_verifier_timeout_returns_failure(monkeypatch) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    def timeout() -> None:
        raise subprocess.TimeoutExpired(["python", "generator"], verifier.GENERATOR_TIMEOUT_SECONDS)

    monkeypatch.setattr(verifier, "_run_generator", timeout)
    assert verifier.main() == 1


def test_verify_existing_is_read_only_and_never_runs_generator(monkeypatch, capsys) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    calls: list[str] = []
    monkeypatch.setattr(verifier, "_run_generator", lambda: calls.append("generator"))
    monkeypatch.setattr(verifier, "_verify_cache", lambda _now, **_kwargs: calls.append("verify"))
    assert verifier.main(["--verify-existing"]) == 0
    assert calls == ["verify"]
    assert "VERIFY_EXISTING_DIAGNOSTIC_ONLY" in capsys.readouterr().out


def test_verify_existing_preserves_existing_cache_mtime_and_digest(monkeypatch, tmp_path: Path) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    cache = tmp_path / "unsigned-cache.json"
    cache.write_bytes(b'{"cache":"existing"}\n')
    before = (cache.stat().st_mtime_ns, hashlib.sha256(cache.read_bytes()).hexdigest())
    monkeypatch.setattr(verifier, "_run_generator", lambda: pytest.fail("must remain read-only"))
    monkeypatch.setattr(verifier, "_verify_cache", lambda _now, **_kwargs: cache.read_bytes())
    assert verifier.main(["--verify-existing"]) == 0
    after = (cache.stat().st_mtime_ns, hashlib.sha256(cache.read_bytes()).hexdigest())
    assert after == before


def test_fresh_generator_environment_drops_parent_sensitive_overrides(monkeypatch) -> None:
    import verify_p3_wp14_fresh_local_engineering as verifier

    monkeypatch.setenv("HOME", "/attacker")
    monkeypatch.setenv("DOCKER_CONFIG", "/attacker/docker")
    monkeypatch.setenv("GIT_DIR", "/attacker/git")
    env = verifier._minimal_command_env()
    assert env["HOME"] == str(verifier.COMMAND_HOME)
    assert env["DOCKER_CONFIG"] == str(verifier.COMMAND_HOME / "docker")
    assert "GIT_DIR" not in env
    assert "TMPDIR" in env and {key for key in env if key.startswith("PYTHON")} == {"PYTHONDONTWRITEBYTECODE"}
    assert "DOCKER_HOST" not in env and "DOCKER_CONTEXT" not in env


def test_ledger_command_environment_drops_parent_sensitive_overrides(monkeypatch) -> None:
    import generate_p3_wp14_verification_ledger as ledger

    monkeypatch.setenv("HOME", "/attacker")
    monkeypatch.setenv("TMPDIR", "/attacker/tmp")
    monkeypatch.setenv("PYTHONPATH", "/attacker/python")
    monkeypatch.setenv("GIT_DIR", "/attacker/git")
    monkeypatch.setenv("COMPOSE_FILE", "/attacker/compose.yml")
    monkeypatch.setenv("DOCKER_CONFIG", "/attacker/docker")
    monkeypatch.setenv("DOCKER_HOST", "unix:///attacker.sock")
    env = ledger._minimal_command_env()
    assert env["HOME"] == str(ledger.COMMAND_HOME)
    assert env["TMPDIR"] == str(ledger.COMMAND_HOME / "tmp")
    assert env["DOCKER_CONFIG"] == str(ledger.COMMAND_HOME / "docker")
    assert "GIT_DIR" not in env and "COMPOSE_FILE" not in env and "DOCKER_HOST" not in env