orchestrator/agent-team/tests/test_fixer.py

"""Unit tests for agent_team.nodes.fixer (Plane-1 Tier-3 fixer, §4 / §7 Phase 5).

The fixer is exercised with FAKE spec/build callables — no Claude, no DeepSeek,
no network, no subprocess. The load-bearing properties under test:

* **Happy path.** A confirmed dependency-cve finding + a fake build returning a
  valid one-line bump diff yields an ``ok`` :class:`FixPlan` with the right diff,
  a real content-hash, the single-manifest declared scope, and the exact CI
  ``workflow_dispatch`` inputs (task_id / artifact / hash / scope).
* **Scope confinement (untrusted-patch hygiene).** A generated diff that touches
  anything beyond the single declared manifest — or the trust-control surface —
  fails SAFE to a no-op plan (no patch, no dispatch).
* **INERT / opt-in.** ``plan_fix`` dispatches nothing; ``dispatch_fix`` with no
  dispatcher raises (the box has no write token) and never fires a workflow.
* **Fail-safe on bad/ambiguous findings.** Wrong check/status/category, missing
  package/fixed_version, an ambiguous fixed version, or an unparseable/empty
  model diff all yield a FAILED plan, never a fabricated fix.
* **No mutation surface.** The module exposes no git/patch/fs-write path.
"""

from __future__ import annotations

import inspect
from typing import Any

import pytest

from agent_team.nodes import fixer
from agent_team.nodes.fixer import (
    FixDispatchInputs,
    FixPlan,
    describe_plan,
    dispatch_fix,
    is_fixable_dependency_finding,
    plan_fix,
)
from agent_team.state_store import compute_content_hash


# A confirmed dependency-cve finding, exactly the shape
# security-review/checkers/dependency-cve.sh emits.
def _finding(**overrides: Any) -> dict[str, Any]:
    base: dict[str, Any] = {
        "repo": "example-repo",
        "id": "example-repo-vuln-requests-2-19-0-CVE-2018-18074",
        "title": "requests 2.19.0 is vulnerable (CVE-2018-18074)",
        "severity": "high",
        "category": "other",
        "check": "vulnerable-dependency",
        "status": "confirmed",
        "proof": {
            "package": "requests",
            "version": "2.19.0",
            "advisory_id": "CVE-2018-18074",
            "summary": "requests before 2.20.0 leaks auth on redirect",
            "fixed_version": "2.20.0",
        },
    }
    base.update(overrides)
    return base


# A valid one-line bump diff confined to requirements.txt.
_BUMP_DIFF = (
    "diff --git a/requirements.txt b/requirements.txt\n"
    "--- a/requirements.txt\n"
    "+++ b/requirements.txt\n"
    "@@ -1,1 +1,1 @@\n"
    "-requests==2.19.0\n"
    "+requests==2.20.0\n"
)


def _fake_build(diff_text: str) -> Any:
    """Return a fake BuildCallable that yields ``diff_text`` (ignores instruction)."""

    def _build(_instruction: str) -> str:
        return diff_text

    return _build


def _fake_spec(_finding: Any) -> str:
    return "Bump requests==2.19.0 to requests==2.20.0 in the manifest only."


# --------------------------------------------------------------------------- #
# Happy path: correct dep-bump patch + hash + dispatch inputs
# --------------------------------------------------------------------------- #


def test_plan_fix_produces_correct_patch_hash_and_dispatch() -> None:
    plan = plan_fix(
        _finding(),
        task_id="t-123",
        spec=_fake_spec,
        build=_fake_build(_BUMP_DIFF),
    )
    assert plan.ok is True
    assert plan.reason == ""
    # _extract_diff normalizes (strips trailing whitespace); the body is intact.
    assert plan.diff == _BUMP_DIFF.strip()
    assert "requests==2.19.0" in plan.diff and "requests==2.20.0" in plan.diff
    # Hash matches the foundation content-hash CI re-computes over the SAME bytes
    # CI uploads as candidate.diff (boundary #3).
    assert plan.diff_hash == compute_content_hash(plan.diff.encode("utf-8"))
    assert plan.manifest_path == "requirements.txt"
    assert plan.declared_scope == ["requirements.txt"]

    # Exact CI workflow_dispatch contract (ci/agent-team-apply-verify.yml inputs).
    assert plan.dispatch is not None
    inputs = plan.dispatch.as_inputs()
    assert set(inputs) == {
        "task_id",
        "diff_artifact_name",
        "expected_diff_hash",
        "declared_scope",
    }
    assert inputs["task_id"] == "t-123"
    assert inputs["diff_artifact_name"] == "candidate-diff-t-123"
    assert inputs["expected_diff_hash"] == plan.diff_hash
    # Single-manifest scope, newline-joined (the workflow splits on newlines).
    assert inputs["declared_scope"] == "requirements.txt"


