"""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_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): 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"), ( 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 @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")