Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 77 Pull Requests 10 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for issue 321

Browse Source
This commit is contained in:
sysadmin
2026-07-07 05:20:39 -04:00
parent 8c75b47709 deb228d629
commit 10997230e3
9 changed files with 1148 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_git_ref_mutation_report.py
+101
View File
@@ -0,0 +1,101 @@
"""Git ref mutations must be reported, never hidden as diagnostics (#297).
``git fetch`` updates remote-tracking refs even though it edits no files.
Reports that ran fetch (or any ref-updating command) must carry a
``Git ref mutations`` entry and may not claim ``Git/worktree mutations:
None`` or classify the fetch as read-only diagnostics.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import review_proofs
def _assess(report, commands):
return review_proofs.assess_git_ref_mutation_report(
report, command_log=commands)
class TestGitRefMutationDetection(unittest.TestCase):
def test_fetch_only_review_with_proper_ledger_passes(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Review decision: APPROVE",
])
res = _assess(report, ["git fetch prgs master"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], ["git fetch prgs master"])
def test_fetch_without_ref_mutation_line_is_blocked(self):
report = "Review decision: APPROVE\nMutations: None\n"
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("git ref mutations" in r.lower() for r in res["reasons"]))
def test_fetch_with_remote_branch_requires_fetched_target_in_report(self):
report = "Git ref mutations: some refs updated\n"
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("prgs/master" in r for r in res["reasons"]))
def test_git_worktree_mutations_none_rejected_when_fetch_occurred(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Git/worktree mutations: None",
])
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("git/worktree mutations" in r.lower() for r in res["reasons"]))
def test_fetch_classified_as_read_only_diagnostics_rejected(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Read-only diagnostics: git fetch prgs master, git status",
])
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("read-only" in r.lower() for r in res["reasons"]))
def test_worktree_creation_is_not_a_ref_mutation(self):
report = "Worktree mutations: created branches/review-pr9\n"
res = _assess(report, ["git worktree add branches/review-pr9 prgs/master"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], [])
def test_merge_command_counts_as_ref_mutation(self):
report = "Review decision: APPROVE\n"
res = _assess(report, ["git merge --no-ff feat/x"])
self.assertFalse(res["proven"])
self.assertIn("git merge --no-ff feat/x", res["ref_mutating_commands"])
def test_cleanup_branch_delete_counts_as_ref_mutation(self):
report = "Cleanup mutations: deleted branch feat/x\n"
res = _assess(report, ["git branch -d feat/x"])
self.assertFalse(res["proven"])
self.assertIn("git branch -d feat/x", res["ref_mutating_commands"])
def test_no_mutation_blocked_handoff_passes(self):
report = "\n".join([
"Git/worktree mutations: None",
"Current status: blocked handoff, no mutations performed",
])
res = _assess(report, ["git status --porcelain", "git log --oneline -1"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], [])
def test_command_log_dict_entries_supported(self):
report = "Git ref mutations: fetched prgs/master\n"
res = _assess(report, [{"command": "git fetch prgs master"}])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(
res["ref_mutating_commands"], ["git fetch prgs master"])
if __name__ == "__main__":
unittest.main()
tests/test_llm_workflow_split.py
+7 -1
View File
@@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow():
def test_create_issue_final_report_verifier_exported():
from review_proofs import assess_create_issue_final_report
assert callable(assess_create_issue_final_report)
assert callable(assess_create_issue_final_report)
def test_non_mergeable_skip_verifier_exported():
from review_proofs import assess_non_mergeable_skip_proof
assert callable(assess_non_mergeable_skip_proof)
tests/test_mcp_server.py
+18 -3
View File
@@ -48,6 +48,21 @@ import mcp_server
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
_NO_BLOCKER_FEEDBACK = {
"success": True,
"has_blocking_change_requests": False,
"review_feedback_stale": False,
}
def _mark_request_changes_ready(pr_number=8, **kwargs):
"""Mark a request_changes decision ready with the #332 duplicate-
suppression feedback fetch stubbed to 'no existing blocker'."""
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
return gitea_mark_final_review_decision(
pr_number, "request_changes", **kwargs)
def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
return {
@@ -1839,7 +1854,7 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 9, "state": "REQUEST_CHANGES"},
@@ -1862,7 +1877,7 @@ class TestSubmitPrReview(unittest.TestCase):
self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
def test_request_changes_blocked_without_eligibility(self):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -2144,7 +2159,7 @@ class TestSubmitPrReview(unittest.TestCase):
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
tests/test_review_proofs.py
+120
View File
@@ -25,6 +25,7 @@ from review_proofs import ( # noqa: E402
ISSUE_SELECTION_REPRESENTED_BY_OPEN_PR,
assess_author_pr_report,
assess_queue_ordering_proof,
assess_reviewer_baseline_validation_proof,
assess_capability_evidence,
assess_capability_proof,
assess_contradictory_no_pr_claim,
@@ -2266,6 +2267,125 @@ class TestQueueOrderingProof(unittest.TestCase):
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["queue_ordering_proven"])
class TestReviewerBaselineValidationProof(unittest.TestCase):
"""Issue #325: baseline validation must use branches/ worktrees."""
ROOT = "/repo/Gitea-Tools"
def _good_baseline_proof(self, **overrides):
proof = {
"worktree_path": f"{self.ROOT}/branches/baseline-master-pr280",
"baseline_target_sha": PINNED,
"pr_head_sha": OTHER,
"baseline_failures": ["tests/test_foo.py::test_bar"],
"pr_failures": ["tests/test_foo.py::test_bar"],
"failure_signatures_match": True,
"clean_before": True,
"clean_after": True,
}
proof.update(overrides)
return proof
def test_main_checkout_test_run_blocks(self):
result = assess_reviewer_baseline_validation_proof(
validation_runs=[
{
"command": "venv/bin/pytest tests/",
"working_directory": self.ROOT,
"project_root": self.ROOT,
}
],
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(result["violations"])
def test_branches_worktree_test_run_passes(self):
result = assess_reviewer_baseline_validation_proof(
validation_runs=[
{
"command": "venv/bin/pytest tests/",
"working_directory": f"{self.ROOT}/branches/review-pr-280",
"project_root": self.ROOT,
}
],
project_root=self.ROOT,
)
self.assertTrue(result["proven"])
def test_preexisting_claim_without_baseline_proof_fails(self):
result = assess_reviewer_baseline_validation_proof(
report_text=(
"Validation: full-suite failures are pre-existing on master."
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["claims_preexisting"])
def test_preexisting_claim_with_complete_baseline_proof_passes(self):
result = assess_reviewer_baseline_validation_proof(
report_text="Failures are pre-existing on master.",
baseline_proof=self._good_baseline_proof(),
project_root=self.ROOT,
)
self.assertTrue(result["proven"])
def test_mismatched_failures_block_preexisting_claim(self):
result = assess_reviewer_baseline_validation_proof(
report_text="Same as master.",
baseline_proof=self._good_baseline_proof(
failure_signatures_match=False,
pr_failures=["tests/test_other.py::test_other"],
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
def test_report_parsed_main_checkout_cwd_blocks(self):
result = assess_reviewer_baseline_validation_proof(
report_text=(
"- Validation command: venv/bin/pytest tests/\n"
f"- Working directory: {self.ROOT}\n"
),
project_root=self.ROOT,
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
def test_build_final_report_downgrades_main_checkout_validation(self):
report = build_final_report(
checkout_proof=_good_checkout(),
inventory=_good_inventory(),
validation=_good_validation(),
contamination=_good_contamination(),
identity_eligible=True,
merge_performed=False,
issue_status_verified=True,
capability_evidence=_good_capability_evidence(),
sweep=_good_sweep(),
live_state=_good_live_state(),
role_boundary=_good_role_boundary(),
review_mutation=_good_review_mutation(),
controller_handoff=_good_handoff(),
capability_proof=_good_capability_proof(),
sweep_proof=_good_secret_sweep(),
worktree_proof={
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
},
report_text=(
"- Validation command: venv/bin/pytest tests/\n"
f"- Working directory: {self.ROOT}\n"
),
project_root=self.ROOT,
)
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["baseline_validation_proven"])
if __name__ == "__main__":
unittest.main()
tests/test_reviewer_mergeability_skip.py
+150
View File
@@ -0,0 +1,150 @@
"""Tests for non-mergeable skip conflict-proof verifier (#322)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_mergeability_skip import assess_non_mergeable_skip_proof # noqa: E402
HEAD_A = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
HEAD_B = "1111111111111111111111111111111111111111"
def _good_skip_report(pr_number: int, head: str, files: str = "review_proofs.py") -> str:
return "\n".join([
f"Skipped PR #{pr_number} (non-mergeable).",
f"- PR number: #{pr_number}",
f"- Current head SHA: {head}",
"- Mergeability result: false",
"- Conflict proof command: git merge-tree $(git merge-base prgs/master prgs/feat) prgs/master prgs/feat",
f"- Conflicting files: {files}",
"- Head unchanged since prior blocker",
"- Blocking category: merge conflict",
])
class TestNonMergeableSkipProof(unittest.TestCase):
def test_no_skips_passes(self):
report = "Selected PR #203 for review. No earlier PRs skipped."
result = assess_non_mergeable_skip_proof(report, skipped_prs=[])
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_documented_non_mergeable_skip_passes(self):
report = _good_skip_report(282, HEAD_A)
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
"conflicting_files": ["review_proofs.py"],
"head_changed_since_prior_blocker": False,
}],
)
self.assertTrue(result["proven"])
def test_skip_without_conflict_proof_blocks(self):
report = "\n".join([
"Skipped PR #284 because it was not mergeable.",
f"Head SHA: {HEAD_B}",
])
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 284,
"mergeable": False,
"head_sha": HEAD_B,
"mergeability_verified": True,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertTrue(any("conflict proof" in r for r in result["reasons"]))
def test_missing_head_sha_blocks(self):
report = _good_skip_report(282, HEAD_A).replace(HEAD_A, "")
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
}],
)
self.assertFalse(result["proven"])
def test_mergeability_unverified_classification_passes(self):
report = "\n".join([
"Skipped PR #282.",
"Classification: MERGEABILITY_UNVERIFIED",
"Conflict proof could not be retrieved in this session.",
])
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"mergeability_verified": False,
}],
)
self.assertTrue(result["proven"])
def test_unverified_without_classification_blocks(self):
report = "Skipped PR #282 due to conflicts."
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"mergeability_verified": False,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(any("MERGEABILITY_UNVERIFIED" in r for r in result["reasons"]))
def test_non_mergeable_without_file_details_still_passes_with_command(self):
report = _good_skip_report(284, HEAD_B, files="not available")
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 284,
"mergeable": False,
"head_sha": HEAD_B,
"mergeability_verified": True,
"conflicting_files": [],
}],
)
self.assertTrue(result["proven"])
def test_stale_head_requires_head_change_field(self):
report = _good_skip_report(282, HEAD_A).replace(
"- Head unchanged since prior blocker",
"",
)
result = assess_non_mergeable_skip_proof(
report,
skipped_prs=[{
"pr_number": 282,
"mergeable": False,
"head_sha": HEAD_A,
"mergeability_verified": True,
"head_changed_since_prior_blocker": True,
}],
)
self.assertFalse(result["proven"])
self.assertTrue(any("head-changed" in r for r in result["reasons"]))
class TestExport(unittest.TestCase):
def test_review_proofs_reexport(self):
from review_proofs import assess_non_mergeable_skip_proof as exported
self.assertTrue(callable(exported))
if __name__ == "__main__":
unittest.main()
tests/test_terminal_review_hard_stop.py
+187
View File
@@ -0,0 +1,187 @@
"""Tests for the session hard-stop after a terminal review mutation (#332).
Extends the single-terminal-decision machinery (#211): after a live
REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
for that same PR may continue; a different PR can never be reviewed or
merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
suppressed.
"""
import os
import unittest
from unittest.mock import patch
import mcp_server
def _lock(mutations=None, correction=False):
return {
"task": "review_pr",
"remote": "prgs",
"session_pid": os.getpid(),
"session_profile": "prgs-reviewer",
"session_profile_lock": "",
"final_review_decision_ready": False,
"ready_pr_number": None,
"ready_action": None,
"ready_expected_head_sha": None,
"ready_remote": None,
"ready_org": None,
"ready_repo": None,
"live_mutations": list(mutations or []),
"correction_authorized": correction,
"correction_reason": None,
}
def _seed(mutations=None, correction=False):
mcp_server._save_review_decision_lock(_lock(mutations, correction))
APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
"review_state": "approve"}
RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
"review_state": "request_changes"}
class TestTerminalHardStopReasons(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_no_lock_no_reasons(self):
mcp_server._save_review_decision_lock(None)
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
def test_no_terminal_mutation_no_reasons(self):
_seed()
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
def test_request_changes_blocks_everything(self):
_seed([RC_A])
for op in ("merge", "mark_ready", "review"):
for pr in (5, 6):
reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
self.assertTrue(reasons, f"{op}/{pr}")
self.assertIn("request_changes on PR #5", reasons[0])
self.assertIn("#332", reasons[0])
def test_approve_allows_same_pr_merge_only(self):
_seed([APPROVED_A])
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "review"))
def test_correction_reopens_review_path_only(self):
_seed([RC_A], correction=True)
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
# correction never opens a cross-PR merge
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
class TestMergeHardStopWiring(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_merge_blocked_after_request_changes(self):
_seed([RC_A])
result = mcp_server.gitea_merge_pr(
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
self.assertFalse(result["performed"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
def test_merge_of_other_pr_blocked_after_approve(self):
_seed([APPROVED_A])
result = mcp_server.gitea_merge_pr(
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
self.assertFalse(result["performed"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
class TestMarkFinalHardStopWiring(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_mark_ready_blocked_after_terminal_mutation(self):
_seed([RC_A])
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="approve", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
def _feedback(blocking, stale=False, success=True):
return {
"success": success,
"has_blocking_change_requests": blocking,
"review_feedback_stale": stale,
"current_head_sha": "abc123",
}
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_duplicate_request_changes_blocked_at_same_head(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("duplicate" in r for r in result["reasons"]),
result["reasons"])
def test_request_changes_allowed_when_prior_blocker_stale(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=True)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_allowed_when_no_blocker(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_fails_closed_when_feedback_unavailable(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False,
success=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("could not verify" in r for r in result["reasons"]),
result["reasons"])
def test_approve_does_not_fetch_feedback(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
side_effect=AssertionError("must not be called")):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="approve", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
if __name__ == "__main__":
unittest.main()