def test_dispatch_hash_binds_to_diff_for_ci_gate() -> None:
    """The dispatched expected_diff_hash equals sha256(candidate.diff) — the
    binding ci_gate / the guard job verify before apply."""
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    assert plan.dispatch is not None
    # What CI uploads as candidate.diff is plan.diff; its hash is the binding.
    assert (
        compute_content_hash(plan.diff.encode("utf-8"))
        == plan.dispatch.expected_diff_hash
    )


def test_spec_callable_raising_fails_safe() -> None:
    """A spec callable that raises fails SAFE to a no-op plan (no patch/dispatch)."""

    def _boom(_f: Any) -> str:
        raise RuntimeError("claude down")

    plan = plan_fix(
        _finding(), task_id="t-1", spec=_boom, build=_fake_build(_BUMP_DIFF)
    )
    assert plan.ok is False
    assert "fix-spec generation failed" in plan.reason
    assert plan.diff == ""
    assert plan.dispatch is None


def test_empty_spec_is_tolerated_but_patch_still_required() -> None:
    plan = plan_fix(
        _finding(),
        task_id="t-1",
        spec=lambda _f: "   ",
        build=_fake_build(_BUMP_DIFF),
    )
    assert plan.ok is True
    assert plan.spec == ""  # empty spec normalized; patch carries the fix


# --------------------------------------------------------------------------- #
# Scope confinement / untrusted-patch hygiene
# --------------------------------------------------------------------------- #


def test_patch_touching_extra_file_is_rejected() -> None:
    out_of_scope = _BUMP_DIFF + (
        "diff --git a/agent_team/secret.py b/agent_team/secret.py\n"
        "--- a/agent_team/secret.py\n"
        "+++ b/agent_team/secret.py\n"
        "@@ -1,1 +1,1 @@\n"
        "-x = 1\n"
        "+x = 2\n"
    )
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(out_of_scope)
    )
    assert plan.ok is False
    assert "escapes the declared single-manifest scope" in plan.reason
    assert plan.dispatch is None


def test_patch_touching_trust_control_surface_is_rejected() -> None:
    """A diff that touches a denylisted path (a workflow) fails SAFE even though
    it claims to be a dep bump."""
    malicious = (
        "diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml\n"
        "--- a/.github/workflows/ci.yml\n"
        "+++ b/.github/workflows/ci.yml\n"
        "@@ -1,1 +1,1 @@\n"
        "-on: push\n"
        "+on: pull_request_target\n"
    )
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(malicious)
    )
    assert plan.ok is False
    assert "trust-control denylist" in plan.reason
    assert plan.dispatch is None


def test_patch_renaming_into_denied_path_is_rejected() -> None:
    rename_diff = (
        "diff --git a/requirements.txt b/.github/workflows/evil.yml\n"
        "rename from requirements.txt\n"
        "rename to .github/workflows/evil.yml\n"
    )
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(rename_diff)
    )
    assert plan.ok is False
    assert plan.dispatch is None


# --------------------------------------------------------------------------- #
# INERT / opt-in: nothing dispatched by default
# --------------------------------------------------------------------------- #


def test_plan_fix_dispatches_nothing() -> None:
    """plan_fix only PLANS — it never calls a dispatcher (none is bindable here)."""
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    # The plan carries dispatch inputs but the act of planning fired nothing.
    assert isinstance(plan.dispatch, FixDispatchInputs)


def test_dispatch_fix_without_dispatcher_raises_and_fires_nothing() -> None:
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    with pytest.raises(RuntimeError, match="no workflow dispatcher bound"):
        dispatch_fix(plan, dispatcher=None)


def test_dispatch_fix_refuses_non_ok_plan() -> None:
    bad = FixPlan.failed("nope", finding_id="x")
    calls: list[Any] = []

    def _disp(_inputs: FixDispatchInputs, _diff: str) -> str:
        calls.append(_inputs)
        return "ref"

    with pytest.raises(ValueError, match="non-ok fix plan"):
        dispatch_fix(bad, dispatcher=_disp)
    assert calls == []  # the dispatcher was never called


