diff --git a/tests/test_issue_lock_recovery.py b/tests/test_issue_lock_recovery.py index ee860b0..66986ec 100644 --- a/tests/test_issue_lock_recovery.py +++ b/tests/test_issue_lock_recovery.py @@ -40,6 +40,12 @@ def _clean_master_git_state_for_lock(): def _lock_record(**overrides): + import issue_lock_provenance + + work_lease = { + "operation_type": "author_issue_work", + "expires_at": "2999-01-01T00:00:00Z", + } record = { "issue_number": 420, "branch_name": BRANCH, @@ -47,10 +53,11 @@ def _lock_record(**overrides): "org": "Scaled-Tech-Consulting", "repo": PRGS_REPO, "worktree_path": os.path.realpath(os.getcwd()), - "work_lease": { - "operation_type": "author_issue_work", - "expires_at": "2999-01-01T00:00:00Z", - }, + "work_lease": work_lease, + "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance( + tool="gitea_lock_issue", + claimant=work_lease.get("claimant"), + ), } record.update(overrides) return record @@ -84,11 +91,12 @@ class TestIssueLockRecovery(unittest.TestCase): ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_lock_recovery_after_restart_adopts_own_branch(self, _auth, mock_api, _git): - mock_api.side_effect = [ - [], - [{"name": BRANCH, "commit": {"id": "934688a"}}], - ] + @patch( + "mcp_server.issue_duplicate_context_fetcher", + return_value=([], [], {"status": "not_claimed"}), + ) + def test_lock_recovery_after_restart_adopts_own_branch(self, _dup_fetcher, _auth, mock_api, _git): + mock_api.return_value = [{"name": BRANCH, "commit": {"id": "934688a"}}] env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir} with patch.dict(os.environ, env, clear=True): with patch("os.getpid", return_value=9999): @@ -141,16 +149,21 @@ class TestIssueLockRecovery(unittest.TestCase): ) @patch("mcp_server.api_get_all") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_open_pr_blocks_recovery_lock(self, _auth, mock_api, _git): - mock_api.side_effect = [ - [{"number": 99, "head": {"ref": BRANCH}, "title": "WIP", "body": ""}], - [], - ] + @patch( + "mcp_server.issue_duplicate_context_fetcher", + return_value=([{ + "number": 99, + "head": {"ref": BRANCH}, + "title": "WIP", + "body": "", + }], [], {"status": "not_claimed"}), + ) + def test_open_pr_blocks_recovery_lock(self, _dup_fetcher, _auth, mock_api, _git): env = {**ISSUE_WRITE_ENV, "GITEA_ISSUE_LOCK_DIR": self.lock_dir} with patch.dict(os.environ, env, clear=True): with self.assertRaises(ValueError) as ctx: gitea_lock_issue(issue_number=420, branch_name=BRANCH, remote="prgs") - self.assertIn("already tied to an open PR", str(ctx.exception)) + self.assertIn("open PR #99 already covers issue", str(ctx.exception)) if __name__ == "__main__": diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 704e0cc..f7fc1dc 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -99,6 +99,9 @@ CREATE_PR_ENV = { ), } +ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json" + + def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides): import issue_lock_provenance @@ -3123,8 +3126,9 @@ class TestIssueLocking(unittest.TestCase): "mcp_server.issue_lock_worktree.read_worktree_git_state", return_value=_clean_master_git_state_for_lock(), ) + @patch("mcp_server.api_get_all", return_value=[]) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - def test_lock_issue_success(self, _auth, _git_state): + def test_lock_issue_success(self, _auth, _api, _git_state): res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs") self.assertTrue(res["success"]) self.assertEqual(res["work_lease"]["operation_type"], "author_issue_work") @@ -3199,10 +3203,7 @@ class TestIssueLocking(unittest.TestCase): @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_lock_issue_adopts_exact_own_branch(self, _auth, mock_api, _git_state): branch = "feat/issue-196-mutations" - mock_api.side_effect = [ - [], - [{"name": branch, "commit": {"id": "abc123"}}], - ] + mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}] res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs") self.assertTrue(res["success"]) self.assertIn("adoption", res) @@ -3428,21 +3429,33 @@ class TestIssueLocking(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_create_pr_manual_lock_seed_blocked(self, _auth, _role): - with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f: - json.dump( + worktree = os.path.realpath(os.getcwd()) + env = self._create_pr_env() + with patch.dict(os.environ, env, clear=True): + issue_lock_store.save_lock_file( + issue_lock_store.lock_file_path( + remote="prgs", + org="Scaled-Tech-Consulting", + repo=mcp_server.REMOTES["prgs"]["repo"], + issue_number=447, + lock_dir=self._lock_dir.name, + ), _sample_issue_lock( issue_number=447, branch_name="feat/issue-447-lock-provenance", + remote="prgs", + org="Scaled-Tech-Consulting", + repo=mcp_server.REMOTES["prgs"]["repo"], + worktree_path=worktree, lock_provenance=None, ), - f, ) - with patch.dict(os.environ, CREATE_PR_ENV, clear=True): with self.assertRaises(RuntimeError) as ctx: gitea_create_pr( title="feat: lock provenance Closes #447", head="feat/issue-447-lock-provenance", remote="prgs", + worktree_path=worktree, ) self.assertIn("lock provenance", str(ctx.exception).lower())