atlas-iac/testing/tests/test_hermes_cli_evidence_edges.py

"""Failure-boundary coverage for immutable lane evidence and retention."""

from __future__ import annotations

import errno
import os
from pathlib import Path
from types import SimpleNamespace

import pytest

from testing.tests.test_hermes_cli_support import _pending_terminal_record, lanes


def test_rename_noreplace_rejects_missing_kernel_support(monkeypatch):
    monkeypatch.setattr(lanes.ctypes, "CDLL", lambda *_args, **_kwargs: object())
    with pytest.raises(OSError) as raised:
        lanes._rename_noreplace("a", "b", source_dir=1, destination_dir=1)
    assert raised.value.errno == errno.ENOSYS


def test_rename_noreplace_surfaces_kernel_errno(monkeypatch):
    class Rename:
        argtypes = None
        restype = None

        def __call__(self, *_args):
            return -1

    monkeypatch.setattr(
        lanes.ctypes,
        "CDLL",
        lambda *_args, **_kwargs: SimpleNamespace(renameat2=Rename()),
    )
    monkeypatch.setattr(lanes.ctypes, "get_errno", lambda: errno.EEXIST)
    with pytest.raises(FileExistsError):
        lanes._rename_noreplace("a", "b", source_dir=1, destination_dir=1)


def test_retirement_reports_missing_replacement_collision_and_restoration(
    tmp_path: Path,
    monkeypatch,
):
    path = tmp_path / "pending"
    path.write_text("old", encoding="utf-8")
    source = path.stat()
    directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    try:
        path.unlink()
        assert lanes._retire_terminal_entry(
            path,
            source,
            board_descriptor=directory,
            quarantine_descriptor=directory,
        ) == "missing"

        path.write_text("new", encoding="utf-8")
        assert lanes._retire_terminal_entry(
            path,
            source,
            board_descriptor=directory,
            quarantine_descriptor=directory,
        ) == "replacement"

        replacement = path.stat()
        monkeypatch.setattr(
            lanes,
            "_rename_noreplace",
            lambda *_args, **_kwargs: (_ for _ in ()).throw(FileExistsError()),
        )
        assert lanes._retire_terminal_entry(
            path,
            replacement,
            board_descriptor=directory,
            quarantine_descriptor=directory,
        ) == "collision"
    finally:
        os.close(directory)


def test_retirement_preserves_a_postcheck_replacement(tmp_path: Path, monkeypatch):
    path = tmp_path / "pending"
    path.write_text("old", encoding="utf-8")
    source = path.stat()
    directory = os.open(tmp_path, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
    calls = []

    def race(source_name, destination, **_kwargs):
        calls.append((source_name, destination))
        if len(calls) == 1:
            os.rename(source_name, destination, src_dir_fd=directory, dst_dir_fd=directory)
            staged = tmp_path / destination
            staged.unlink()
            staged.write_text("replacement", encoding="utf-8")
        else:
            os.rename(source_name, destination, src_dir_fd=directory, dst_dir_fd=directory)

    monkeypatch.setattr(lanes, "_rename_noreplace", race)
    try:
        assert lanes._retire_terminal_entry(
            path,
            source,
            board_descriptor=directory,
            quarantine_descriptor=directory,
        ) == "replacement"
    finally:
        os.close(directory)
    assert path.read_text(encoding="utf-8") == "replacement"


def test_evidence_writer_bounds_payload_and_cleans_failed_temporary(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 8)
    with pytest.raises(ValueError, match="bounded"):
        lanes._write_json_noreplace(tmp_path / "large.json", {"value": "x" * 5000})

    monkeypatch.setattr(lanes, "MAX_TERMINAL_RECORD_BYTES", 1024 * 1024)
    real_retire = lanes._retire_terminal_entry
    monkeypatch.setattr(
        lanes,
        "_rename_noreplace",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("rename failed")),
    )
    monkeypatch.setattr(
        lanes,
        "_retire_terminal_entry",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("cleanup failed")),
    )
    with pytest.raises(OSError, match="rename failed"):
        lanes._write_json_noreplace(tmp_path / "failed.json", {"value": "small"})
    monkeypatch.setattr(lanes, "_retire_terminal_entry", real_retire)


def test_evidence_paths_and_collision_validation_fail_closed(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    identity = lanes.TerminalIdentity("cassandra", "t_evidence", 9, "pending")
    document = _pending_terminal_record("cassandra", "t_evidence", 9, "accepted")
    with pytest.raises(ValueError, match="evidence state"):
        lanes._terminal_evidence_path(identity, "pending", document)

    monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False)
    monkeypatch.setattr(lanes, "_terminal_evidence_identity", lambda _path: None)
    with pytest.raises(OSError, match="identity is invalid"):
        lanes._persist_prepared_evidence(identity, document)
    with pytest.raises(OSError, match="conflict evidence identity"):
        lanes._persist_conflict_evidence(identity, document, "loser")


def test_existing_evidence_with_different_result_is_rejected(
    tmp_path: Path,
    monkeypatch,
):
    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "cli-lanes")
    identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "pending")
    document = _pending_terminal_record("cassandra", "t_evidence", 10, "accepted")
    prepared_identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "prepared")
    conflict_identity = lanes.TerminalIdentity("cassandra", "t_evidence", 10, "conflict")
    monkeypatch.setattr(lanes, "_write_json_noreplace", lambda *_args: False)
    monkeypatch.setattr(
        lanes,
        "_terminal_evidence_identity",
        lambda path: conflict_identity if "conflict" in path.name else prepared_identity,
    )
    monkeypatch.setattr(lanes, "_load_small_json", lambda _path: {})
    with pytest.raises(OSError, match="conflicting result"):
        lanes._persist_prepared_evidence(identity, document)
    with pytest.raises(OSError, match="different result"):
        lanes._persist_conflict_evidence(identity, document, "loser")


def test_retention_failure_paths_are_bounded(tmp_path: Path, monkeypatch, capsys):
    path = tmp_path / "artifact"
    path.write_text("data", encoding="utf-8")
    observed = path.stat()
    monkeypatch.setattr(
        lanes,
        "_retire_terminal_entry",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("busy")),
    )
    assert lanes._unlink_artifact_if_same(path, observed) is False

    monkeypatch.setattr(lanes, "LAST_ARTIFACT_GC", 0.0)
    monkeypatch.setattr(
        lanes,
        "gc_lane_artifacts",
        lambda **_kwargs: (_ for _ in ()).throw(OSError("volume")),
    )
    assert lanes.maybe_gc_lane_artifacts(now=1000) == 0
    assert "retention deferred" in capsys.readouterr().err