Gitea-Tools/tests/test_stable_branch_contamination_server.py

"""Server wiring for stable-branch push contamination (#671).

Exercises the MCP tools and the pre-flight enforcement gate against the durable
contamination marker (isolated per-test state dir from conftest).
"""

from __future__ import annotations

import os
from unittest.mock import patch

import gitea_mcp_server as srv


def _clear_marker(remote="prgs"):
    srv._clear_stable_contamination_marker(remote=remote)


def teardown_function():
    _clear_marker()


# ── record tool marks a real direct push ─────────────────────────────────────

def test_record_tool_marks_direct_master_push():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    assert res["contaminated"] is True
    assert res["marked"] is True
    assert res["marker"] is not None
    assert res["marker"]["reason_class"] == "stable_branch_push"
    # loaded back through the durable store
    loaded = srv._load_stable_contamination_marker("prgs")
    assert loaded is not None
    assert loaded["ref"] == "master"


def test_record_tool_dry_run_still_marks():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git push --dry-run prgs master", remote="prgs"
    )
    assert res["contaminated"] is True
    assert res["marked"] is True


def test_record_tool_feature_branch_does_not_mark():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs fix/issue-671-block-stable-branch-push",
        remote="prgs",
    )
    assert res["contaminated"] is False
    assert res["marked"] is False
    assert srv._load_stable_contamination_marker("prgs") is None


def test_record_tool_fetch_only_does_not_mark():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git fetch prgs", remote="prgs"
    )
    assert res["contaminated"] is False
    assert res["marked"] is False


def test_record_tool_mark_false_is_readonly():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs", mark=False
    )
    assert res["contaminated"] is True
    assert res["marked"] is False
    assert srv._load_stable_contamination_marker("prgs") is None


def test_record_tool_redacts_secret_in_marker():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command="git push https://u:supersecret@host/o/r.git master",
        remote="prgs",
    )
    assert res["marked"] is True
    assert "supersecret" not in res["marker"]["command_summary"]


def test_record_tool_root_checkout_local_commit_marks():
    _clear_marker()
    res = srv.gitea_record_stable_branch_push_attempt(
        command=None,
        remote="prgs",
        current_branch="master",
        head_sha="a" * 40,
        remote_master_sha="b" * 40,
        is_under_branches=False,
    )
    assert res["contaminated"] is True
    assert res["marked"] is True
    assert res["marker"]["reason_class"] == "root_checkout_commit"


# ── audit tool: inspect + reconciler-only clear ──────────────────────────────

def test_audit_inspect_reports_marker():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    out = srv.gitea_audit_stable_branch_contamination(action="inspect", remote="prgs")
    assert out["contaminated"] is True
    assert out["read_only"] is True


def test_audit_clear_refused_for_non_reconciler():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    with patch.object(srv, "_actual_profile_role", return_value="author"):
        out = srv.gitea_audit_stable_branch_contamination(action="clear", remote="prgs")
    assert out["success"] is False
    assert out["reasons"]
    # marker survives a refused clear
    assert srv._load_stable_contamination_marker("prgs") is not None


def test_audit_clear_allowed_for_reconciler():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    identity = srv._stable_contamination_profile_identity()
    with patch.object(srv, "_actual_profile_role", return_value="reconciler"):
        out = srv.gitea_audit_stable_branch_contamination(
            action="clear", remote="prgs", profile_identity=identity
        )
    assert out["success"] is True
    assert srv._load_stable_contamination_marker("prgs") is None


# ── pre-flight enforcement gate ──────────────────────────────────────────────

def _force_gate_env():
    return patch.dict(os.environ, {"GITEA_TEST_FORCE_STABLE_CONTAMINATION": "1"})


def test_gate_blocks_gated_mutation_when_contaminated():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
        for task in ("merge_pr", "review_pr", "close_issue", "create_pr"):
            try:
                srv._enforce_stable_branch_contamination_gate(task, "prgs")
                raised = False
            except RuntimeError as exc:
                raised = True
                assert "#671" in str(exc)
            assert raised, task


def test_gate_allows_comment_for_handoff_when_contaminated():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
        # must not raise — worker can still post the durable audit comment
        srv._enforce_stable_branch_contamination_gate("comment_issue", "prgs")
        srv._enforce_stable_branch_contamination_gate("lock_issue", "prgs")


def test_gate_exempts_reconciler_when_contaminated():
    _clear_marker()
    srv.gitea_record_stable_branch_push_attempt(
        command="git push prgs master", remote="prgs"
    )
    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="reconciler"):
        srv._enforce_stable_branch_contamination_gate("merge_pr", "prgs")


def test_gate_noop_when_not_contaminated():
    _clear_marker()
    with _force_gate_env(), patch.object(srv, "_actual_profile_role", return_value="author"):
        srv._enforce_stable_branch_contamination_gate("merge_pr", "prgs")