orchestrator/agent-team/tests/test_ci_gate.py

"""Unit tests for agent_team.ci_gate — the pure-code pass/fail gate (§3.3.2)."""

from __future__ import annotations

import pytest

from agent_team.ci_gate import (
    DENYLIST_GLOBS,
    CiGateError,
    GateDecision,
    denylist_violations,
    diff_touched_paths,
    evaluate_ci_gate,
    verify_diff_hash,
)
from agent_team.state_store import compute_content_hash


def _diff_for(*paths: str) -> str:
    """Build a minimal unified diff touching ``paths`` (no rename)."""
    chunks = []
    for p in paths:
        chunks.append(
            f"diff --git a/{p} b/{p}\n--- a/{p}\n+++ b/{p}\n@@ -1 +1 @@\n-old\n+new\n"
        )
    return "".join(chunks)


def _ledger_hash(diff: str) -> str:
    return compute_content_hash(diff.encode("utf-8"))


def _good_ci(run_id: str, diff: str, conclusion: str = "success") -> dict:
    return {
        "run_id": run_id,
        "conclusion": conclusion,
        "diff_hash": _ledger_hash(diff),
    }


# --------------------------------------------------------------------------- #
# diff_touched_paths
# --------------------------------------------------------------------------- #


def test_touched_paths_basic() -> None:
    diff = _diff_for("src/foo.py", "tests/test_foo.py")
    assert diff_touched_paths(diff) == ["src/foo.py", "tests/test_foo.py"]


def test_touched_paths_strips_git_prefix_and_dedups() -> None:
    diff = "diff --git a/pkg/mod.py b/pkg/mod.py\n@@ @@\n+x\n"
    assert diff_touched_paths(diff) == ["pkg/mod.py"]


def test_touched_paths_includes_rename_lines() -> None:
    diff = (
        "diff --git a/safe.txt b/.github/workflows/evil.yml\n"
        "similarity index 100%\n"
        "rename from safe.txt\n"
        "rename to .github/workflows/evil.yml\n"
    )
    paths = diff_touched_paths(diff)
    assert ".github/workflows/evil.yml" in paths
    assert "safe.txt" in paths


def test_touched_paths_non_string_raises() -> None:
    with pytest.raises(CiGateError):
        diff_touched_paths(None)  # type: ignore[arg-type]


# --------------------------------------------------------------------------- #
# denylist_violations
# --------------------------------------------------------------------------- #


def test_clean_diff_has_no_violations() -> None:
    diff = _diff_for("src/foo.py", "README.md")
    assert denylist_violations(diff) == []


@pytest.mark.parametrize(
    "path",
    [
        ".github/workflows/ci.yml",
        ".github/actions/deploy/action.yml",
        ".github/CODEOWNERS",
        "CODEOWNERS",
        ".github/dependabot.yml",
        "infra/cdk.json",
        "service/template.yaml",
        "deploy/iam/role.json",
        "stacks/policies/admin.json",
        "modules/main.tf",
    ],
)
def test_denylisted_paths_flagged(path: str) -> None:
    violations = denylist_violations(_diff_for(path))
    assert violations, f"expected {path!r} to be denylisted"


def test_rename_into_workflow_is_flagged() -> None:
    diff = (
        "diff --git a/safe.txt b/.github/workflows/evil.yml\n"
        "rename from safe.txt\n"
        "rename to .github/workflows/evil.yml\n"
    )
    violations = denylist_violations(diff)
    assert any(".github/workflows/evil.yml" in v for v in violations)


def test_path_escape_via_dotdot_flagged() -> None:
    diff = "diff --git a/x b/../../etc/passwd\n@@ @@\n+x\n"
    violations = denylist_violations(diff)
    assert any("escapes repo root" in v for v in violations)


def test_allowed_scope_blocks_out_of_scope_path() -> None:
    diff = _diff_for("src/in_scope.py", "other/out_of_scope.py")
    violations = denylist_violations(diff, allowed_scope=["src/"])
    assert any("declared scope" in v for v in violations)
    # In-scope path alone is clean.
    assert (
        denylist_violations(_diff_for("src/in_scope.py"), allowed_scope=["src/"]) == []
    )


def test_allowed_scope_exact_file_prefix() -> None:
    diff = _diff_for("pkg/exact.py")
    assert denylist_violations(diff, allowed_scope=["pkg/exact.py"]) == []


def test_denylist_globs_exported_nonempty() -> None:
    assert isinstance(DENYLIST_GLOBS, tuple)
    assert ".github/workflows/**" in DENYLIST_GLOBS


# --------------------------------------------------------------------------- #
# verify_diff_hash
# --------------------------------------------------------------------------- #


