diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index e672030..20dbb64 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -6,6 +6,7 @@ the MCP protocol) with mocked API responses. import json import os import sys +import tempfile import unittest from unittest.mock import patch, MagicMock @@ -45,6 +46,7 @@ from gitea_auth import get_profile # noqa: E402 import gitea_config # noqa: E402 import mcp_server +import issue_lock_store FAKE_AUTH = "Basic dGVzdDp0ZXN0" @@ -131,6 +133,30 @@ def _clear_duplicate_context_fetcher(*_args, **_kwargs): return [], [], {"status": "not_claimed"} +def _bind_test_lock(**overrides) -> str: + remote = overrides.get("remote", "dadeschools") + record = _sample_issue_lock(**overrides) + if remote in mcp_server.REMOTES: + profile = mcp_server.REMOTES[remote] + record.setdefault("org", profile["org"]) + record.setdefault("repo", profile["repo"]) + record["remote"] = remote + worktree = record.get("worktree_path") or os.path.realpath(os.getcwd()) + acquisition = issue_lock_store.acquire_issue_lock( + issue_number=int(record["issue_number"]), + branch_name=str(record["branch_name"]), + remote=remote, + org=str(record["org"]), + repo=str(record["repo"]), + worktree_path=worktree, + work_lease=record["work_lease"], + claimant=record["work_lease"].get("claimant"), + lock_provenance=record.get("lock_provenance"), + lock_dir=os.environ.get("GITEA_ISSUE_LOCK_DIR"), + ) + return acquisition["lock_path"] + + # --------------------------------------------------------------------------- # Create Issue # --------------------------------------------------------------------------- @@ -193,18 +219,21 @@ class TestCreatePR(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - @patch("os.path.exists", return_value=True) - @patch("builtins.open") - def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role, _dup_fetcher): - 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 + def test_creates_pr(self, _auth, mock_api, _role, _dup_fetcher): + worktree = os.path.realpath(os.getcwd()) 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") + with tempfile.TemporaryDirectory() as lock_dir: + env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir} + with patch.dict(os.environ, env, clear=True): + _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree) + result = gitea_create_pr( + title="feat: X Closes #123", + head="feat/x", + base="main", + worktree_path=worktree, + ) 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") @@ -218,30 +247,42 @@ class TestCreatePR(unittest.TestCase): return_value=(True, [])) @patch("mcp_server.api_request") @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) - @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, _dup_fetcher): - 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 + def test_create_pr_reveal_opt_in_includes_url(self, _auth, mock_api, _role, _dup_fetcher): + worktree = os.path.realpath(os.getcwd()) 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") + with tempfile.TemporaryDirectory() as lock_dir: + env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir, "GITEA_MCP_REVEAL_ENDPOINTS": "1"} + with patch.dict(os.environ, env, clear=True): + _bind_test_lock(issue_number=123, branch_name="feat/x", worktree_path=worktree) + result = gitea_create_pr( + title="feat: X Closes #123", + head="feat/x", + base="main", + worktree_path=worktree, + ) 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") + def test_create_pr_locked_issue_mismatch_fails(self, _auth, _role): + worktree = os.path.realpath(os.getcwd()) + with tempfile.TemporaryDirectory() as lock_dir: + env = {**CREATE_PR_ENV, "GITEA_ISSUE_LOCK_DIR": lock_dir} + with patch.dict(os.environ, env, clear=True): + _bind_test_lock( + issue_number=123, + branch_name="feat/x", + worktree_path=worktree, + ) + with self.assertRaises(ValueError) as ctx: + gitea_create_pr( + title="feat: X Closes #999", + head="feat/x", + base="main", + worktree_path=worktree, + ) self.assertIn("Closes #123", str(ctx.exception)) - mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8") # ---------------------------------------------------------------------------