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

resolve conflicts for PR #464

Browse Source
This commit is contained in:
sysadmin
2026-07-07 17:25:13 -04:00
parent 3ad7547a3f
commit 2cace4f7cd
1 changed files with 67 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mcp_server.py
+67 -26
View File
@@ -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")
# ---------------------------------------------------------------------------