Merge pull request 'Fix test env isolation for credentials and create_pr gate tests' (#229) from feat/issue-227-fix-test-env-isolation into master
This commit was merged in pull request #229.
This commit is contained in:
@@ -107,6 +107,24 @@ class TestGetCredentials(unittest.TestCase):
|
||||
class TestGetAuthHeader(unittest.TestCase):
|
||||
"""Test the get_auth_header function."""
|
||||
|
||||
def setUp(self):
|
||||
self._saved_env = os.environ.copy()
|
||||
self._saved_configs = gitea_auth.DYNAMIC_CONFIGS.copy()
|
||||
for key in (
|
||||
"GITEA_MCP_CONFIG",
|
||||
"GITEA_MCP_PROFILE",
|
||||
"GITEA_TOKEN",
|
||||
"GITEA_TOKEN_PRGS",
|
||||
):
|
||||
os.environ.pop(key, None)
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
|
||||
def tearDown(self):
|
||||
os.environ.clear()
|
||||
os.environ.update(self._saved_env)
|
||||
gitea_auth.DYNAMIC_CONFIGS.clear()
|
||||
gitea_auth.DYNAMIC_CONFIGS.update(self._saved_configs)
|
||||
|
||||
@patch("gitea_auth.get_credentials", return_value=("user", "pass"))
|
||||
def test_returns_basic_header(self, _cred):
|
||||
header = gitea_auth.get_auth_header("example.com")
|
||||
|
||||
+46
-18
@@ -54,6 +54,17 @@ ISSUE_WRITE_ENV = {
|
||||
),
|
||||
}
|
||||
|
||||
# create_pr is gated on author profile + namespace (#69, #209).
|
||||
CREATE_PR_ENV = {
|
||||
"GITEA_PROFILE_NAME": "author-test",
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.read,gitea.pr.create,gitea.branch.push"
|
||||
),
|
||||
"GITEA_FORBIDDEN_OPERATIONS": (
|
||||
"gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Create Issue
|
||||
@@ -109,14 +120,16 @@ class TestCreateIssue(unittest.TestCase):
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCreatePR(unittest.TestCase):
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
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):
|
||||
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"}'
|
||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
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)
|
||||
@@ -124,14 +137,17 @@ class TestCreatePR(unittest.TestCase):
|
||||
self.assertEqual(payload["head"], "feat/x")
|
||||
self.assertEqual(payload["base"], "main")
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
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):
|
||||
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"}'
|
||||
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
|
||||
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
|
||||
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"])
|
||||
|
||||
@@ -2796,37 +2812,49 @@ class TestIssueLocking(unittest.TestCase):
|
||||
gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("already tied to an open PR", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_missing_lock_fails(self, _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")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
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")
|
||||
self.assertIn("Issue lock is missing", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_branch_mismatch_fails(self, _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 self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
|
||||
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")
|
||||
self.assertIn("does not match locked branch", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_forbidden_terms_fails(self, _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)
|
||||
for term in ("equivalent to #196", "related to #196", "same as #196"):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("contains forbidden term", str(ctx.exception))
|
||||
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:
|
||||
gitea_create_pr(title=f"feat: X {term}", head="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertIn("contains forbidden term", str(ctx.exception))
|
||||
|
||||
@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)
|
||||
def test_create_pr_missing_closes_ref_fails(self, _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 self.assertRaises(ValueError) as ctx:
|
||||
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
|
||||
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")
|
||||
self.assertIn("must contain 'Closes #196' or 'Fixes #196' exactly", str(ctx.exception))
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user