def test_hash_matches_ledger() -> None:
    diff = _diff_for("a.py")
    assert verify_diff_hash(diff, ledger_hash=_ledger_hash(diff)) is True


def test_hash_mismatch_ledger() -> None:
    diff = _diff_for("a.py")
    assert verify_diff_hash(diff, ledger_hash="deadbeef") is False


def test_hash_none_ledger_fails() -> None:
    diff = _diff_for("a.py")
    assert verify_diff_hash(diff, ledger_hash=None) is False


def test_hash_ci_verified_must_also_match() -> None:
    diff = _diff_for("a.py")
    h = _ledger_hash(diff)
    assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash=h) is True
    assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash="nope") is False


def test_hash_non_string_raises() -> None:
    with pytest.raises(CiGateError):
        verify_diff_hash(123, ledger_hash="x")  # type: ignore[arg-type]


# --------------------------------------------------------------------------- #
# evaluate_ci_gate — the block decision
# --------------------------------------------------------------------------- #


def test_gate_pass_on_authenticated_success() -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-1", diff, "success"),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.PASS
    assert result.passed is True
    assert result.ci_conclusion == "success"


def test_gate_fail_on_authenticated_failure() -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-1", diff, "failure"),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.FAIL
    assert result.ci_conclusion == "failure"


@pytest.mark.parametrize(
    "conclusion", ["timed_out", "cancelled", "startup_failure", "action_required"]
)
def test_gate_recognised_failures(conclusion: str) -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-1", diff, conclusion),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.FAIL


def test_gate_blocks_denylisted_diff_even_if_ci_success() -> None:
    # A diff that touches the trust-control surface BLOCKs regardless of CI.
    diff = _diff_for(".github/workflows/ci.yml")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-1", diff, "success"),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK
    assert result.blocked is True
    assert any("denylisted" in r for r in result.reasons)


def test_gate_blocks_on_hash_mismatch() -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash="deadbeef",  # does not match recomputed hash
        ci_result=_good_ci("run-1", diff, "success"),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK
    assert any("hash" in r for r in result.reasons)


def test_gate_blocks_on_run_id_mismatch() -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-OTHER", diff, "success"),
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK
    assert any("run-id" in r for r in result.reasons)


def test_gate_blocks_on_missing_ci_result() -> None:
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=None,
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK


@pytest.mark.parametrize("conclusion", [None, "neutral", "skipped", "in_progress"])
def test_gate_blocks_on_ambiguous_conclusion(conclusion) -> None:
    # Anything not an explicit success or recognised failure must NOT silently
    # pass — it BLOCKs (refuse-to-proceed).
    diff = _diff_for("src/foo.py")
    ci = _good_ci("run-1", diff, "success")
    ci["conclusion"] = conclusion
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=ci,
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK


def test_gate_ignores_patch_written_success_field() -> None:
    # The gate reads only the authenticated conclusion. A patch-controlled
    # "passed" flag must not flip a failing run to pass.
    diff = _diff_for("src/foo.py")
    ci = _good_ci("run-1", diff, "failure")
    ci["passed"] = True  # attacker-controlled artifact field
    ci["success"] = "true"
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=ci,
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.FAIL


def test_gate_blocks_when_ci_diff_hash_mismatch() -> None:
    # CI verified a different diff than the ledger recorded -> BLOCK.
    diff = _diff_for("src/foo.py")
    ci = _good_ci("run-1", diff, "success")
    ci["diff_hash"] = "tampered"
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=ci,
        expected_run_id="run-1",
    )
    assert result.decision is GateDecision.BLOCK


def test_gate_scope_violation_blocks() -> None:
    diff = _diff_for("src/foo.py", "unrelated/bar.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-1", diff, "success"),
        expected_run_id="run-1",
        allowed_scope=["src/"],
    )
    assert result.decision is GateDecision.BLOCK
    assert any("scope" in r for r in result.reasons)


def test_gate_empty_run_id_raises() -> None:
    diff = _diff_for("src/foo.py")
    with pytest.raises(CiGateError):
        evaluate_ci_gate(
            candidate_diff=diff,
            ledger_hash=_ledger_hash(diff),
            ci_result=_good_ci("run-1", diff),
            expected_run_id="",
        )


def test_gate_result_carries_provenance() -> None:
    diff = _diff_for("src/foo.py")
    h = _ledger_hash(diff)
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=h,
        ci_result=_good_ci("run-7", diff, "success"),
        expected_run_id="run-7",
    )
    assert result.run_id == "run-7"
    assert result.diff_hash == h
    assert result.reasons  # always records why