"""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_FORBIDDEN_OPERATIONS": "gitea.pr.create,gitea.branch.push,gitea.pr.merge", } MERGER_ENV = { "GITEA_PROFILE_NAME": "merger-test", "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.merge", "GITEA_FORBIDDEN_OPERATIONS": "gitea.pr.create,gitea.branch.push,gitea.pr.approve", } 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", # Canonical workflow router (#551) — same skill, multi-runtime names "gitea-workflow", "llm-project-workflow", "git-pr-workflows", ] 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.api_request", return_value={"login": "merger-bot"}) @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) def test_merger_profile_guidance(self, _auth, _api): with patch.dict(os.environ, MERGER_ENV, clear=True): g = mcp_get_control_plane_guide(remote="prgs") self.assertEqual(g["profile"]["role_kind"], "merger") blob = " ".join(g["guidance"]).lower() self.assertIn("merge", 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", "global_worktree", "shell_spawn_hard_stop", "subagent_tool_budget", "subagent_delegation"): 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()