sysadminandClaude Opus 4.8
2111c84e7d
Require lease verification (open PRs, branches, worktrees, leases, merged completion) before any issue or PR work. Surface the rule in the portable skill, Gitea runbooks, operator guide, and start-issue template; add doc-contract tests. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
317 lines
14 KiB
Python
317 lines
14 KiB
Python
"""Tests for the operator guide / project skills MCP tools (#128).
|
|
|
|
Read-only capability-discovery tools: mcp_get_control_plane_guide,
|
|
mcp_list_project_skills, mcp_get_skill_guide. Each is tested by calling the
|
|
underlying function directly with mocked API responses.
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
|
|
|
import mcp_server # noqa: E402
|
|
from mcp_server import ( # noqa: E402
|
|
mcp_get_control_plane_guide,
|
|
mcp_list_project_skills,
|
|
mcp_get_skill_guide,
|
|
)
|
|
|
|
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
|
|
|
AUTHOR_ENV = {
|
|
"GITEA_PROFILE_NAME": "author-test",
|
|
"GITEA_ALLOWED_OPERATIONS":
|
|
"gitea.read,gitea.repo.commit,gitea.branch.create,"
|
|
"gitea.branch.push,gitea.pr.create,gitea.pr.comment",
|
|
"GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.approve,gitea.pr.merge",
|
|
}
|
|
|
|
REVIEWER_ENV = {
|
|
"GITEA_PROFILE_NAME": "reviewer-test",
|
|
"GITEA_ALLOWED_OPERATIONS":
|
|
"gitea.read,gitea.pr.review,gitea.pr.comment,gitea.pr.approve,"
|
|
"gitea.pr.request_changes,gitea.pr.merge",
|
|
"GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.create,gitea.branch.push",
|
|
}
|
|
|
|
EXPECTED_SKILLS = [
|
|
"gitea-issue-authoring",
|
|
"gitea-pr-creation",
|
|
"gitea-pr-review",
|
|
"gitea-pr-merge",
|
|
"gitea-issue-comments",
|
|
"gitea-resolve-task-capability",
|
|
"profile-switching",
|
|
"redaction-security-review",
|
|
"jenkins-mcp",
|
|
"glitchtip-mcp",
|
|
"release-operator",
|
|
]
|
|
|
|
|
|
class GuideTestBase(unittest.TestCase):
|
|
def setUp(self):
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
|
|
def tearDown(self):
|
|
mcp_server._IDENTITY_CACHE.clear()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mcp_get_control_plane_guide
|
|
# ---------------------------------------------------------------------------
|
|
class TestControlPlaneGuide(GuideTestBase):
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_author_profile_guidance(self, _auth, _api):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
self.assertTrue(g["read_only"])
|
|
self.assertEqual(g["profile"]["role_kind"], "author")
|
|
self.assertEqual(g["identity"]["authenticated_username"], "author-bot")
|
|
self.assertEqual(g["identity"]["status"], "verified")
|
|
blob = " ".join(g["guidance"]).lower()
|
|
self.assertIn("forbidden", blob)
|
|
self.assertIn("review", blob)
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "reviewer-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_reviewer_profile_guidance(self, _auth, _api):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
self.assertEqual(g["profile"]["role_kind"], "reviewer")
|
|
blob = " ".join(g["guidance"]).lower()
|
|
self.assertIn("eligibility", blob)
|
|
self.assertIn("pinned", blob)
|
|
|
|
@patch("mcp_server.get_auth_header", return_value=None)
|
|
def test_unresolved_identity_instructs_stop(self, _auth):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
self.assertEqual(g["identity"]["status"], "unresolved")
|
|
self.assertIsNone(g["identity"]["authenticated_username"])
|
|
self.assertIn("STOP", g["identity"]["instruction"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_guide_is_read_only(self, _auth, mock_api):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
mcp_get_control_plane_guide(remote="prgs")
|
|
for call in mock_api.call_args_list:
|
|
self.assertEqual(call[0][0], "GET")
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_no_urls_or_keychain_ids_by_default(self, _auth, _api):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
blob = json.dumps(g)
|
|
self.assertNotIn("https://", blob)
|
|
self.assertNotIn("http://", blob)
|
|
self.assertNotIn("keychain:", blob)
|
|
self.assertNotIn(FAKE_AUTH, blob)
|
|
self.assertNotIn("server", g["identity"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_reveal_opt_in_includes_server(self, _auth, _api):
|
|
env = dict(AUTHOR_ENV, GITEA_MCP_REVEAL_ENDPOINTS="1")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
self.assertIn("gitea.prgs.cc", g["identity"]["server"])
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_rules_cover_required_topics(self, _auth, _api):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
rules = g["rules"]
|
|
for key in ("hard_stops", "fail_closed", "head_sha_pinning",
|
|
"merge_confirmation", "redaction", "separation",
|
|
"profile_switching", "identity_verification",
|
|
"work_selection"):
|
|
self.assertIn(key, rules)
|
|
self.assertIn("MERGE PR", json.dumps(rules["merge_confirmation"]))
|
|
self.assertTrue(rules["hard_stops"])
|
|
|
|
def test_unknown_remote_raises(self):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
with self.assertRaises(ValueError):
|
|
mcp_get_control_plane_guide(remote="nope")
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "new-session-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_new_session_can_call_guide_for_operating_model(self, _auth, _api):
|
|
"""Covers #129 AC: New LLM sessions can call one guide tool to understand the MCP Control Plane operating model."""
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
self.assertTrue(g["read_only"])
|
|
self.assertIn("profile", g)
|
|
self.assertIn("identity", g)
|
|
self.assertIn("guidance", g)
|
|
self.assertIn("rules", g)
|
|
self.assertIn("workflows", g)
|
|
self.assertEqual(g["skills_tool"], "mcp_list_project_skills")
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_guide_mentions_resolve_capability_and_issue_comment_separation(self, _auth, _api):
|
|
"""Covers #144 AC: guide tells to resolve task cap, issue comments require gitea.issue.comment (not implied by pr.comment)."""
|
|
env = dict(AUTHOR_ENV, GITEA_ALLOWED_OPERATIONS="gitea.read,gitea.pr.comment,gitea.pr.create")
|
|
with patch.dict(os.environ, env, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
blob = " ".join(g["guidance"]).lower()
|
|
self.assertIn("resolve", blob)
|
|
self.assertIn("gitea.issue.comment", blob)
|
|
self.assertIn("pr.comment", blob) # separation noted
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_guide_static_profile_safe_next_action(self, _auth, _api):
|
|
"""Covers #144 AC: static-profile safe next action guidance."""
|
|
# default is static (no allow_runtime_switching)
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
# since author, guidance includes author info; resolver advice always present
|
|
blob = json.dumps(g).lower()
|
|
self.assertTrue("static" in blob or "relaunch" in blob or "namespace" in blob)
|
|
self.assertIn("resolve", " ".join(g["guidance"]).lower())
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "no-comment-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_missing_issue_comment_guidance(self, _auth, _api):
|
|
"""Covers #144 AC: Guide lists issue comments require gitea.issue.comment and PR comments do not imply them."""
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
hard_stops = g["rules"]["hard_stops"]
|
|
self.assertTrue(any("gitea.issue.comment" in r for r in hard_stops))
|
|
self.assertTrue(any("gitea.pr.comment" in r for r in hard_stops))
|
|
|
|
@patch("mcp_server.api_request", return_value={"login": "static-bot"})
|
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
|
def test_static_profile_safe_next_action(self, _auth, _api):
|
|
"""Covers #144 AC: Guide covers static-profile safe next action switching instruction."""
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
g = mcp_get_control_plane_guide(remote="prgs")
|
|
switching_rules = g["rules"]["profile_switching"]
|
|
self.assertTrue(any("static-profile mode" in r for r in switching_rules))
|
|
self.assertTrue(any("GITEA_MCP_PROFILE" in r for r in switching_rules))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mcp_list_project_skills
|
|
# ---------------------------------------------------------------------------
|
|
class TestProjectSkills(GuideTestBase):
|
|
|
|
@patch("mcp_server.api_request")
|
|
def test_registry_complete_and_no_api_calls(self, mock_api):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
r = mcp_list_project_skills()
|
|
self.assertTrue(r["read_only"])
|
|
names = [s["name"] for s in r["skills"]]
|
|
for expected in EXPECTED_SKILLS:
|
|
self.assertIn(expected, names)
|
|
self.assertEqual(r["count"], len(r["skills"]))
|
|
for s in r["skills"]:
|
|
self.assertTrue(s["description"])
|
|
self.assertTrue(s["when_to_use"])
|
|
self.assertIn("required_operations", s)
|
|
self.assertIn("status", s)
|
|
self.assertIn("available_to_current_profile", s)
|
|
mock_api.assert_not_called()
|
|
|
|
def test_profile_aware_availability(self):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
r = mcp_list_project_skills()
|
|
by_name = {s["name"]: s for s in r["skills"]}
|
|
self.assertTrue(by_name["gitea-pr-creation"]["available_to_current_profile"])
|
|
self.assertFalse(by_name["gitea-pr-merge"]["available_to_current_profile"])
|
|
|
|
def test_unimplemented_services_marked(self):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
r = mcp_list_project_skills()
|
|
by_name = {s["name"]: s for s in r["skills"]}
|
|
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
|
|
self.assertEqual(by_name["glitchtip-mcp"]["status"], "external-mcp")
|
|
|
|
def test_no_urls_in_registry(self):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
r = mcp_list_project_skills()
|
|
blob = json.dumps(r)
|
|
self.assertNotIn("https://", blob)
|
|
self.assertNotIn("http://", blob)
|
|
self.assertNotIn("keychain:", blob)
|
|
|
|
def test_enabled_but_no_usable_tools_negative_assertion(self):
|
|
"""Negative assertion for 'enabled but no usable tools' (per issue #151)."""
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
r = mcp_list_project_skills()
|
|
by_name = {s["name"]: s for s in r["skills"]}
|
|
# jenkins-mcp is an external boundary; Gitea profile discovery must not
|
|
# treat a local config entry as a usable Jenkins tool surface.
|
|
self.assertIn("jenkins-mcp", by_name)
|
|
self.assertEqual(by_name["jenkins-mcp"]["status"], "external-mcp")
|
|
self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False))
|
|
|
|
def test_external_mcp_skill_guides_name_expected_tools(self):
|
|
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
|
jenkins = mcp_get_skill_guide("jenkins-mcp")
|
|
glitchtip = mcp_get_skill_guide("glitchtip-mcp")
|
|
self.assertEqual(jenkins["skill"]["status"], "external-mcp")
|
|
self.assertIn("jenkins_whoami", " ".join(jenkins["steps"]))
|
|
self.assertIn("jenkins_get_build", " ".join(jenkins["steps"]))
|
|
self.assertIn("SKIPPED", " ".join(jenkins["steps"]))
|
|
self.assertEqual(glitchtip["skill"]["status"], "external-mcp")
|
|
self.assertIn("glitchtip_whoami", " ".join(glitchtip["steps"]))
|
|
self.assertIn("glitchtip_search", " ".join(glitchtip["steps"]))
|
|
self.assertIn("SKIPPED", " ".join(glitchtip["steps"]))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# mcp_get_skill_guide
|
|
# ---------------------------------------------------------------------------
|
|
class TestSkillGuide(GuideTestBase):
|
|
|
|
def test_known_skill_returns_steps(self):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
r = mcp_get_skill_guide("gitea-pr-merge")
|
|
self.assertTrue(r["success"])
|
|
self.assertEqual(r["skill"]["name"], "gitea-pr-merge")
|
|
self.assertTrue(r["steps"])
|
|
self.assertIn("MERGE PR", " ".join(r["steps"]))
|
|
|
|
def test_case_and_whitespace_normalized(self):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
r = mcp_get_skill_guide(" Gitea-PR-Merge ")
|
|
self.assertTrue(r["success"])
|
|
|
|
def test_unknown_skill_fails_closed(self):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
r = mcp_get_skill_guide("no-such-skill")
|
|
self.assertFalse(r["success"])
|
|
self.assertTrue(r["reasons"])
|
|
for expected in EXPECTED_SKILLS:
|
|
self.assertIn(expected, r["valid_skills"])
|
|
|
|
@patch("mcp_server.api_request")
|
|
def test_read_only_no_api_calls(self, mock_api):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
mcp_get_skill_guide("gitea-pr-review")
|
|
mock_api.assert_not_called()
|
|
|
|
def test_no_urls_in_skill_guides(self):
|
|
with patch.dict(os.environ, REVIEWER_ENV, clear=True):
|
|
for name in EXPECTED_SKILLS:
|
|
r = mcp_get_skill_guide(name)
|
|
blob = json.dumps(r)
|
|
self.assertNotIn("https://", blob)
|
|
self.assertNotIn("keychain:", blob)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|