"""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", "profile-switching", "redaction-security-review", "jenkins-readonly", "glitchtip-readonly", "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"): 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") # --------------------------------------------------------------------------- # 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.assertNotEqual(by_name["jenkins-readonly"]["status"], "available") self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available") 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) # --------------------------------------------------------------------------- # 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()