diff --git a/review_proofs.py b/review_proofs.py index 31b0b65..b8b41a5 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1005,6 +1005,7 @@ HANDOFF_ROLE_FIELDS = { ), "author": ( ("Selected issue", ("selected issue",)), + ("Issue lock proof", ("issue lock proof", "lock before diff")), ("Claim/comment status", ("claim/comment status", "claim status", "claim")), ("PR number opened", ("pr number opened", "pr opened", "pr number")), diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index aa90d88..9a80d8d 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -65,6 +65,20 @@ CREATE_PR_ENV = { ), } +ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" + + +def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides): + record = { + "issue_number": issue_number, + "branch_name": branch_name, + "remote": "dadeschools", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + } + record.update(overrides) + return record + # --------------------------------------------------------------------------- # Create Issue @@ -127,15 +141,19 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} with patch.dict(os.environ, CREATE_PR_ENV, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertEqual(result["number"], 3) self.assertNotIn("url", result) + mock_exists.assert_called_with(ISSUE_LOCK_FILE) + mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") payload = mock_api.call_args[0][3] self.assertEqual(payload["head"], "feat/x") self.assertEqual(payload["base"], "main") + self.assertIn("Closes #123", payload["title"]) @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, [])) @@ -144,13 +162,28 @@ class TestCreatePR(unittest.TestCase): @patch("os.path.exists", return_value=True) @patch("builtins.open") def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role): - mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}' + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"} env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} with patch.dict(os.environ, env, clear=True): result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main") self.assertIn("pulls/3", result["url"]) + @patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", + return_value=(True, [])) + @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) + @patch("os.path.exists", return_value=True) + @patch("builtins.open") + def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role): + lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x")) + mock_open.return_value.__enter__.return_value.read.return_value = lock_json + with patch.dict(os.environ, CREATE_PR_ENV, clear=True): + with self.assertRaises(ValueError) as ctx: + gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main") + self.assertIn("Closes #123", str(ctx.exception)) + mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") + # --------------------------------------------------------------------------- # Close Issue @@ -2770,8 +2803,8 @@ class TestIssueLocking(unittest.TestCase): """Test issue locking and PR gating constraints.""" def tearDown(self): - if os.path.exists("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) @@ -2779,7 +2812,7 @@ class TestIssueLocking(unittest.TestCase): mock_api.return_value = [] # no open PRs res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertTrue(res["success"]) - self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json")) + self.assertTrue(os.path.exists(ISSUE_LOCK_FILE)) def test_lock_issue_mismatch_branch_fails(self): with self.assertRaises(ValueError) as ctx: @@ -2816,8 +2849,8 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_lock_fails(self, _auth, _role): - if os.path.exists("/tmp/gitea_issue_lock.json"): - os.remove("/tmp/gitea_issue_lock.json") + if os.path.exists(ISSUE_LOCK_FILE): + os.remove(ISSUE_LOCK_FILE) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(RuntimeError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs") @@ -2827,8 +2860,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_branch_mismatch_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs") @@ -2838,8 +2872,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_forbidden_terms_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): for term in ("equivalent to #196", "related to #196", "same as #196"): with self.assertRaises(ValueError) as ctx: @@ -2850,8 +2885,9 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_missing_closes_ref_fails(self, _auth, _role): - with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f: - json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f) + with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: + json.dump(_sample_issue_lock( + issue_number=196, branch_name="feat/issue-196-mutations"), f) with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(ValueError) as ctx: gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs") diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 22748ce..9f9e57c 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -944,24 +944,46 @@ class TestControllerHandoff(unittest.TestCase): result = assess_controller_handoff(complete, role="review") self.assertEqual(result["verdict"], "complete") - def test_author_role_requires_author_fields(self): - complete = self.BASE_HANDOFF + "\n" + "\n".join([ - "- Selected issue: #182", + def _author_role_fields(self, issue_number=182, pr_number=999): + return [ + f"- Selected issue: #{issue_number}", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", - "- PR number opened: #999", + f"- PR number opened: #{pr_number}", "- No review/merge: confirmed", - ]) + ] + + def test_handoff_role_fields_author_includes_issue_lock_proof(self): + from review_proofs import HANDOFF_ROLE_FIELDS + names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]] + self.assertIn("Issue lock proof", names) + + def test_author_role_requires_author_fields(self): + complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields()) result = assess_controller_handoff(complete, role="author") self.assertEqual(result["verdict"], "complete") result = assess_controller_handoff(self.BASE_HANDOFF, role="author") self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Issue lock proof", result["missing_fields"]) self.assertIn("No review/merge confirmation", result["missing_fields"]) + def test_author_role_requires_issue_lock_proof(self): + without_lock = self.BASE_HANDOFF + "\n" + "\n".join([ + "- Selected issue: #182", + "- Claim/comment status: comment-claimed", + "- PR number opened: #999", + "- No review/merge: confirmed", + ]) + result = assess_controller_handoff(without_lock, role="author") + self.assertEqual(result["verdict"], "incomplete") + self.assertIn("Issue lock proof", result["missing_fields"]) + def test_author_role_rejects_equivalent_or_multiple_issues(self): # 1. equivalent reference blocked incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: Issue #194 / #196 equivalent", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -973,6 +995,7 @@ class TestControllerHandoff(unittest.TestCase): # 2. multiple issues blocked incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #194, #196", + "- Issue lock proof: lock before diff on feat/x @ master", "- Claim/comment status: comment-claimed", "- PR number opened: #999", "- No review/merge: confirmed", @@ -984,6 +1007,7 @@ class TestControllerHandoff(unittest.TestCase): def test_author_role_rejects_fuzzy_pr_number(self): incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([ "- Selected issue: #196", + "- Issue lock proof: lock before diff on feat/issue-196 @ master", "- Claim/comment status: comment-claimed", "- PR number opened: PR #203 / #204 equivalent", "- No review/merge: confirmed", @@ -1015,23 +1039,17 @@ class TestControllerHandoff(unittest.TestCase): def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self): # 1. Workspace mutations: none is rejected when local_edits is True - incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join( + self._author_role_fields(issue_number=196, pr_number=203)) res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True) self.assertEqual(res["verdict"], "incomplete") self.assertIn("Workspace mutations", res["missing_fields"]) # 2. Workspace mutations: edited files is allowed when local_edits is True - complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([ - "- Selected issue: #196", - "- Claim/comment status: comment-claimed", - "- PR number opened: #203", - "- No review/merge: confirmed", - ]) + complete_eq = self.BASE_HANDOFF.replace( + "- Workspace mutations: none", + "- Workspace mutations: edited review_proofs.py", + ) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203)) res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True) self.assertEqual(res2["verdict"], "complete")