Gitea-Tools/tests/test_stable_branch_push_guard.py

"""Tests for the direct stable-branch push guard (#671).

Covers the acceptance criteria:
1. detect shell ``git push <remote> master`` equivalents
2. detect root/control-checkout local commits not on an issue branch
3. contamination record shape
4. fail-closed gate on review/merge/close/completion (reconciler-exempt)
5. AC5 case matrix: no-op dry-run push, real direct push, sanctioned Gitea
   merge, fetch-only, root-checkout local commit; plus feature-branch push
   still allowed and sanctioned merge still allowed
6. redaction of credentials in logged command summaries
"""

import stable_branch_push_guard as guard


# ── AC1: detect direct stable-branch push equivalents ────────────────────────

def test_plain_git_push_remote_master_is_contamination():
    res = guard.classify_push_command("git push prgs master")
    assert res["is_git_push"] is True
    assert res["targets_stable"] is True
    assert res["stable_refs"] == ["master"]
    assert res["contamination"] is True
    assert res["reasons"]


def test_push_main_and_dev_detected():
    for ref in ("main", "dev", "develop", "development"):
        res = guard.classify_push_command(f"git push origin {ref}")
        assert res["contamination"] is True, ref
        assert res["stable_refs"] == [ref]


def test_head_colon_master_refspec_detected():
    res = guard.classify_push_command("git push prgs HEAD:master")
    assert res["targets_stable"] is True
    assert res["stable_refs"] == ["master"]
    assert res["contamination"] is True


def test_full_refspec_force_plus_detected():
    res = guard.classify_push_command(
        "git push prgs +refs/heads/tmp:refs/heads/master"
    )
    assert res["contamination"] is True
    assert res["is_force"] is True
    assert res["stable_refs"] == ["master"]


def test_force_flag_to_master_detected():
    res = guard.classify_push_command("git push --force prgs master")
    assert res["contamination"] is True
    assert res["is_force"] is True


def test_delete_refspec_master_detected():
    res = guard.classify_push_command("git push prgs :master")
    assert res["contamination"] is True
    assert res["is_delete"] is True


def test_delete_flag_master_detected():
    res = guard.classify_push_command("git push --delete prgs master")
    assert res["contamination"] is True
    assert res["is_delete"] is True


def test_push_detected_inside_compound_command():
    res = guard.classify_push_command("cd repo && git push prgs master && echo ok")
    assert res["contamination"] is True
    assert res["stable_refs"] == ["master"]


# ── AC5: no-op / dry-run push still proves intent ────────────────────────────

def test_dry_run_push_to_master_proves_intent():
    res = guard.classify_push_command("git push --dry-run prgs master")
    assert res["is_dry_run"] is True
    assert res["contamination"] is True
    assert res["proves_intent"] is True
    assert "intent" in " ".join(res["reasons"]).lower()


def test_short_dry_run_flag_n_detected():
    res = guard.classify_push_command("git push -n prgs master")
    assert res["is_dry_run"] is True
    assert res["contamination"] is True


# ── AC5 negative: feature-branch push still allowed ──────────────────────────

def test_feature_branch_push_not_flagged():
    res = guard.classify_push_command(
        "git push prgs fix/issue-671-block-stable-branch-push"
    )
    assert res["is_git_push"] is True
    assert res["targets_stable"] is False
    assert res["contamination"] is False
    assert res["reasons"] == []


def test_feature_branch_head_refspec_not_flagged():
    res = guard.classify_push_command(
        "git push prgs HEAD:fix/issue-671-block-stable-branch-push"
    )
    assert res["contamination"] is False
    assert res["targets_stable"] is False


def test_branch_named_like_master_substring_not_flagged():
    # 'master-notes' is not the stable 'master'.
    res = guard.classify_push_command("git push prgs master-notes")
    assert res["contamination"] is False
    assert res["targets_stable"] is False


# ── AC5 negative: fetch-only operations ──────────────────────────────────────

def test_git_fetch_not_a_push():
    res = guard.classify_push_command("git fetch prgs")
    assert res["is_git_push"] is False
    assert res["is_fetch_or_pull"] is True
    assert res["contamination"] is False


def test_git_pull_ff_only_master_not_a_push():
    res = guard.classify_push_command("git pull --ff-only prgs master")
    assert res["is_git_push"] is False
    assert res["is_fetch_or_pull"] is True
    assert res["contamination"] is False


# ── AC5 negative: sanctioned Gitea merge is not a push ───────────────────────

def test_gitea_merge_pr_tool_is_not_a_push():
    res = guard.classify_push_command("gitea_merge_pr(pr_number=671, remote='prgs')")
    assert res["is_git_push"] is False
    assert res["contamination"] is False


def test_gitea_api_merge_is_not_a_push():
    res = guard.classify_push_command(
        "curl -X POST https://gitea.example/api/v1/repos/o/r/pulls/671/merge"
    )
    assert res["is_git_push"] is False
    assert res["contamination"] is False


# ── ambiguous bare push: reported, not auto-contaminating ────────────────────

def test_bare_push_is_ambiguous_not_contaminating():
    res = guard.classify_push_command("git push prgs")
    assert res["is_git_push"] is True
    assert res["ambiguous_target"] is True
    assert res["contamination"] is False
    assert res["reasons"]  # surfaced as a warning


# ── AC2: root/control-checkout local commit detection ────────────────────────

