Files
AegisGitea-MCP/tests/test_tools_extended.py
T
Latte 41749fd7b4 fix: harden get_issue parsing and surface real errors (#27); align CI image publish
get_issue raised 'NoneType' object is not iterable on issues whose
labels/assignees Gitea returns as null or with non-dict elements (the #13
class), which reached clients as an opaque JSON-RPC -32603 with no detail.

- read_tools: skip non-dict label/assignee entries in get_issue_tool
- server: detect a wrapped GiteaNotFoundError via the __cause__ chain and
  return 404 / JSON-RPC -32000 with a clear message; include the exception
  type name in masked internal errors so future masked failures are
  diagnosable without exposing messages or stack traces
- tests: cover non-dict collection elements and the not-found / typed-error
  responses
- ci: rewrite docker.yml to build, smoke-test and push the image to the
  Gitea container registry on merge to main/dev, matching the hiddenden.cafe
  pattern (only REGISTRY_TOKEN required)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 16:51:58 +02:00

543 lines
20 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 GiteaAuthenticationError, GiteaError
from aegis_gitea_mcp.tools.read_tools import (
compare_refs_tool,
get_branch_tool,
get_commit_diff_tool,
get_commit_status_tool,
get_issue_tool,
get_latest_release_tool,
get_pull_request_tool,
get_release_tool,
get_repo_languages_tool,
list_branches_tool,
list_commits_tool,
list_issue_comments_tool,
list_issues_tool,
list_labels_tool,
list_milestones_tool,
list_org_repositories_tool,
list_organizations_tool,
list_pull_request_commits_tool,
list_pull_request_files_tool,
list_pull_requests_tool,
list_releases_tool,
list_repo_topics_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 list_pull_request_files(self, owner, repo, index, *, page, limit):
return [{"filename": "a.py", "status": "modified", "additions": 1, "deletions": 0}]
async def list_pull_request_commits(self, owner, repo, index, *, page, limit):
return [{"sha": "abc", "commit": {"message": "m"}, "author": {"login": "alice"}}]
async def list_issue_comments(self, owner, repo, index, *, page, limit):
return [{"id": 1, "body": "hi", "user": {"login": "alice"}}]
async def list_branches(self, owner, repo, *, page, limit):
return [{"name": "main", "protected": True, "commit": {"id": "abc"}}]
async def get_branch(self, owner, repo, branch):
return {"name": branch, "protected": False, "commit": {"id": "abc"}}
async def get_release(self, owner, repo, release_id):
return {"id": release_id, "tag_name": "v1", "name": "rel"}
async def get_latest_release(self, owner, repo):
return {"id": 1, "tag_name": "v1", "name": "rel"}
async def list_milestones(self, owner, repo, *, state, page, limit):
return [{"id": 1, "title": "M", "state": state}]
async def get_commit_status(self, owner, repo, sha):
return {"state": "success", "statuses": [{"context": "ci", "status": "success"}]}
async def list_org_repositories(self, org, *, page, limit):
return [{"name": "r", "owner": {"login": org}, "full_name": f"{org}/r"}]
async def list_organizations(self, *, page, limit):
return [{"id": 1, "username": "acme", "description": "d"}]
async def get_repo_languages(self, owner, repo):
return {"Python": 100, "HTML": 5}
async def list_repo_topics(self, owner, repo):
return ["python", "mcp"]
async def create_issue(
self, owner, repo, *, title, body, labels=None, assignees=None, milestone=None
):
result = {"number": 1, "title": title, "state": "open"}
if milestone is not None:
result["milestone"] = {"id": 4, "title": str(milestone)}
return result
async def update_issue(
self, owner, repo, index, *, title=None, body=None, state=None, milestone=None
):
result = {"number": index, "title": title or "Issue", "state": state or "open"}
if milestone is not None:
result["milestone"] = {"id": 4, "title": str(milestone)}
return result
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"),
(
list_pull_request_files_tool,
{"owner": "acme", "repo": "app", "pull_number": 1},
"files",
),
(
list_pull_request_commits_tool,
{"owner": "acme", "repo": "app", "pull_number": 1},
"commits",
),
(
list_issue_comments_tool,
{"owner": "acme", "repo": "app", "issue_number": 1},
"comments",
),
(list_branches_tool, {"owner": "acme", "repo": "app"}, "branches"),
(get_branch_tool, {"owner": "acme", "repo": "app", "branch": "main"}, "name"),
(get_release_tool, {"owner": "acme", "repo": "app", "release_id": 1}, "tag_name"),
(get_latest_release_tool, {"owner": "acme", "repo": "app"}, "tag_name"),
(list_milestones_tool, {"owner": "acme", "repo": "app"}, "milestones"),
(
get_commit_status_tool,
{"owner": "acme", "repo": "app", "sha": "abc1234"},
"state",
),
(list_org_repositories_tool, {"org": "acme"}, "repositories"),
(list_organizations_tool, {}, "organizations"),
(get_repo_languages_tool, {"owner": "acme", "repo": "app"}, "languages"),
(list_repo_topics_tool, {"owner": "acme", "repo": "app"}, "topics"),
],
)
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
async def test_get_issue_tolerates_null_collections() -> None:
"""Regression for #13: Gitea may return null for labels/assignees/user.
`.get(key, [])` returns None when the key is present with a null value, so
iterating the result raised `'NoneType' object is not iterable`.
"""
class NullFieldsGitea(StubGitea):
async def get_issue(self, owner, repo, index):
return {
"number": index,
"title": "Issue",
"body": "Body",
"state": "open",
"user": None,
"labels": None,
"assignees": None,
}
result = await get_issue_tool(
NullFieldsGitea(), {"owner": "acme", "repo": "app", "issue_number": 1}
)
assert result["author"] == ""
assert result["labels"] == []
assert result["assignees"] == []
@pytest.mark.asyncio
async def test_get_issue_skips_non_dict_collection_elements() -> None:
"""Defense-in-depth for #27: tolerate non-dict entries inside labels/assignees.
A stray null/non-object element would otherwise raise AttributeError when
`.get()` is called on it, surfacing as an opaque internal error.
"""
class MalformedElementsGitea(StubGitea):
async def get_issue(self, owner, repo, index):
return {
"number": index,
"title": "Issue",
"body": "Body",
"state": "open",
"user": {"login": "alice"},
"labels": [{"name": "bug"}, None, "weird"],
"assignees": [{"login": "bob"}, None, 42],
}
result = await get_issue_tool(
MalformedElementsGitea(), {"owner": "acme", "repo": "app", "issue_number": 1}
)
assert result["labels"] == ["bug"]
assert result["assignees"] == ["bob"]
@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")
@pytest.mark.asyncio
async def test_create_issue_returns_assigned_milestone_title() -> None:
"""create_issue surfaces the assigned milestone title in its response."""
result = await create_issue_tool(
StubGitea(),
{"owner": "acme", "repo": "app", "title": "Issue", "milestone": "Sprint 1"},
)
assert result["milestone"] == "Sprint 1"
@pytest.mark.asyncio
async def test_update_issue_accepts_milestone_only() -> None:
"""update_issue may change only the milestone (no title/body/state needed)."""
result = await update_issue_tool(
StubGitea(),
{"owner": "acme", "repo": "app", "issue_number": 1, "milestone": 4},
)
assert result["milestone"] == "4"
def test_update_issue_args_require_a_changed_field() -> None:
"""An update with no mutable field (incl. milestone) is rejected."""
import pydantic
from aegis_gitea_mcp.tools.arguments import UpdateIssueArgs
with pytest.raises(pydantic.ValidationError):
UpdateIssueArgs(owner="o", repo="r", issue_number=1)
# Supplying only a milestone satisfies the change requirement.
assert UpdateIssueArgs(owner="o", repo="r", issue_number=1, milestone=0).milestone == 0
def test_issue_args_reject_boolean_milestone() -> None:
"""A boolean is rejected as a milestone reference (it subclasses int)."""
import pydantic
from aegis_gitea_mcp.tools.arguments import CreateIssueArgs
with pytest.raises(pydantic.ValidationError):
CreateIssueArgs(owner="o", repo="r", title="x", milestone=True)
# (tool, valid_args) for every write tool, used to exercise error branches.
WRITE_TOOL_ERROR_CASES = [
(create_issue_tool, {"owner": "acme", "repo": "app", "title": "Issue"}),
(update_issue_tool, {"owner": "acme", "repo": "app", "issue_number": 1, "title": "x"}),
(create_issue_comment_tool, {"owner": "acme", "repo": "app", "issue_number": 1, "body": "c"}),
(create_pr_comment_tool, {"owner": "acme", "repo": "app", "pull_number": 1, "body": "c"}),
(add_labels_tool, {"owner": "acme", "repo": "app", "issue_number": 1, "labels": ["bug"]}),
(assign_issue_tool, {"owner": "acme", "repo": "app", "issue_number": 1, "assignees": ["al"]}),
(create_label_tool, {"owner": "acme", "repo": "app", "name": "bug", "color": "#ff0000"}),
(update_label_tool, {"owner": "acme", "repo": "app", "name": "bug", "new_name": "defect"}),
(remove_labels_tool, {"owner": "acme", "repo": "app", "issue_number": 1, "labels": ["bug"]}),
(
create_pull_request_tool,
{"owner": "acme", "repo": "app", "title": "PR", "head": "feature", "base": "main"},
),
(create_release_tool, {"owner": "acme", "repo": "app", "tag_name": "v1.0.0"}),
(edit_release_tool, {"owner": "acme", "repo": "app", "release_id": 3, "name": "x"}),
(create_branch_tool, {"owner": "acme", "repo": "app", "new_branch_name": "feature/x"}),
(create_milestone_tool, {"owner": "acme", "repo": "app", "title": "M1"}),
(edit_issue_comment_tool, {"owner": "acme", "repo": "app", "comment_id": 5, "body": "e"}),
]
class _WriteBackendErrorGitea:
"""Stub whose every method raises a generic Gitea backend error."""
def __getattr__(self, name: str):
async def _raise(*args: object, **kwargs: object) -> object:
raise GiteaError("backend failure")
return _raise
class _WriteAuthErrorGitea:
"""Stub whose every method raises an authentication error."""
def __getattr__(self, name: str):
async def _raise(*args: object, **kwargs: object) -> object:
raise GiteaAuthenticationError("token expired")
return _raise
@pytest.mark.asyncio
@pytest.mark.parametrize("tool,args", WRITE_TOOL_ERROR_CASES)
async def test_write_tools_wrap_backend_errors(tool, args) -> None:
"""Every write tool wraps a backend GiteaError as a RuntimeError."""
with pytest.raises(RuntimeError):
await tool(_WriteBackendErrorGitea(), args)
@pytest.mark.asyncio
@pytest.mark.parametrize("tool,args", WRITE_TOOL_ERROR_CASES)
async def test_write_tools_propagate_auth_errors(tool, args) -> None:
"""Every write tool lets auth failures surface for re-authorization."""
with pytest.raises(GiteaAuthenticationError):
await tool(_WriteAuthErrorGitea(), args)