178 lines
6.5 KiB
Python
178 lines
6.5 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_issue_comment_tool,
|
|
create_issue_tool,
|
|
create_pr_comment_tool,
|
|
update_issue_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]}
|
|
|
|
|
|
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",
|
|
),
|
|
],
|
|
)
|
|
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
|