Latte
7837ff43ad
Adds six opt-in write tools (write-mode + policy + per-user permission still
enforced; no destructive or admin actions):
- create_pull_request (POST /pulls)
- create_release / edit_release (POST/PATCH /releases)
- create_branch (POST /branches; create only, no deletion)
- create_milestone (POST /milestones)
- edit_issue_comment (PATCH /issues/comments/{id})
Each: arg schema (extra=forbid, GitRef on branch/ref-like fields), Gitea client
method with url-encoded path segments, handler that surfaces auth errors, MCP
registration (write_operation=True), server wiring, docs, and success tests.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""Tests for expanded read/write MCP tool handlers."""
|
|
|
|
import pytest
|
|
|
|
from aegis_gitea_mcp.config import reset_settings
|
|
from aegis_gitea_mcp.gitea_client import GiteaError
|
|
from aegis_gitea_mcp.tools.read_tools import (
|
|
compare_refs_tool,
|
|
get_commit_diff_tool,
|
|
get_issue_tool,
|
|
get_pull_request_tool,
|
|
list_commits_tool,
|
|
list_issues_tool,
|
|
list_labels_tool,
|
|
list_pull_requests_tool,
|
|
list_releases_tool,
|
|
list_tags_tool,
|
|
search_code_tool,
|
|
)
|
|
from aegis_gitea_mcp.tools.write_tools import (
|
|
add_labels_tool,
|
|
assign_issue_tool,
|
|
create_branch_tool,
|
|
create_issue_comment_tool,
|
|
create_issue_tool,
|
|
create_label_tool,
|
|
create_milestone_tool,
|
|
create_pr_comment_tool,
|
|
create_pull_request_tool,
|
|
create_release_tool,
|
|
edit_issue_comment_tool,
|
|
edit_release_tool,
|
|
remove_labels_tool,
|
|
update_issue_tool,
|
|
update_label_tool,
|
|
)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def tool_env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Provide minimal settings environment for response limit helpers."""
|
|
reset_settings()
|
|
monkeypatch.setenv("GITEA_URL", "https://gitea.example.com")
|
|
monkeypatch.setenv("GITEA_TOKEN", "test-token")
|
|
monkeypatch.setenv("MCP_API_KEYS", "a" * 64)
|
|
monkeypatch.setenv("ENVIRONMENT", "test")
|
|
|
|
|
|
class StubGitea:
|
|
"""Stubbed Gitea client for tool unit tests."""
|
|
|
|
async def search_code(self, owner, repo, query, *, ref, page, limit):
|
|
return {"hits": [{"path": "src/main.py", "snippet": "match text", "score": 1.0}]}
|
|
|
|
async def list_commits(self, owner, repo, *, ref, page, limit):
|
|
return [{"sha": "abc1234", "commit": {"message": "Fix bug", "author": {"date": "now"}}}]
|
|
|
|
async def get_commit_diff(self, owner, repo, sha):
|
|
return {
|
|
"commit": {"message": "Fix bug"},
|
|
"files": [{"filename": "a.py", "status": "modified"}],
|
|
}
|
|
|
|
async def compare_refs(self, owner, repo, base, head):
|
|
return {
|
|
"commits": [{"sha": "abc", "commit": {"message": "Msg"}}],
|
|
"files": [{"filename": "a.py", "status": "modified"}],
|
|
}
|
|
|
|
async def list_issues(self, owner, repo, *, state, page, limit, labels=None):
|
|
return [{"number": 1, "title": "Issue", "state": "open", "labels": []}]
|
|
|
|
async def get_issue(self, owner, repo, index):
|
|
return {"number": index, "title": "Issue", "body": "Body", "state": "open", "labels": []}
|
|
|
|
async def list_pull_requests(self, owner, repo, *, state, page, limit):
|
|
return [{"number": 1, "title": "PR", "state": "open"}]
|
|
|
|
async def get_pull_request(self, owner, repo, index):
|
|
return {"number": index, "title": "PR", "body": "Body", "state": "open"}
|
|
|
|
async def list_labels(self, owner, repo, *, page, limit):
|
|
return [{"id": 1, "name": "bug", "color": "ff0000", "description": "desc"}]
|
|
|
|
async def list_tags(self, owner, repo, *, page, limit):
|
|
return [{"name": "v1.0.0", "commit": {"sha": "abc"}}]
|
|
|
|
async def list_releases(self, owner, repo, *, page, limit):
|
|
return [{"id": 1, "tag_name": "v1.0.0", "name": "release"}]
|
|
|
|
async def create_issue(self, owner, repo, *, title, body, labels=None, assignees=None):
|
|
return {"number": 1, "title": title, "state": "open"}
|
|
|
|
async def update_issue(self, owner, repo, index, *, title=None, body=None, state=None):
|
|
return {"number": index, "title": title or "Issue", "state": state or "open"}
|
|
|
|
async def create_issue_comment(self, owner, repo, index, body):
|
|
return {"id": 1, "body": body}
|
|
|
|
async def create_pr_comment(self, owner, repo, index, body):
|
|
return {"id": 2, "body": body}
|
|
|
|
async def add_labels(self, owner, repo, index, labels):
|
|
return {"labels": [{"name": label} for label in labels]}
|
|
|
|
async def assign_issue(self, owner, repo, index, assignees):
|
|
return {"assignees": [{"login": user} for user in assignees]}
|
|
|
|
async def create_label(self, owner, repo, *, name, color, description="", exclusive=False):
|
|
return {"id": 5, "name": name, "color": color, "description": description, "url": "u"}
|
|
|
|
async def update_label(self, owner, repo, *, name, new_name=None, color=None, description=None):
|
|
return {
|
|
"id": 5,
|
|
"name": new_name or name,
|
|
"color": color or "#ffffff",
|
|
"description": description or "",
|
|
}
|
|
|
|
async def remove_labels(self, owner, repo, index, labels):
|
|
return []
|
|
|
|
async def create_pull_request(self, owner, repo, *, title, head, base, body=""):
|
|
return {"number": 7, "title": title, "state": "open"}
|
|
|
|
async def create_release(
|
|
self, owner, repo, *, tag_name, name="", body="", draft=False, prerelease=False, target=None
|
|
):
|
|
return {"id": 3, "tag_name": tag_name, "name": name or tag_name}
|
|
|
|
async def edit_release(
|
|
self, owner, repo, release_id, *, name=None, body=None, draft=None, prerelease=None
|
|
):
|
|
return {"id": release_id, "tag_name": "v1", "name": name or "rel"}
|
|
|
|
async def create_branch(self, owner, repo, *, new_branch_name, old_branch_name=None):
|
|
return {"name": new_branch_name, "commit": {"id": "abc"}}
|
|
|
|
async def create_milestone(self, owner, repo, *, title, description="", due_on=None):
|
|
return {"id": 4, "title": title, "state": "open"}
|
|
|
|
async def edit_issue_comment(self, owner, repo, comment_id, body):
|
|
return {"id": comment_id, "body": body}
|
|
|
|
|
|
class ErrorGitea(StubGitea):
|
|
"""Stub that raises backend errors for failure-mode coverage."""
|
|
|
|
async def list_commits(self, owner, repo, *, ref, page, limit):
|
|
raise GiteaError("backend failure")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"tool,args,expected_key",
|
|
[
|
|
(search_code_tool, {"owner": "acme", "repo": "app", "query": "foo"}, "results"),
|
|
(list_commits_tool, {"owner": "acme", "repo": "app"}, "commits"),
|
|
(get_commit_diff_tool, {"owner": "acme", "repo": "app", "sha": "abc1234"}, "files"),
|
|
(
|
|
compare_refs_tool,
|
|
{"owner": "acme", "repo": "app", "base": "main", "head": "feature"},
|
|
"commits",
|
|
),
|
|
(list_issues_tool, {"owner": "acme", "repo": "app"}, "issues"),
|
|
(get_issue_tool, {"owner": "acme", "repo": "app", "issue_number": 1}, "title"),
|
|
(list_pull_requests_tool, {"owner": "acme", "repo": "app"}, "pull_requests"),
|
|
(get_pull_request_tool, {"owner": "acme", "repo": "app", "pull_number": 1}, "title"),
|
|
(list_labels_tool, {"owner": "acme", "repo": "app"}, "labels"),
|
|
(list_tags_tool, {"owner": "acme", "repo": "app"}, "tags"),
|
|
(list_releases_tool, {"owner": "acme", "repo": "app"}, "releases"),
|
|
],
|
|
)
|
|
async def test_extended_read_tools_success(tool, args, expected_key):
|
|
"""Each expanded read tool should return expected top-level keys."""
|
|
result = await tool(StubGitea(), args)
|
|
assert expected_key in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extended_read_tools_failure_mode() -> None:
|
|
"""Expanded read tools should wrap backend failures."""
|
|
with pytest.raises(RuntimeError):
|
|
await list_commits_tool(ErrorGitea(), {"owner": "acme", "repo": "app"})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"tool,args,expected_key",
|
|
[
|
|
(create_issue_tool, {"owner": "acme", "repo": "app", "title": "Issue"}, "number"),
|
|
(
|
|
update_issue_tool,
|
|
{"owner": "acme", "repo": "app", "issue_number": 1, "title": "Updated"},
|
|
"number",
|
|
),
|
|
(
|
|
create_issue_comment_tool,
|
|
{"owner": "acme", "repo": "app", "issue_number": 1, "body": "comment"},
|
|
"id",
|
|
),
|
|
(
|
|
create_pr_comment_tool,
|
|
{"owner": "acme", "repo": "app", "pull_number": 1, "body": "comment"},
|
|
"id",
|
|
),
|
|
(
|
|
add_labels_tool,
|
|
{"owner": "acme", "repo": "app", "issue_number": 1, "labels": ["bug"]},
|
|
"labels",
|
|
),
|
|
(
|
|
assign_issue_tool,
|
|
{"owner": "acme", "repo": "app", "issue_number": 1, "assignees": ["alice"]},
|
|
"assignees",
|
|
),
|
|
(
|
|
create_label_tool,
|
|
{"owner": "acme", "repo": "app", "name": "bug", "color": "#ff0000"},
|
|
"id",
|
|
),
|
|
(
|
|
update_label_tool,
|
|
{"owner": "acme", "repo": "app", "name": "bug", "new_name": "defect"},
|
|
"id",
|
|
),
|
|
(
|
|
remove_labels_tool,
|
|
{"owner": "acme", "repo": "app", "issue_number": 1, "labels": ["bug"]},
|
|
"removed",
|
|
),
|
|
(
|
|
create_pull_request_tool,
|
|
{"owner": "acme", "repo": "app", "title": "PR", "head": "feature", "base": "main"},
|
|
"number",
|
|
),
|
|
(
|
|
create_release_tool,
|
|
{"owner": "acme", "repo": "app", "tag_name": "v1.0.0"},
|
|
"id",
|
|
),
|
|
(
|
|
edit_release_tool,
|
|
{"owner": "acme", "repo": "app", "release_id": 3, "name": "x"},
|
|
"id",
|
|
),
|
|
(
|
|
create_branch_tool,
|
|
{"owner": "acme", "repo": "app", "new_branch_name": "feature/x"},
|
|
"name",
|
|
),
|
|
(
|
|
create_milestone_tool,
|
|
{"owner": "acme", "repo": "app", "title": "M1"},
|
|
"id",
|
|
),
|
|
(
|
|
edit_issue_comment_tool,
|
|
{"owner": "acme", "repo": "app", "comment_id": 5, "body": "edited"},
|
|
"id",
|
|
),
|
|
],
|
|
)
|
|
async def test_write_tools_success(tool, args, expected_key):
|
|
"""Write tools should normalize successful backend responses."""
|
|
result = await tool(StubGitea(), args)
|
|
assert expected_key in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_label_normalizes_color_without_hash() -> None:
|
|
"""A hex color without a leading '#' is normalized before hitting Gitea."""
|
|
captured: dict = {}
|
|
|
|
class CaptureStub(StubGitea):
|
|
async def create_label(self, owner, repo, *, name, color, description="", exclusive=False):
|
|
captured["color"] = color
|
|
return {"id": 7, "name": name, "color": color}
|
|
|
|
result = await create_label_tool(
|
|
CaptureStub(),
|
|
{"owner": "acme", "repo": "app", "name": "bug", "color": "ff0000"},
|
|
)
|
|
assert captured["color"] == "#ff0000"
|
|
assert result["id"] == 7
|
|
|
|
|
|
def test_create_label_args_reject_invalid_color() -> None:
|
|
"""Non-hex color values are rejected at the argument layer."""
|
|
import pydantic
|
|
|
|
from aegis_gitea_mcp.tools.arguments import CreateLabelArgs
|
|
|
|
with pytest.raises(pydantic.ValidationError):
|
|
CreateLabelArgs(owner="o", repo="r", name="bug", color="red")
|