atlas-iac/testing/tests/test_hermes_handoff_model.py

"""Contracts for the handoff harness result model, verdict, and rendering."""

from __future__ import annotations

import datetime as dt

import pytest

from testing.tests.test_hermes_handoff_support import (
    load_handoff_module,
    outcome,
    spec,
    step,
)

model = load_handoff_module("hermes_handoff_model")


def result(
    status: str,
    identifier: str = "test.check",
    mandatory: bool = True,
    group: str = "g",
):
    return model.CheckResult(
        spec=model.CheckSpec(
            id=identifier, title="t", group=group, rule="allowed", mandatory=mandatory
        ),
        status=status,
        reason=f"{status} reason",
    )


def report(*results, errors: tuple[str, ...] = ()) -> model.Report:
    return model.Report(
        mode="read-only",
        started_at="2026-08-17T00:00:00Z",
        results=list(results),
        harness_errors=list(errors),
    )


@pytest.mark.parametrize(
    ("status", "mandatory", "blocking"),
    [
        (model.PASS, True, False),
        (model.NOT_APPLICABLE, True, False),
        (model.FAIL, True, True),
        (model.NOT_RUN, True, True),
        (model.FAIL, False, False),
        (model.NOT_RUN, False, False),
    ],
)
def test_only_a_mandatory_failure_or_skip_blocks(status, mandatory, blocking) -> None:
    assert result(status, mandatory=mandatory).blocking is blocking


def test_a_mandatory_skip_is_a_no_go_exactly_like_a_failure() -> None:
    """The whole point of the four-valued classification."""
    assert report(result(model.NOT_RUN)).decision == model.NO_GO
    assert report(result(model.FAIL)).decision == model.NO_GO
    assert report(result(model.PASS), result(model.NOT_APPLICABLE)).decision == model.GO


def test_a_harness_error_is_a_no_go_even_when_every_check_passed() -> None:
    clean = report(
        result(model.PASS), errors=("the two vantages are the same principal",)
    )
    assert clean.decision == model.NO_GO


def test_a_zero_state_or_unknown_status_report_is_no_go() -> None:
    """Absence or an unrecognised classification can never round up to GO."""
    assert report().decision == model.NO_GO
    assert report(errors=("no checks were built",)).decision == model.NO_GO
    unknown = result("MAYBE")
    assert unknown.blocking is True
    built = report(unknown)
    assert built.decision == model.NO_GO
    assert built.counts == dict.fromkeys(model.STATUSES, 0)
    assert "groups:" in report(result(model.PASS)).render_summary()


def test_blocking_results_are_ordered_failures_before_skips() -> None:
    ordered = report(
        result(model.NOT_RUN, "b.skip"),
        result(model.FAIL, "z.fail"),
        result(model.PASS, "a.pass"),
    ).blocking

    assert [item.spec.id for item in ordered] == ["z.fail", "b.skip"]


def test_counts_and_group_summary_tally_every_status() -> None:
    built = report(
        result(model.PASS, "a", group="alpha"),
        result(model.FAIL, "b", group="alpha"),
        result(model.NOT_RUN, "c", group="beta"),
    )

    assert built.counts == {
        model.PASS: 1,
        model.FAIL: 1,
        model.NOT_RUN: 1,
        model.NOT_APPLICABLE: 0,
    }
    assert built.group_summary()["alpha"][model.PASS] == 1
    assert built.group_summary()["beta"][model.NOT_RUN] == 1


def test_the_serialised_report_carries_the_verdict_and_its_reasons() -> None:
    built = report(result(model.FAIL, "x.fail"), result(model.PASS, "y.pass"))
    built.vantages = [
        model.VantageRecord(
            name="operator", description="d", identity="u", available=True
        )
    ]

    payload = built.as_dict()

    assert payload["decision"] == model.NO_GO
    assert payload["blocking"] == ["x.fail"]
    assert payload["harness"] == model.HARNESS
    assert payload["schema_version"] == model.SCHEMA_VERSION
    assert payload["vantages"][0]["identity"] == "u"
    assert {check["id"] for check in payload["checks"]} == {"x.fail", "y.pass"}


def test_check_serialisation_screens_evidence_and_can_omit_commands() -> None:
    built = model.CheckResult(
        spec=spec("allowed", identifier="x", steps=(step("a"),)),
        status=model.PASS,
        evidence={"token": "a-long-enough-secret", "count": 3},
        outcomes=[outcome(stdout="fine")],
    )

    with_commands = built.as_dict()
    without = built.as_dict(include_outcomes=False)

    assert with_commands["evidence"]["token"] == "[redacted]"
    assert with_commands["evidence"]["count"] == 3
    assert len(with_commands["commands"]) == 1
    assert "commands" not in without
    assert with_commands["vantages"] == ["self"]


def test_rationale_is_carried_only_when_a_check_has_one() -> None:
    without = model.CheckResult(spec=spec("allowed"), status=model.PASS).as_dict()
    assert "rationale" not in without

    annotated = model.CheckSpec(
        id="x", title="t", group="g", rule="allowed", rationale="because"
    )
    with_reason = model.CheckResult(spec=annotated, status=model.PASS).as_dict()
    assert with_reason["rationale"] == "because"


def test_distinct_vantages_are_reported_in_first_use_order() -> None:
    built = spec(
        "allowed",
        steps=(
            step("a", vantage="operator"),
            step("b", vantage="self"),
            step("c", vantage="operator"),
        ),
    )
    assert built.vantages == ("operator", "self")


def test_the_human_summary_names_the_verdict_vantages_and_blockers() -> None:
    built = report(result(model.FAIL, "x.fail", group="alpha"), errors=("collision",))
    built.vantages = [
        model.VantageRecord(
            name="operator", description="d", identity="u", available=True
        ),
        model.VantageRecord(name="self", description="d", available=False),
    ]

    summary = built.render_summary()

    assert summary.splitlines()[0] == f"{model.HARNESS}: {model.NO_GO}"
    assert "vantage operator: available identity=u" in summary
    assert "vantage self: unavailable identity=unknown" in summary
    assert "harness error: collision" in summary
    assert "x.fail" in summary
    assert "alpha" in summary


def test_utc_now_renders_a_zulu_timestamp_and_accepts_an_injected_clock() -> None:
    fixed = dt.datetime(2026, 8, 17, 9, 30, 15, 123456, tzinfo=dt.timezone.utc)
    assert model.utc_now(lambda: fixed) == "2026-08-17T09:30:15Z"
    assert model.utc_now().endswith("Z")


def test_unscreened_fields_walks_the_whole_payload() -> None:
    offenders = model.unscreened_fields(
        {
            "clean": "hermes",
            "nested": {"leak": "token: ghp_ABCDEFGHIJKLMNOPQRSTUV1234"},
            "items": [1, "Authorization: Bearer abcdefghijklmnop"],
        }
    )

    assert offenders == ["$.nested.leak", "$.items[1]"]
    assert model.unscreened_fields({"clean": "hermes"}) == []
    assert model.unscreened_fields(None) == []