def test_root_checkout_commit_ahead_of_master_flagged():
    res = guard.assess_root_checkout_local_commit(
        current_branch="master",
        head_sha="a" * 40,
        remote_master_sha="b" * 40,
        is_under_branches=False,
    )
    assert res["contamination"] is True
    assert res["reasons"]


def test_root_checkout_clean_at_master_not_flagged():
    sha = "c" * 40
    res = guard.assess_root_checkout_local_commit(
        current_branch="master",
        head_sha=sha,
        remote_master_sha=sha,
        is_under_branches=False,
    )
    assert res["contamination"] is False
    assert res["unknown"] is False


def test_branches_worktree_commit_is_exempt():
    res = guard.assess_root_checkout_local_commit(
        current_branch="fix/issue-671-block-stable-branch-push",
        head_sha="a" * 40,
        remote_master_sha="b" * 40,
        is_under_branches=True,
    )
    assert res["contamination"] is False


def test_root_checkout_ahead_count_flagged():
    res = guard.assess_root_checkout_local_commit(
        current_branch="master",
        head_sha="",
        remote_master_sha="",
        is_under_branches=False,
        ahead_count=2,
    )
    assert res["contamination"] is True


def test_root_checkout_unknown_when_state_missing():
    res = guard.assess_root_checkout_local_commit(
        current_branch="master",
        head_sha="",
        remote_master_sha="",
        is_under_branches=False,
    )
    assert res["contamination"] is False
    assert res["unknown"] is True


def test_root_checkout_non_stable_branch_out_of_scope():
    res = guard.assess_root_checkout_local_commit(
        current_branch="feature/x",
        head_sha="a" * 40,
        remote_master_sha="b" * 40,
        is_under_branches=False,
    )
    assert res["contamination"] is False


# ── AC3: contamination record shape ──────────────────────────────────────────

def test_build_contamination_record_shape_and_redaction():
    rec = guard.build_contamination_record(
        reason_class="stable_branch_push",
        command_redacted="git push https://user:[email protected]/o/r.git master",
        session_id="prgs-author-123",
        remote="prgs",
        ref="master",
        role="author",
    )
    assert rec["kind"] == guard.CONTAMINATION_KIND
    assert rec["reason_class"] == "stable_branch_push"
    assert rec["remote"] == "prgs"
    assert rec["ref"] == "master"
    assert rec["cleared_by_reconciler"] is False
    # secret must never survive into the durable record
    assert "tok@" not in rec["command_summary"]
    assert "***@" in rec["command_summary"]


# ── AC4: fail-closed gate on gated mutations ─────────────────────────────────

def _marker():
    return guard.build_contamination_record(
        reason_class="stable_branch_push",
        command_redacted="git push prgs master",
        remote="prgs",
        ref="master",
        role="author",
    )


def test_gate_blocks_gated_mutations_for_author():
    marker = _marker()
    for task in ("create_pr", "merge_pr", "close_issue", "review_pr",
                 "submit_pr_review", "commit_files", "close_pr"):
        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
        assert gate["block"] is True, task
        assert gate["reasons"]


def test_gate_allows_comment_and_lock_for_handoff():
    marker = _marker()
    for task in ("comment_issue", "lock_issue", "create_issue", "mark_issue"):
        gate = guard.assess_contamination_gate(marker, task=task, actual_role="author")
        assert gate["block"] is False, task


def test_gate_exempts_reconciler_audit_path():
    marker = _marker()
    gate = guard.assess_contamination_gate(marker, task="close_pr", actual_role="reconciler")
    assert gate["block"] is False


def test_gate_no_marker_allows_everything():
    gate = guard.assess_contamination_gate(None, task="merge_pr", actual_role="merger")
    assert gate["block"] is False


def test_gate_cleared_marker_allows_everything():
    marker = _marker()
    marker["cleared_by_reconciler"] = True
    gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role="merger")
    assert gate["block"] is False


def test_same_worker_cannot_self_clear_by_role():
    # A merger/reviewer/author role is still gated — only reconciler is exempt.
    marker = _marker()
    for role in ("author", "reviewer", "merger"):
        gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role=role)
        assert gate["block"] is True, role


def test_format_gate_error_mentions_issue():
    marker = _marker()
    gate = guard.assess_contamination_gate(marker, task="merge_pr", actual_role="merger")
    msg = guard.format_contamination_gate_error(gate)
    assert "#671" in msg
    assert "contaminat" in msg.lower()


# ── redaction unit coverage ──────────────────────────────────────────────────

def test_redact_url_userinfo():
    out = guard.redact_command("git push https://bob:secretpat@host/o/r.git master")
    assert "secretpat" not in out
    assert "***@" in out
    assert "master" in out  # structure preserved for audit


def test_redact_token_assignment():
    out = guard.redact_command("GITEA_TOKEN=abcdef123456 git push prgs master")
    assert "abcdef123456" not in out
    assert "GITEA_TOKEN=***" in out


def test_redact_empty():
    assert guard.redact_command(None) == ""
    assert guard.redact_command("") == ""


# ── detect_stable_push over iterables ────────────────────────────────────────

def test_detect_stable_push_iterable_finds_first_contamination():
    cmds = ["git status", "git push prgs master", "echo done"]
    res = guard.detect_stable_push(cmds)
    assert res["contamination"] is True


def test_detect_stable_push_iterable_all_safe():
    cmds = ["git status", "git push prgs feature/x", "git fetch prgs"]
    res = guard.detect_stable_push(cmds)
    assert res["contamination"] is False