def test_dispatch_fix_with_explicit_dispatcher_forwards_diff_and_inputs() -> None:
    """When (and only when) a dispatcher is explicitly bound does anything fire."""
    plan = plan_fix(
        _finding(), task_id="t-9", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    seen: dict[str, Any] = {}

    def _disp(inputs: FixDispatchInputs, diff: str) -> str:
        seen["inputs"] = inputs
        seen["diff"] = diff
        return "dispatch-ref-42"

    ref = dispatch_fix(plan, dispatcher=_disp)
    assert ref == "dispatch-ref-42"
    assert seen["diff"] == plan.diff
    assert seen["inputs"].expected_diff_hash == plan.diff_hash


# --------------------------------------------------------------------------- #
# Fail-safe on bad / ambiguous findings
# --------------------------------------------------------------------------- #


@pytest.mark.parametrize(
    "mutation, why_substr",
    [
        ({"check": "secrets"}, "not a dependency-cve"),
        ({"status": "unverified"}, "not confirmed"),
        ({"category": "injection"}, "unexpected finding category"),
    ],
)
def test_non_dependency_findings_are_not_fixable(
    mutation: dict[str, Any], why_substr: str
) -> None:
    fixable, why = is_fixable_dependency_finding(_finding(**mutation))
    assert fixable is False
    assert why_substr in why


@pytest.mark.parametrize(
    "proof_mutation, why_substr",
    [
        ({"package": ""}, "package is missing"),
        ({"version": ""}, "version is missing"),
        ({"fixed_version": ""}, "fixed_version is missing"),
        ({"fixed_version": ">=2.20.0"}, "ambiguous fixed_version"),
        ({"fixed_version": "2.20.0, 2.21.0"}, "ambiguous fixed_version"),
        ({"fixed_version": "2.19.0"}, "equals fixed_version"),
    ],
)
def test_bad_proof_findings_are_not_fixable(
    proof_mutation: dict[str, Any], why_substr: str
) -> None:
    f = _finding()
    f["proof"].update(proof_mutation)
    fixable, why = is_fixable_dependency_finding(f)
    assert fixable is False
    assert why_substr in why


def test_non_mapping_finding_is_not_fixable() -> None:
    fixable, why = is_fixable_dependency_finding("not a dict")
    assert fixable is False
    assert "must be a mapping" in why


def test_plan_fix_on_bad_finding_yields_failed_plan_no_patch() -> None:
    plan = plan_fix(
        _finding(check="secrets"),
        task_id="t-1",
        spec=_fake_spec,
        build=_fake_build(_BUMP_DIFF),
    )
    assert plan.ok is False
    assert "not a fixable dependency-cve" in plan.reason
    assert plan.diff == ""
    assert plan.dispatch is None


def test_plan_fix_missing_task_id_fails_safe() -> None:
    plan = plan_fix(
        _finding(), task_id="  ", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    assert plan.ok is False
    assert "task_id is missing" in plan.reason


def test_plan_fix_unparseable_diff_fails_safe() -> None:
    plan = plan_fix(
        _finding(),
        task_id="t-1",
        spec=_fake_spec,
        build=_fake_build("I could not generate a diff, sorry."),
    )
    assert plan.ok is False
    assert "not a usable unified diff" in plan.reason
    assert plan.dispatch is None


def test_plan_fix_empty_build_output_fails_safe() -> None:
    plan = plan_fix(
        _finding(), task_id="t-1", spec=_fake_spec, build=_fake_build("   ")
    )
    assert plan.ok is False
    assert "empty output" in plan.reason


def test_plan_fix_build_raises_fails_safe() -> None:
    def _boom(_instruction: str) -> str:
        raise RuntimeError("orchestrator down")

    plan = plan_fix(_finding(), task_id="t-1", spec=_fake_spec, build=_boom)
    assert plan.ok is False
    assert "patch generation failed" in plan.reason


# --------------------------------------------------------------------------- #
# describe_plan (dry-run rendering)
# --------------------------------------------------------------------------- #


def test_describe_plan_ok_shows_spec_patch_and_dispatch() -> None:
    plan = plan_fix(
        _finding(), task_id="t-7", spec=_fake_spec, build=_fake_build(_BUMP_DIFF)
    )
    text = describe_plan(plan)
    assert "fix spec (Claude)" in text
    assert "candidate patch (DeepSeek" in text
    assert "workflow_dispatch inputs" in text
    assert "t-7" in text
    assert plan.diff_hash in text
    assert "DRY-RUN: nothing dispatched" in text


def test_describe_plan_failed_shows_failsafe_and_no_dispatch() -> None:
    plan = FixPlan.failed("bad finding", finding_id="abc")
    text = describe_plan(plan)
    assert "FIX NOT PLANNED" in text
    assert "dispatch: NONE" in text


# --------------------------------------------------------------------------- #
# No mutation surface (inert by construction)
# --------------------------------------------------------------------------- #


def test_module_has_no_patch_apply_or_git_or_write_surface() -> None:
    """No git/patch/subprocess/fs-write CALLS in executable code.

    Docstrings legitimately MENTION these terms (to state what the fixer does
    NOT do), so we strip docstrings + comments via the tokenizer and inspect only
    executable tokens. By construction the fixer cannot mutate the tree or push.
    """
    import io
    import tokenize

    src = inspect.getsource(fixer)
    code_tokens: list[str] = []
    for tok in tokenize.generate_tokens(io.StringIO(src).readline):
        if tok.type in (tokenize.COMMENT, tokenize.STRING):
            continue
        code_tokens.append(tok.string)
    code = " ".join(code_tokens)
    # The fixer must not call out to git/patch/subprocess or write the filesystem.
    for forbidden in ("subprocess", "Popen", "write_text", "system", "popen"):
        assert forbidden not in code, f"fixer must not reference {forbidden!r} in code"
    # No bare write-mode open(...) or git invocation in executable code.
    assert "git" not in code, "fixer must not invoke git in code"