orchestrator/agent-team/tests/test_ci_gate_workflow.py

"""Tests for the embedded guard logic in ``ci/agent-team-apply-verify.yml``.

The §3.3.2 trust-boundary guard (diff-integrity hash, the trust-control-surface
denylist, the symlink-escape reject, declared-scope enforcement) lives as an
inline Python heredoc inside the CI workflow, so it cannot be imported directly.
These tests extract that script from the YAML and execute it as the workflow
does — via env vars and a diff file — asserting the exit code for good and
adversarial diffs. This backs the workflow's correctness claim with a real,
runnable suite instead of an "verified during authoring" assertion, and guards
the symlink / non-UTF-8 / header-only-section fixes against regression.

It does NOT enable, provision, or run the workflow itself; it only exercises the
pure-code gate the workflow embeds.
"""

from __future__ import annotations

import hashlib
import os
import subprocess
import sys
from pathlib import Path

import pytest

_WORKFLOW = Path(__file__).resolve().parents[1] / "ci" / "agent-team-apply-verify.yml"


def _extract_guard_script() -> str:
    """Pull the first ``python3 - <<'PY' ... PY`` heredoc (the guard) from the YAML.

    The body is indented to sit under the YAML ``run:`` block; we strip the
    common 10-space lead so it is valid module source.
    """
    lines = _WORKFLOW.read_text(encoding="utf-8").splitlines()
    start = end = None
    for i, line in enumerate(lines):
        if start is None and line.strip() == "python3 - <<'PY'":
            start = i + 1
        elif start is not None and line.strip() == "PY":
            end = i
            break
    assert start is not None and end is not None, "guard heredoc not found"
    body = lines[start:end]
    return "\n".join(ln[10:] if ln.startswith(" " * 10) else ln for ln in body)


@pytest.fixture(scope="module")
def guard_script(tmp_path_factory: pytest.TempPathFactory) -> Path:
    path = tmp_path_factory.mktemp("guard") / "guard.py"
    path.write_text(_extract_guard_script(), encoding="utf-8")
    return path


def _run_guard(
    guard_script: Path,
    tmp_path: Path,
    diff: str | bytes,
    scope: str,
    *,
    bad_hash: bool = False,
) -> int:
    raw = diff.encode("utf-8") if isinstance(diff, str) else diff
    diff_path = tmp_path / "candidate.diff"
    diff_path.write_bytes(raw)
    expected = "deadbeef" if bad_hash else hashlib.sha256(raw).hexdigest()
    env = dict(
        os.environ,
        DIFF_PATH=str(diff_path),
        EXPECTED_DIFF_HASH=expected,
        DECLARED_SCOPE=scope,
    )
    result = subprocess.run(
        [sys.executable, str(guard_script)], env=env, capture_output=True, text=True
    )
    return result.returncode


CLEAN = (
    "diff --git a/src/app.py b/src/app.py\n"
    "--- a/src/app.py\n+++ b/src/app.py\n@@ -1 +1 @@\n-x\n+y\n"
)
SYMLINK = (
    "diff --git a/src/link b/src/link\nnew file mode 120000\n"
    "--- /dev/null\n+++ b/src/link\n@@ -0,0 +1 @@\n+../.github/workflows\n"
)
WORKFLOW_DELETE = (
    "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n"
    "deleted file mode 100644\n--- a/.github/workflows/ci.yml\n+++ /dev/null\n"
    "@@ -1 +0,0 @@\n-on: push\n"
)
COPY_TO_DENIED = (
    "diff --git a/src/x.py b/.github/workflows/evil.yml\nsimilarity index 100%\n"
    "copy from src/x.py\ncopy to .github/workflows/evil.yml\n"
)
NON_UTF8 = (
    b"diff --git a/src/app.py b/src/app.py\n--- a/src/app.py\n"
    b"+++ b/src/app.py\n@@ -1 +1 @@\n-x\n+\xff\xfe\n"
)


def test_clean_in_scope_diff_passes(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "src/**") == 0


def test_hash_mismatch_fails(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "src/**", bad_hash=True) == 2


def test_workflow_delete_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    # Header-only section (delete) caught via diff --git, not a +++ body line.
    assert (
        _run_guard(guard_script, tmp_path, WORKFLOW_DELETE, "src/**\n.github/**") == 4
    )


def test_copy_into_denied_path_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, COPY_TO_DENIED, "src/**\n.github/**") == 4


def test_symlink_addition_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    # The symlink-escape vector: rejected outright (exit 7).
    assert _run_guard(guard_script, tmp_path, SYMLINK, "src/**") == 7


def test_non_utf8_diff_fails_closed(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, NON_UTF8, "src/**") == 8


def test_out_of_scope_path_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "other/**") == 6


def test_unscoped_diff_is_rejected(guard_script: Path, tmp_path: Path) -> None:
    assert _run_guard(guard_script, tmp_path, CLEAN, "") == 5


def test_escaping_scope_entries_are_dropped(guard_script: Path, tmp_path: Path) -> None:
    # A parent-escaping scope entry must not widen coverage; it is dropped, so a
    # diff under it is treated as unscoped.
    assert _run_guard(guard_script, tmp_path, CLEAN, "../../etc") == 5