atlas-iac/testing/tests/test_hermes_cli_finalization_edges.py

"""Exact-run finalization guard and convergence edge coverage."""

from __future__ import annotations

from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace

from testing.tests.test_hermes_cli_support import (
    _pending_terminal_record,
    lanes,
)


class _Connection:
    def close(self):
        return None


def _db(task):
    return SimpleNamespace(
        scoped_current_board=lambda _board: nullcontext(),
        connect=lambda board: _Connection(),
        get_task=lambda *_args: task,
        reclaim_task=lambda *_args, expected_run_id=None, **_kwargs: True,
        complete_task=lambda *_args, expected_run_id=None,
        replay_ended_run_id=None, **_kwargs: False,
    )


def test_exact_recovery_rejects_missing_capability_and_nonauthoritative_tasks(
    monkeypatch,
):
    identity = lanes.TerminalIdentity("cassandra", "t_recover", 3, "pending")
    assert lanes._recover_exact_run(_db(None), None, "missing") is False

    monkeypatch.setattr(
        lanes,
        "kanban_capabilities",
        lambda _db: lanes.KanbanCapabilities(True, True, False),
    )
    assert lanes._recover_exact_run(_db(None), identity, "old image") is False

    monkeypatch.setattr(
        lanes,
        "kanban_capabilities",
        lambda _db: lanes.KanbanCapabilities(True, True, True),
    )
    assert lanes._recover_exact_run(_db(None), identity, "missing task") is False
    internal = SimpleNamespace(
        status="running",
        current_run_id=3,
        assignee="internal",
    )
    assert lanes._recover_exact_run(_db(internal), identity, "internal") is False


def test_exact_recovery_bounds_board_access_failure(monkeypatch):
    identity = lanes.TerminalIdentity("cassandra", "t_recover", 4, "pending")
    db = _db(SimpleNamespace(status="running", current_run_id=4, assignee="cli-auto"))
    db.connect = lambda **_kwargs: (_ for _ in ()).throw(OSError("volume"))
    errors = []
    monkeypatch.setattr(lanes, "KANBAN_STORAGE_ATTEMPTS", 1)
    monkeypatch.setattr(
        lanes,
        "_record_board_access_error",
        lambda board, error: errors.append((board, str(error))),
    )
    assert lanes._recover_exact_run(db, identity, "storage") is False
    assert errors[-1] == ("cassandra", "volume")


def test_finalizer_guards_invalid_runtime_identity_and_missing_task(monkeypatch):
    identity = object.__new__(lanes.TerminalIdentity)
    object.__setattr__(identity, "board", "cassandra")
    object.__setattr__(identity, "task_id", "t_invalid")
    object.__setattr__(identity, "run_id", 0)
    object.__setattr__(identity, "state", "pending")
    assert lanes._finalize_document_db(_db(None), identity, {}) == "stale"

    valid = lanes.TerminalIdentity("cassandra", "missing", 5, "pending")
    assert lanes._finalize_document_db(_db(None), valid, {}) == "stale"

    running = SimpleNamespace(
        status="running",
        current_run_id=5,
        completed_run_id=None,
        result=None,
    )
    monkeypatch.setattr(
        lanes,
        "kanban_capabilities",
        lambda _db: lanes.KanbanCapabilities(False, False, False),
    )
    assert lanes._finalize_document_db(_db(running), valid, {}) == "deferred"


class _Snapshot:
    def __init__(self, document):
        self.document = document
        self.closed = 0

    def close(self):
        self.closed += 1


def test_resolve_pending_rejects_invalid_document_and_unknown_retirement(
    tmp_path: Path,
    monkeypatch,
):
    identity = lanes.TerminalIdentity("cassandra", "t_pending", 6, "pending")
    invalid = _Snapshot({})
    monkeypatch.setattr(lanes, "_open_terminal_snapshot", lambda *_args: invalid)
    assert lanes._resolve_pending_after_winner(tmp_path / "pending", identity, {}) is False
    assert invalid.closed == 1

    valid_document = _pending_terminal_record("cassandra", "t_pending", 6, "winner")
    unknown = _Snapshot(valid_document)
    monkeypatch.setattr(lanes, "_open_terminal_snapshot", lambda *_args: unknown)
    monkeypatch.setattr(lanes, "_retire_snapshot_after_db", lambda *_args: "deferred")
    assert (
        lanes._resolve_pending_after_winner(
            tmp_path / "pending",
            identity,
            valid_document,
        )
        is False
    )
    assert unknown.closed == 1


def test_resolve_pending_stops_after_bounded_replacement_churn(
    tmp_path: Path,
    monkeypatch,
):
    identity = lanes.TerminalIdentity("cassandra", "t_churn", 7, "pending")
    document = _pending_terminal_record("cassandra", "t_churn", 7, "winner")
    snapshots = []

    def snapshot(*_args):
        value = _Snapshot(document)
        snapshots.append(value)
        return value

    monkeypatch.setattr(lanes, "_open_terminal_snapshot", snapshot)
    monkeypatch.setattr(lanes, "_retire_snapshot_after_db", lambda *_args: "replacement")
    assert lanes._resolve_pending_after_winner(tmp_path / "pending", identity, document) is False
    assert len(snapshots) == 8
    assert all(item.closed == 1 for item in snapshots)


def test_terminal_finalizer_closes_supplied_invalid_snapshots(
    tmp_path: Path,
    monkeypatch,
):
    snapshot = _Snapshot({})
    assert lanes._finalize_terminal_record(_db(None), tmp_path / "bad", snapshot=snapshot) == "invalid"
    assert snapshot.closed == 1

    monkeypatch.setattr(lanes, "STATE_ROOT", tmp_path / "lanes")
    path = lanes._terminal_path(lanes.state_path("board", "task"), 8)
    empty = _Snapshot(None)
    assert lanes._finalize_terminal_record(_db(None), path, snapshot=empty) == "invalid"
    assert empty.closed == 1