Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 74 Pull Requests 9 Actions Packages Projects Releases Wiki Activity
Files
683649424b5b6e474007a900847b21579f1835b8
Gitea-Tools/tests/test_operator_guide.py
T
sysadmin 57b2023611 fix: control-plane guide org/repo params and local repo alignment (#588) 
#530's remote/repo guard blocked mcp_get_control_plane_guide(remote=prgs)
because the bare default resolves to Timesheet while local git is
Gitea-Tools, and remediation told callers to pass org/repo on a tool that
did not accept those parameters.

- Accept optional org/repo on mcp_get_control_plane_guide
- Prefer local git remote org/repo when bare defaults mismatch (read-only guide)
- Keep mutation tools fail-closed on bare mismatch
- Align remediation text with tools that accept org/repo

Closes #588
2026-07-09 13:17:27 -04:00

354 lines
16 KiB
Python
Raw Blame History

"""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)
self.assertIn("resolved_repository", g)
@patch("mcp_server.api_request", return_value={"login": "author-bot"})
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_guide_explicit_org_repo_parameters(self, _auth, _api):
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
g = mcp_get_control_plane_guide(
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertEqual(
g["resolved_repository"],
{"org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools"},
)
@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()
Reference in New Issue View Git Blame Copy Permalink