atlas-iac/testing/tests/test_hermes_handoff_rules.py

"""Contracts for the shared evaluator primitives and dispatch."""

from __future__ import annotations

import pytest

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

rules = load_handoff_module("hermes_handoff_rules")
model = load_handoff_module("hermes_handoff_model")
load_handoff_module("hermes_handoff_evaluators")


PAYLOAD = {
    "a": {"b": [{"c": 1}, {"c": 2}]},
    "items": [{"number": 14, "merged": True}, {"number": 15, "merged": False}],
    "flag": True,
}


@pytest.mark.parametrize(
    ("path", "expected"),
    [
        ("", PAYLOAD),
        ("flag", True),
        ("a.b[0].c", 1),
        ("a.b[].c", [1, 2]),
        ("a.b[]", [{"c": 1}, {"c": 2}]),
        ("items[].number", [14, 15]),
    ],
)
def test_dotted_paths_index_and_map(path: str, expected) -> None:
    assert rules.dotted(PAYLOAD, path) == expected


@pytest.mark.parametrize(
    "path", ["missing", "a.missing", "a.b[9].c", "flag[0]", "flag.b"]
)
def test_an_absent_path_raises_rather_than_returning_none(path: str) -> None:
    with pytest.raises(KeyError):
        rules.dotted(PAYLOAD, path)


def test_parse_json_rejects_raw_control_characters_and_empty_output() -> None:
    """Only RFC-compliant, nonempty JSON is accepted as release evidence."""
    with pytest.raises(ValueError):
        rules.parse_json(outcome(stdout='{"title": "a\nb"}'))
    with pytest.raises(ValueError, match="empty output"):
        rules.parse_json(outcome(stdout="   "))
    with pytest.raises(ValueError):
        rules.parse_json(outcome(stdout="not json"))


def test_lines_drops_blank_rows_and_trims() -> None:
    assert rules.lines(outcome(stdout="  a  \n\n b\n")) == ["a", "b"]


def test_missing_steps_reports_the_first_step_that_produced_nothing() -> None:
    built = spec("allowed", steps=(step("a"), step("b")))

    never_run = rules.missing_steps(built, {})
    assert never_run.status == model.NOT_RUN
    assert "never executed" in never_run.reason

    errored = rules.missing_steps(
        built, {"a": outcome(error="timeout"), "b": outcome()}
    )
    assert errored.status == model.NOT_RUN
    assert "did not run" in errored.reason

    truncated = rules.missing_steps(
        built, {"a": outcome(truncated=True), "b": outcome()}
    )
    assert truncated.status == model.NOT_RUN and "truncated" in truncated.reason

    assert rules.missing_steps(built, {"a": outcome(), "b": outcome()}) is None


def test_an_optional_step_may_be_absent() -> None:
    built = spec("allowed", steps=(step("a"), step("b", optional=True)))
    assert rules.missing_steps(built, {"a": outcome()}) is None


@pytest.mark.parametrize(
    ("stderr", "returncode", "denied"),
    [
        ("Error from server (Forbidden): pods is forbidden", 1, True),
        ("User cannot list resource", 1, True),
        # The deployed broker client's two real refusal lines.
        ("SCM broker request failed with HTTP 400", 1, True),
        ("SCM broker request failed with HTTP 403", 1, True),
        (
            "Forgejo request rejected or unavailable; no credential was disclosed",
            1,
            True,
        ),
        # Lines the deployed client never emits, or that are not refusals.
        ("Gitea API returned HTTP 403", 1, False),
        ("SCM broker request failed with HTTP 404", 1, False),
        ("connection refused", 1, False),
        ("", 0, False),
    ],
)
def test_denial_detection_needs_a_refusal_not_merely_a_failure(
    stderr, returncode, denied
) -> None:
    assert rules.looks_denied(outcome(stderr=stderr, returncode=returncode)) is denied


def test_step_keys_selects_by_kind() -> None:
    built = spec(
        "denied",
        steps=(step("r", kind=model.REVIEW), step("a", kind=model.ATTEMPT), step("x")),
    )
    assert rules.step_keys(built, model.REVIEW) == ("r",)
    assert rules.step_keys(built, model.ATTEMPT) == ("a",)
    assert rules.step_keys(built, model.READ) == ("x",)


def test_parsed_step_returns_the_payload_or_the_skip_to_report() -> None:
    built = spec("json_field", {"step": "s"}, steps=(step("s"),))

    payload, key = rules.parsed_step(built, {"s": outcome(stdout='{"a": 1}')})
    assert (payload, key) == ({"a": 1}, "s")

    failed = rules.parsed_step(built, {"s": outcome(returncode=1)})
    assert failed.status == model.NOT_RUN and "failed" in failed.reason

    unparsable = rules.parsed_step(built, {"s": outcome(stdout="nope")})
    assert (
        unparsable.status == model.NOT_RUN
        and "did not return JSON" in unparsable.reason
    )


def test_an_unknown_rule_is_not_run_rather_than_a_crash() -> None:
    evaluation = rules.evaluate(spec("no-such-rule"), {})
    assert evaluation.status == model.NOT_RUN
    assert "unknown evaluator" in evaluation.reason


def test_an_evaluator_that_raises_is_not_run_rather_than_a_pass() -> None:
    """A harness bug must never be indistinguishable from a clean release."""

    @rules.evaluator("explodes-for-test")
    def _explode(_spec, _outcomes):
        raise RuntimeError("boom")

    evaluation = rules.evaluate(spec("explodes-for-test"), {})

    assert evaluation.status == model.NOT_RUN
    assert "RuntimeError" in evaluation.reason and "boom" in evaluation.reason
    del rules.EVALUATORS["explodes-for-test"]


def test_an_evaluator_returning_an_unknown_status_is_not_run() -> None:
    @rules.evaluator("unknown-status-for-test")
    def _unknown(_spec, _outcomes):
        return rules.Evaluation("MAYBE", "bad")

    evaluation = rules.evaluate(spec("unknown-status-for-test"), {})
    assert evaluation.status == model.NOT_RUN
    assert "unknown status" in evaluation.reason
    del rules.EVALUATORS["unknown-status-for-test"]


def test_is_list_distinguishes_arrays_from_strings() -> None:
    assert rules.is_list([1]) and rules.is_list(())
    assert not rules.is_list("text")
    assert not rules.is_list(b"bytes")
    assert not rules.is_list({"a": 1})