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,
    gate_weakening_violations,
    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",
        # P3-flip §4.2 — direct code-execution / supply-chain vectors.
        ".gitmodules",
        "vendor/.gitmodules",
        ".husky/pre-commit",
        "frontend/.husky/commit-msg",
        ".githooks/pre-push",
        ".gitattributes",
        "pkg/.gitattributes",
        ".npmrc",
        "web/.npmrc",
        "src/__generated__/schema.ts",
        "api/types.generated.ts",
        "web/dist/bundle.js",
        "service/build/output.o",
        "assets/app.min.js",
    ],
)
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_lockfiles_are_NOT_denylisted_so_the_tier3_fixer_can_ship() -> None:
    """Deliberate (§4.2 note): lockfile RCE is contained by the credential-less
    egress-blocked build sandbox, and the dep-CVE fixer rewrites lockfiles to
    produce draft PRs — so lockfiles are intentionally NOT on the denylist."""
    for lock in ("package-lock.json", "web/yarn.lock", "poetry.lock", "Cargo.lock"):
        assert denylist_violations(_diff_for(lock)) == [], (
            f"{lock!r} must not be denylisted (would break the Tier-3 fixer)"
        )


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_empty_ledger_fails_closed() -> None:
    # An EMPTY expected hash must never be treated as a match (empty != real
    # sha). Fail-closed: there is nothing to bind to.
    diff = _diff_for("a.py")
    assert verify_diff_hash(diff, ledger_hash="") is False


def test_hash_empty_ci_verified_does_not_silently_pass() -> None:
    # A real ledger hash but an EMPTY ci_verified hash must not match (the empty
    # string is not the recomputed sha).
    diff = _diff_for("a.py")
    h = _ledger_hash(diff)
    assert verify_diff_hash(diff, ledger_hash=h, ci_verified_hash="") 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_on_empty_ledger_hash() -> None:
    # Fail-closed: an empty ledger hash binds to nothing -> BLOCK, never a pass,
    # even with an authenticated CI success.
    diff = _diff_for("src/foo.py")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_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_empty_ci_diff_hash() -> None:
    # An empty CI-verified diff_hash must not silently pass the integrity check.
    diff = _diff_for("src/foo.py")
    ci = _good_ci("run-1", diff, "success")
    ci["diff_hash"] = ""
    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_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


# --------------------------------------------------------------------------- #
# §4.5 gate-weakening detector
# --------------------------------------------------------------------------- #


def _diff_adding(added_line: str, path: str = "src/mod.py") -> str:
    """A unified diff that ADDS one line (plus an unchanged context line)."""
    return (
        f"diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n"
        f"@@ -1 +1,2 @@\n unchanged\n+{added_line}\n"
    )


@pytest.mark.parametrize(
    "added_line",
    [
        "x = 1  # noqa",
        "y = 2  # noqa: E501",
        "z = bad()  # type: ignore",
        "z = bad()  # type:ignore[arg-type]",
        "def f():  # pragma: no cover",
        "def f():  # pragma: no-cover",
        "subprocess.run(cmd)  # nosec",
        "eval(x)  # nosemgrep",
        "    git commit --no-verify",
        "@pytest.mark.skip(reason='flaky')",
        "@pytest.mark.xfail",
        "    pytest.skip('todo')",
        "    self.skipTest('later')",
    ],
)
def test_gate_weakening_added_line_flagged(added_line: str) -> None:
    violations = gate_weakening_violations(_diff_adding(added_line))
    assert violations, f"expected {added_line!r} to be flagged as gate-weakening"


def test_gate_weakening_clean_diff_has_no_violations() -> None:
    assert gate_weakening_violations(_diff_for("src/foo.py")) == []


def test_gate_weakening_only_inspects_added_lines() -> None:
    # A diff that REMOVES a suppression (line starts with '-') must not be flagged.
    removing = (
        "diff --git a/m.py b/m.py\n--- a/m.py\n+++ b/m.py\n"
        "@@ -1,2 +1 @@\n-x = 1  # noqa\n unchanged\n"
    )
    assert gate_weakening_violations(removing) == []


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


def test_gate_blocks_weakening_diff_even_if_ci_success() -> None:
    # A diff that adds a coverage suppression BLOCKs even with a green CI run.
    diff = _diff_adding("def f():  # pragma: no cover")
    result = evaluate_ci_gate(
        candidate_diff=diff,
        ledger_hash=_ledger_hash(diff),
        ci_result=_good_ci("run-w", diff, "success"),
        expected_run_id="run-w",
    )
    assert result.decision is GateDecision.BLOCK
    assert result.blocked is True
    assert any("gate-weakening" in r for r in result.reasons)