generated from personal-projects/leo-claude-mktplace
Merge feat/9: Remove incorrect standalone MCP implementation
This commit is contained in:
@@ -1,3 +0,0 @@
|
||||
"""Gitea MCP Server - MCP server for Gitea API integration."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -1,51 +0,0 @@
|
||||
"""Authentication and configuration management for Gitea MCP server."""
|
||||
|
||||
import os
|
||||
from typing import Optional
|
||||
from dotenv import load_dotenv
|
||||
|
||||
|
||||
class AuthConfig:
|
||||
"""Manages authentication configuration for Gitea API."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize authentication configuration from environment variables."""
|
||||
load_dotenv()
|
||||
|
||||
self.api_url: Optional[str] = os.getenv("GITEA_API_URL")
|
||||
self.api_token: Optional[str] = os.getenv("GITEA_API_TOKEN")
|
||||
|
||||
self._validate()
|
||||
|
||||
def _validate(self) -> None:
|
||||
"""Validate that required configuration is present.
|
||||
|
||||
Raises:
|
||||
ValueError: If required environment variables are missing.
|
||||
"""
|
||||
if not self.api_url:
|
||||
raise ValueError(
|
||||
"GITEA_API_URL environment variable is required. "
|
||||
"Please set it in your .env file or environment."
|
||||
)
|
||||
|
||||
if not self.api_token:
|
||||
raise ValueError(
|
||||
"GITEA_API_TOKEN environment variable is required. "
|
||||
"Please set it in your .env file or environment."
|
||||
)
|
||||
|
||||
# Remove trailing slash from URL if present
|
||||
if self.api_url.endswith("/"):
|
||||
self.api_url = self.api_url[:-1]
|
||||
|
||||
def get_auth_headers(self) -> dict[str, str]:
|
||||
"""Get authentication headers for API requests.
|
||||
|
||||
Returns:
|
||||
dict: HTTP headers with authorization token.
|
||||
"""
|
||||
return {
|
||||
"Authorization": f"token {self.api_token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
@@ -1,216 +0,0 @@
|
||||
"""HTTP client for Gitea API."""
|
||||
|
||||
import httpx
|
||||
from typing import Any, Optional
|
||||
from .auth import AuthConfig
|
||||
|
||||
|
||||
class GiteaClientError(Exception):
|
||||
"""Base exception for Gitea client errors."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class GiteaAuthError(GiteaClientError):
|
||||
"""Authentication error."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class GiteaNotFoundError(GiteaClientError):
|
||||
"""Resource not found error."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class GiteaServerError(GiteaClientError):
|
||||
"""Server error."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Async HTTP client for Gitea API."""
|
||||
|
||||
def __init__(self, config: AuthConfig, timeout: float = 30.0):
|
||||
"""Initialize Gitea API client.
|
||||
|
||||
Args:
|
||||
config: Authentication configuration.
|
||||
timeout: Request timeout in seconds (default: 30.0).
|
||||
"""
|
||||
self.config = config
|
||||
self.timeout = timeout
|
||||
self._client: Optional[httpx.AsyncClient] = None
|
||||
|
||||
async def __aenter__(self):
|
||||
"""Async context manager entry."""
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=self.config.api_url,
|
||||
headers=self.config.get_auth_headers(),
|
||||
timeout=self.timeout,
|
||||
)
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
"""Async context manager exit."""
|
||||
if self._client:
|
||||
await self._client.aclose()
|
||||
|
||||
def _handle_error(self, response: httpx.Response) -> None:
|
||||
"""Handle HTTP error responses.
|
||||
|
||||
Args:
|
||||
response: HTTP response object.
|
||||
|
||||
Raises:
|
||||
GiteaAuthError: For 401/403 errors.
|
||||
GiteaNotFoundError: For 404 errors.
|
||||
GiteaServerError: For 500+ errors.
|
||||
GiteaClientError: For other errors.
|
||||
"""
|
||||
status = response.status_code
|
||||
|
||||
if status == 401:
|
||||
raise GiteaAuthError(
|
||||
"Authentication failed. Please check your GITEA_API_TOKEN."
|
||||
)
|
||||
elif status == 403:
|
||||
raise GiteaAuthError(
|
||||
"Access forbidden. Your API token may not have required permissions."
|
||||
)
|
||||
elif status == 404:
|
||||
raise GiteaNotFoundError(
|
||||
f"Resource not found: {response.request.url}"
|
||||
)
|
||||
elif status >= 500:
|
||||
raise GiteaServerError(
|
||||
f"Gitea server error (HTTP {status}): {response.text}"
|
||||
)
|
||||
else:
|
||||
raise GiteaClientError(
|
||||
f"API request failed (HTTP {status}): {response.text}"
|
||||
)
|
||||
|
||||
async def get(self, path: str, **kwargs) -> dict[str, Any]:
|
||||
"""Make GET request to Gitea API.
|
||||
|
||||
Args:
|
||||
path: API endpoint path (e.g., "/api/v1/repos/owner/repo").
|
||||
**kwargs: Additional arguments for httpx request.
|
||||
|
||||
Returns:
|
||||
dict: JSON response data.
|
||||
|
||||
Raises:
|
||||
GiteaClientError: If request fails.
|
||||
"""
|
||||
if not self._client:
|
||||
raise GiteaClientError(
|
||||
"Client not initialized. Use 'async with' context manager."
|
||||
)
|
||||
|
||||
try:
|
||||
response = await self._client.get(path, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except httpx.HTTPStatusError:
|
||||
self._handle_error(response)
|
||||
except httpx.RequestError as e:
|
||||
raise GiteaClientError(
|
||||
f"Request failed: {e}"
|
||||
) from e
|
||||
|
||||
async def post(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
|
||||
"""Make POST request to Gitea API.
|
||||
|
||||
Args:
|
||||
path: API endpoint path.
|
||||
json: JSON data to send in request body.
|
||||
**kwargs: Additional arguments for httpx request.
|
||||
|
||||
Returns:
|
||||
dict: JSON response data.
|
||||
|
||||
Raises:
|
||||
GiteaClientError: If request fails.
|
||||
"""
|
||||
if not self._client:
|
||||
raise GiteaClientError(
|
||||
"Client not initialized. Use 'async with' context manager."
|
||||
)
|
||||
|
||||
try:
|
||||
response = await self._client.post(path, json=json, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except httpx.HTTPStatusError:
|
||||
self._handle_error(response)
|
||||
except httpx.RequestError as e:
|
||||
raise GiteaClientError(
|
||||
f"Request failed: {e}"
|
||||
) from e
|
||||
|
||||
async def patch(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
|
||||
"""Make PATCH request to Gitea API.
|
||||
|
||||
Args:
|
||||
path: API endpoint path.
|
||||
json: JSON data to send in request body.
|
||||
**kwargs: Additional arguments for httpx request.
|
||||
|
||||
Returns:
|
||||
dict: JSON response data.
|
||||
|
||||
Raises:
|
||||
GiteaClientError: If request fails.
|
||||
"""
|
||||
if not self._client:
|
||||
raise GiteaClientError(
|
||||
"Client not initialized. Use 'async with' context manager."
|
||||
)
|
||||
|
||||
try:
|
||||
response = await self._client.patch(path, json=json, **kwargs)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
except httpx.HTTPStatusError:
|
||||
self._handle_error(response)
|
||||
except httpx.RequestError as e:
|
||||
raise GiteaClientError(
|
||||
f"Request failed: {e}"
|
||||
) from e
|
||||
|
||||
async def delete(self, path: str, **kwargs) -> Optional[dict[str, Any]]:
|
||||
"""Make DELETE request to Gitea API.
|
||||
|
||||
Args:
|
||||
path: API endpoint path.
|
||||
**kwargs: Additional arguments for httpx request.
|
||||
|
||||
Returns:
|
||||
dict or None: JSON response data if available, None for 204 responses.
|
||||
|
||||
Raises:
|
||||
GiteaClientError: If request fails.
|
||||
"""
|
||||
if not self._client:
|
||||
raise GiteaClientError(
|
||||
"Client not initialized. Use 'async with' context manager."
|
||||
)
|
||||
|
||||
try:
|
||||
response = await self._client.delete(path, **kwargs)
|
||||
response.raise_for_status()
|
||||
|
||||
# DELETE requests may return 204 No Content
|
||||
if response.status_code == 204:
|
||||
return None
|
||||
|
||||
return response.json()
|
||||
except httpx.HTTPStatusError:
|
||||
self._handle_error(response)
|
||||
except httpx.RequestError as e:
|
||||
raise GiteaClientError(
|
||||
f"Request failed: {e}"
|
||||
) from e
|
||||
@@ -1,173 +0,0 @@
|
||||
"""MCP server implementation for Gitea API integration."""
|
||||
|
||||
import asyncio
|
||||
import argparse
|
||||
from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
from . import __version__
|
||||
from .auth import AuthConfig
|
||||
from .client import GiteaClient, GiteaClientError
|
||||
from .tools import (
|
||||
get_issue_tools,
|
||||
handle_issue_tool,
|
||||
get_label_tools,
|
||||
handle_label_tool,
|
||||
get_milestone_tools,
|
||||
handle_milestone_tool,
|
||||
)
|
||||
|
||||
|
||||
# Global client instance
|
||||
gitea_client: GiteaClient | None = None
|
||||
|
||||
|
||||
async def serve() -> None:
|
||||
"""Run the MCP server."""
|
||||
server = Server("gitea-mcp")
|
||||
|
||||
# Initialize authentication config
|
||||
try:
|
||||
config = AuthConfig()
|
||||
except ValueError as e:
|
||||
print(f"Configuration error: {e}")
|
||||
raise
|
||||
|
||||
# Initialize Gitea client
|
||||
global gitea_client
|
||||
gitea_client = GiteaClient(config)
|
||||
|
||||
@server.list_tools()
|
||||
async def list_tools() -> list[Tool]:
|
||||
"""List available MCP tools.
|
||||
|
||||
Returns:
|
||||
list: Available tools including issue, label, and milestone operations.
|
||||
"""
|
||||
# Get issue, label, and milestone tools
|
||||
tools = get_issue_tools()
|
||||
tools.extend(get_label_tools())
|
||||
tools.extend(get_milestone_tools())
|
||||
|
||||
# Placeholder for future tools (PR tools, etc.)
|
||||
tools.extend([
|
||||
Tool(
|
||||
name="list_repositories",
|
||||
description="List repositories in an organization (coming soon)",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"org": {
|
||||
"type": "string",
|
||||
"description": "Organization name",
|
||||
}
|
||||
},
|
||||
"required": ["org"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="create_pull_request",
|
||||
description="Create a new pull request (coming soon)",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Pull request title",
|
||||
},
|
||||
"head": {
|
||||
"type": "string",
|
||||
"description": "Source branch",
|
||||
},
|
||||
"base": {
|
||||
"type": "string",
|
||||
"description": "Target branch",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "title", "head", "base"],
|
||||
},
|
||||
),
|
||||
])
|
||||
|
||||
return tools
|
||||
|
||||
@server.call_tool()
|
||||
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
||||
"""Handle tool calls.
|
||||
|
||||
Args:
|
||||
name: Tool name.
|
||||
arguments: Tool arguments.
|
||||
|
||||
Returns:
|
||||
list: Tool response.
|
||||
"""
|
||||
# Handle issue tools
|
||||
if name.startswith("gitea_") and any(
|
||||
name.endswith(suffix) for suffix in ["_issues", "_issue"]
|
||||
):
|
||||
return await handle_issue_tool(name, arguments, gitea_client)
|
||||
|
||||
# Handle label tools
|
||||
if name.startswith("gitea_") and any(
|
||||
name.endswith(suffix) for suffix in ["_labels", "_label"]
|
||||
):
|
||||
return await handle_label_tool(gitea_client, name, arguments)
|
||||
|
||||
# Handle milestone tools
|
||||
if name.startswith("gitea_") and any(
|
||||
name.endswith(suffix) for suffix in ["_milestones", "_milestone"]
|
||||
):
|
||||
return await handle_milestone_tool(name, arguments, gitea_client)
|
||||
|
||||
# Placeholder for other tools
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Tool '{name}' is not yet implemented.",
|
||||
)
|
||||
]
|
||||
|
||||
# Run the server using stdio transport
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await server.run(
|
||||
read_stream,
|
||||
write_stream,
|
||||
server.create_initialization_options(),
|
||||
)
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Main entry point with CLI argument parsing."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Gitea MCP Server - MCP server for Gitea API integration"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--version",
|
||||
action="version",
|
||||
version=f"gitea-mcp {__version__}",
|
||||
)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
# Run the server
|
||||
try:
|
||||
asyncio.run(serve())
|
||||
except KeyboardInterrupt:
|
||||
print("\nServer stopped by user")
|
||||
except Exception as e:
|
||||
print(f"Server error: {e}")
|
||||
raise
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,14 +0,0 @@
|
||||
"""Gitea MCP tools package."""
|
||||
|
||||
from .issues import get_issue_tools, handle_issue_tool
|
||||
from .labels import get_label_tools, handle_label_tool
|
||||
from .milestones import get_milestone_tools, handle_milestone_tool
|
||||
|
||||
__all__ = [
|
||||
"get_issue_tools",
|
||||
"handle_issue_tool",
|
||||
"get_label_tools",
|
||||
"handle_label_tool",
|
||||
"get_milestone_tools",
|
||||
"handle_milestone_tool",
|
||||
]
|
||||
@@ -1,392 +0,0 @@
|
||||
"""Gitea issue operations tools for MCP server."""
|
||||
|
||||
from typing import Any, Optional
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
from ..client import GiteaClient, GiteaClientError
|
||||
|
||||
|
||||
def get_issue_tools() -> list[Tool]:
|
||||
"""Get list of issue operation tools.
|
||||
|
||||
Returns:
|
||||
list[Tool]: List of MCP tools for issue operations.
|
||||
"""
|
||||
return [
|
||||
Tool(
|
||||
name="gitea_list_issues",
|
||||
description="List issues in a Gitea repository with optional filters",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "Filter by state: open, closed, or all",
|
||||
"enum": ["open", "closed", "all"],
|
||||
"default": "open",
|
||||
},
|
||||
"labels": {
|
||||
"type": "string",
|
||||
"description": "Comma-separated list of label names to filter by",
|
||||
},
|
||||
"milestone": {
|
||||
"type": "string",
|
||||
"description": "Milestone name to filter by",
|
||||
},
|
||||
"page": {
|
||||
"type": "integer",
|
||||
"description": "Page number for pagination (default: 1)",
|
||||
"default": 1,
|
||||
},
|
||||
"limit": {
|
||||
"type": "integer",
|
||||
"description": "Number of issues per page (default: 30)",
|
||||
"default": 30,
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="gitea_get_issue",
|
||||
description="Get details of a specific issue by number",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"index": {
|
||||
"type": "integer",
|
||||
"description": "Issue number/index",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "index"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="gitea_create_issue",
|
||||
description="Create a new issue in a Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Issue title",
|
||||
},
|
||||
"body": {
|
||||
"type": "string",
|
||||
"description": "Issue body/description",
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
"description": "Array of label IDs to assign",
|
||||
},
|
||||
"milestone": {
|
||||
"type": "integer",
|
||||
"description": "Milestone ID to assign",
|
||||
},
|
||||
"assignees": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Array of usernames to assign",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "title"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="gitea_update_issue",
|
||||
description="Update an existing issue in a Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"index": {
|
||||
"type": "integer",
|
||||
"description": "Issue number/index",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "New issue title",
|
||||
},
|
||||
"body": {
|
||||
"type": "string",
|
||||
"description": "New issue body/description",
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "Issue state: open or closed",
|
||||
"enum": ["open", "closed"],
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "integer"},
|
||||
"description": "Array of label IDs to assign (replaces existing)",
|
||||
},
|
||||
"milestone": {
|
||||
"type": "integer",
|
||||
"description": "Milestone ID to assign",
|
||||
},
|
||||
"assignees": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Array of usernames to assign (replaces existing)",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "index"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def handle_issue_tool(
|
||||
name: str, arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Handle issue tool calls.
|
||||
|
||||
Args:
|
||||
name: Tool name.
|
||||
arguments: Tool arguments.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Tool response.
|
||||
"""
|
||||
try:
|
||||
if name == "gitea_list_issues":
|
||||
return await _list_issues(arguments, client)
|
||||
elif name == "gitea_get_issue":
|
||||
return await _get_issue(arguments, client)
|
||||
elif name == "gitea_create_issue":
|
||||
return await _create_issue(arguments, client)
|
||||
elif name == "gitea_update_issue":
|
||||
return await _update_issue(arguments, client)
|
||||
else:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Unknown issue tool: {name}",
|
||||
)
|
||||
]
|
||||
except GiteaClientError as e:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Error: {str(e)}",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
async def _list_issues(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""List issues in a repository.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, and optional filters.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: List of issues.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
state = arguments.get("state", "open")
|
||||
labels = arguments.get("labels")
|
||||
milestone = arguments.get("milestone")
|
||||
page = arguments.get("page", 1)
|
||||
limit = arguments.get("limit", 30)
|
||||
|
||||
params = {
|
||||
"state": state,
|
||||
"page": page,
|
||||
"limit": limit,
|
||||
}
|
||||
|
||||
if labels:
|
||||
params["labels"] = labels
|
||||
|
||||
if milestone:
|
||||
params["milestone"] = milestone
|
||||
|
||||
async with client:
|
||||
issues = await client.get(f"/repos/{owner}/{repo}/issues", params=params)
|
||||
|
||||
# Format response
|
||||
if not issues:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"No {state} issues found in {owner}/{repo}",
|
||||
)
|
||||
]
|
||||
|
||||
result = f"Found {len(issues)} {state} issue(s) in {owner}/{repo}:\n\n"
|
||||
for issue in issues:
|
||||
result += f"#{issue['number']} - {issue['title']}\n"
|
||||
result += f" State: {issue['state']}\n"
|
||||
if issue.get('labels'):
|
||||
labels_str = ", ".join([label['name'] for label in issue['labels']])
|
||||
result += f" Labels: {labels_str}\n"
|
||||
if issue.get('milestone'):
|
||||
result += f" Milestone: {issue['milestone']['title']}\n"
|
||||
result += f" Created: {issue['created_at']}\n"
|
||||
result += f" Updated: {issue['updated_at']}\n\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
|
||||
|
||||
async def _get_issue(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Get a specific issue by number.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, and index.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Issue details.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
index = arguments["index"]
|
||||
|
||||
async with client:
|
||||
issue = await client.get(f"/repos/{owner}/{repo}/issues/{index}")
|
||||
|
||||
# Format response
|
||||
result = f"Issue #{issue['number']}: {issue['title']}\n\n"
|
||||
result += f"State: {issue['state']}\n"
|
||||
result += f"Created: {issue['created_at']}\n"
|
||||
result += f"Updated: {issue['updated_at']}\n"
|
||||
|
||||
if issue.get('labels'):
|
||||
labels_str = ", ".join([label['name'] for label in issue['labels']])
|
||||
result += f"Labels: {labels_str}\n"
|
||||
|
||||
if issue.get('milestone'):
|
||||
result += f"Milestone: {issue['milestone']['title']}\n"
|
||||
|
||||
if issue.get('assignees'):
|
||||
assignees_str = ", ".join([user['login'] for user in issue['assignees']])
|
||||
result += f"Assignees: {assignees_str}\n"
|
||||
|
||||
result += f"\nBody:\n{issue.get('body', '(no description)')}\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
|
||||
|
||||
async def _create_issue(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Create a new issue.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, title, and optional fields.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Created issue details.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
|
||||
data = {
|
||||
"title": arguments["title"],
|
||||
}
|
||||
|
||||
if "body" in arguments:
|
||||
data["body"] = arguments["body"]
|
||||
|
||||
if "labels" in arguments:
|
||||
data["labels"] = arguments["labels"]
|
||||
|
||||
if "milestone" in arguments:
|
||||
data["milestone"] = arguments["milestone"]
|
||||
|
||||
if "assignees" in arguments:
|
||||
data["assignees"] = arguments["assignees"]
|
||||
|
||||
async with client:
|
||||
issue = await client.post(f"/repos/{owner}/{repo}/issues", json=data)
|
||||
|
||||
result = f"Created issue #{issue['number']}: {issue['title']}\n"
|
||||
result += f"URL: {issue.get('html_url', 'N/A')}\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
|
||||
|
||||
async def _update_issue(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Update an existing issue.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, index, and fields to update.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Updated issue details.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
index = arguments["index"]
|
||||
|
||||
data = {}
|
||||
|
||||
if "title" in arguments:
|
||||
data["title"] = arguments["title"]
|
||||
|
||||
if "body" in arguments:
|
||||
data["body"] = arguments["body"]
|
||||
|
||||
if "state" in arguments:
|
||||
data["state"] = arguments["state"]
|
||||
|
||||
if "labels" in arguments:
|
||||
data["labels"] = arguments["labels"]
|
||||
|
||||
if "milestone" in arguments:
|
||||
data["milestone"] = arguments["milestone"]
|
||||
|
||||
if "assignees" in arguments:
|
||||
data["assignees"] = arguments["assignees"]
|
||||
|
||||
async with client:
|
||||
issue = await client.patch(f"/repos/{owner}/{repo}/issues/{index}", json=data)
|
||||
|
||||
result = f"Updated issue #{issue['number']}: {issue['title']}\n"
|
||||
result += f"State: {issue['state']}\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
@@ -1,170 +0,0 @@
|
||||
"""Gitea label operations MCP tools."""
|
||||
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
from ..client import GiteaClient, GiteaClientError
|
||||
|
||||
|
||||
def get_label_tools() -> list[Tool]:
|
||||
"""Get label operation tool definitions.
|
||||
|
||||
Returns:
|
||||
list: Tool definitions for label operations.
|
||||
"""
|
||||
return [
|
||||
Tool(
|
||||
name="gitea_list_labels",
|
||||
description="List all labels in a Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (user or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="gitea_create_label",
|
||||
description="Create a new label in a Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (user or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "Label name",
|
||||
},
|
||||
"color": {
|
||||
"type": "string",
|
||||
"description": "Label color (hex without #, e.g., 'ff0000' for red)",
|
||||
},
|
||||
"description": {
|
||||
"type": "string",
|
||||
"description": "Label description (optional)",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "name", "color"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def handle_label_tool(
|
||||
client: GiteaClient, name: str, arguments: dict
|
||||
) -> list[TextContent]:
|
||||
"""Handle label tool execution.
|
||||
|
||||
Args:
|
||||
client: Gitea client instance.
|
||||
name: Tool name.
|
||||
arguments: Tool arguments.
|
||||
|
||||
Returns:
|
||||
list: Tool response content.
|
||||
"""
|
||||
try:
|
||||
async with client:
|
||||
if name == "gitea_list_labels":
|
||||
return await _list_labels(client, arguments)
|
||||
elif name == "gitea_create_label":
|
||||
return await _create_label(client, arguments)
|
||||
else:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Unknown label tool: {name}",
|
||||
)
|
||||
]
|
||||
except GiteaClientError as e:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Gitea API error: {e}",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
async def _list_labels(client: GiteaClient, arguments: dict) -> list[TextContent]:
|
||||
"""List labels in a repository.
|
||||
|
||||
Args:
|
||||
client: Gitea client instance.
|
||||
arguments: Tool arguments with owner and repo.
|
||||
|
||||
Returns:
|
||||
list: Label listing response.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
|
||||
labels = await client.get(f"/repos/{owner}/{repo}/labels")
|
||||
|
||||
if not labels:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"No labels found in {owner}/{repo}",
|
||||
)
|
||||
]
|
||||
|
||||
# Format labels for display
|
||||
lines = [f"Labels in {owner}/{repo}:", ""]
|
||||
for label in labels:
|
||||
color = label.get("color", "")
|
||||
desc = label.get("description", "")
|
||||
desc_text = f" - {desc}" if desc else ""
|
||||
lines.append(f" • {label['name']} (#{color}){desc_text}")
|
||||
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text="\n".join(lines),
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
async def _create_label(client: GiteaClient, arguments: dict) -> list[TextContent]:
|
||||
"""Create a new label.
|
||||
|
||||
Args:
|
||||
client: Gitea client instance.
|
||||
arguments: Tool arguments with owner, repo, name, color, and optional description.
|
||||
|
||||
Returns:
|
||||
list: Creation response.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
name = arguments["name"]
|
||||
color = arguments["color"].lstrip("#") # Remove # if present
|
||||
description = arguments.get("description", "")
|
||||
|
||||
payload = {
|
||||
"name": name,
|
||||
"color": color,
|
||||
}
|
||||
if description:
|
||||
payload["description"] = description
|
||||
|
||||
label = await client.post(f"/repos/{owner}/{repo}/labels", payload)
|
||||
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Created label '{label['name']}' (#{label['color']}) in {owner}/{repo}",
|
||||
)
|
||||
]
|
||||
@@ -1,190 +0,0 @@
|
||||
"""Gitea milestone operations tools for MCP server."""
|
||||
|
||||
from typing import Any, Optional
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
from ..client import GiteaClient, GiteaClientError
|
||||
|
||||
|
||||
def get_milestone_tools() -> list[Tool]:
|
||||
"""Get list of milestone operation tools.
|
||||
|
||||
Returns:
|
||||
list[Tool]: List of MCP tools for milestone operations.
|
||||
"""
|
||||
return [
|
||||
Tool(
|
||||
name="gitea_list_milestones",
|
||||
description="List milestones in a Gitea repository with optional state filter",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"description": "Filter by state: open, closed, or all (default: open)",
|
||||
"enum": ["open", "closed", "all"],
|
||||
"default": "open",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="gitea_create_milestone",
|
||||
description="Create a new milestone in a Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {
|
||||
"type": "string",
|
||||
"description": "Repository owner (username or organization)",
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Milestone title",
|
||||
},
|
||||
"description": {
|
||||
"type": "string",
|
||||
"description": "Milestone description (optional)",
|
||||
},
|
||||
"due_on": {
|
||||
"type": "string",
|
||||
"description": "Due date in ISO 8601 format, e.g., '2024-12-31T23:59:59Z' (optional)",
|
||||
},
|
||||
},
|
||||
"required": ["owner", "repo", "title"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
async def handle_milestone_tool(
|
||||
name: str, arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Handle milestone tool calls.
|
||||
|
||||
Args:
|
||||
name: Tool name.
|
||||
arguments: Tool arguments.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Tool response.
|
||||
"""
|
||||
try:
|
||||
if name == "gitea_list_milestones":
|
||||
return await _list_milestones(arguments, client)
|
||||
elif name == "gitea_create_milestone":
|
||||
return await _create_milestone(arguments, client)
|
||||
else:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Unknown milestone tool: {name}",
|
||||
)
|
||||
]
|
||||
except GiteaClientError as e:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"Error: {str(e)}",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
async def _list_milestones(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""List milestones in a repository.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, and optional state filter.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: List of milestones.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
state = arguments.get("state", "open")
|
||||
|
||||
params = {"state": state}
|
||||
|
||||
async with client:
|
||||
milestones = await client.get(
|
||||
f"/repos/{owner}/{repo}/milestones", params=params
|
||||
)
|
||||
|
||||
# Format response
|
||||
if not milestones:
|
||||
return [
|
||||
TextContent(
|
||||
type="text",
|
||||
text=f"No {state} milestones found in {owner}/{repo}",
|
||||
)
|
||||
]
|
||||
|
||||
result = f"Found {len(milestones)} {state} milestone(s) in {owner}/{repo}:\n\n"
|
||||
for milestone in milestones:
|
||||
result += f"{milestone.get('title', 'Untitled')}\n"
|
||||
result += f" State: {milestone.get('state', 'unknown')}\n"
|
||||
if milestone.get("description"):
|
||||
result += f" Description: {milestone['description']}\n"
|
||||
if milestone.get("due_on"):
|
||||
result += f" Due: {milestone['due_on']}\n"
|
||||
result += f" Open Issues: {milestone.get('open_issues', 0)}\n"
|
||||
result += f" Closed Issues: {milestone.get('closed_issues', 0)}\n"
|
||||
result += f" Created: {milestone.get('created_at', 'N/A')}\n\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
|
||||
|
||||
async def _create_milestone(
|
||||
arguments: dict[str, Any], client: GiteaClient
|
||||
) -> list[TextContent]:
|
||||
"""Create a new milestone.
|
||||
|
||||
Args:
|
||||
arguments: Tool arguments containing owner, repo, title, and optional fields.
|
||||
client: Gitea API client instance.
|
||||
|
||||
Returns:
|
||||
list[TextContent]: Created milestone details.
|
||||
"""
|
||||
owner = arguments["owner"]
|
||||
repo = arguments["repo"]
|
||||
|
||||
data = {
|
||||
"title": arguments["title"],
|
||||
}
|
||||
|
||||
if "description" in arguments:
|
||||
data["description"] = arguments["description"]
|
||||
|
||||
if "due_on" in arguments:
|
||||
data["due_on"] = arguments["due_on"]
|
||||
|
||||
async with client:
|
||||
milestone = await client.post(
|
||||
f"/repos/{owner}/{repo}/milestones", json=data
|
||||
)
|
||||
|
||||
result = f"Created milestone: {milestone['title']}\n"
|
||||
if milestone.get("description"):
|
||||
result += f"Description: {milestone['description']}\n"
|
||||
if milestone.get("due_on"):
|
||||
result += f"Due: {milestone['due_on']}\n"
|
||||
|
||||
return [TextContent(type="text", text=result)]
|
||||
@@ -1 +0,0 @@
|
||||
"""Tests for Gitea MCP server."""
|
||||
@@ -1,103 +0,0 @@
|
||||
"""Shared pytest fixtures for Gitea MCP tests."""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from gitea_mcp.auth import AuthConfig
|
||||
from gitea_mcp.client import GiteaClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config(monkeypatch):
|
||||
"""Create mock authentication config.
|
||||
|
||||
This fixture sets up test environment variables and returns
|
||||
a configured AuthConfig instance for testing.
|
||||
"""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
return AuthConfig()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_client(mock_config):
|
||||
"""Create a mock GiteaClient instance.
|
||||
|
||||
Returns a GiteaClient with mocked HTTP methods that don't make
|
||||
real API calls. Use this for testing tool handlers.
|
||||
"""
|
||||
client = GiteaClient(mock_config, timeout=10.0)
|
||||
|
||||
# Mock the internal HTTP client methods
|
||||
client.get = AsyncMock()
|
||||
client.post = AsyncMock()
|
||||
client.patch = AsyncMock()
|
||||
client.delete = AsyncMock()
|
||||
|
||||
# Mock context manager
|
||||
client.__aenter__ = AsyncMock(return_value=client)
|
||||
client.__aexit__ = AsyncMock(return_value=None)
|
||||
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_issue():
|
||||
"""Sample issue data for testing.
|
||||
|
||||
Returns a dict representing a typical Gitea issue response.
|
||||
"""
|
||||
return {
|
||||
"id": 1,
|
||||
"number": 42,
|
||||
"title": "Test Issue",
|
||||
"body": "This is a test issue",
|
||||
"state": "open",
|
||||
"created_at": "2024-01-15T10:00:00Z",
|
||||
"updated_at": "2024-01-15T12:00:00Z",
|
||||
"html_url": "http://gitea.example.com/test/repo/issues/42",
|
||||
"labels": [
|
||||
{"id": 1, "name": "bug", "color": "ff0000"},
|
||||
{"id": 2, "name": "priority-high", "color": "ff9900"},
|
||||
],
|
||||
"milestone": {
|
||||
"id": 10,
|
||||
"title": "v1.0",
|
||||
"state": "open",
|
||||
},
|
||||
"assignees": [
|
||||
{"id": 100, "login": "testuser"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_label():
|
||||
"""Sample label data for testing.
|
||||
|
||||
Returns a dict representing a typical Gitea label response.
|
||||
"""
|
||||
return {
|
||||
"id": 1,
|
||||
"name": "bug",
|
||||
"color": "ff0000",
|
||||
"description": "Something isn't working",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def sample_milestone():
|
||||
"""Sample milestone data for testing.
|
||||
|
||||
Returns a dict representing a typical Gitea milestone response.
|
||||
"""
|
||||
return {
|
||||
"id": 10,
|
||||
"title": "v1.0",
|
||||
"description": "First major release",
|
||||
"state": "open",
|
||||
"open_issues": 5,
|
||||
"closed_issues": 15,
|
||||
"created_at": "2024-01-01T00:00:00Z",
|
||||
"updated_at": "2024-01-15T12:00:00Z",
|
||||
"due_on": "2024-12-31T23:59:59Z",
|
||||
}
|
||||
@@ -1,56 +0,0 @@
|
||||
"""Tests for authentication module."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from gitea_mcp.auth import AuthConfig
|
||||
|
||||
|
||||
def test_auth_config_success(monkeypatch):
|
||||
"""Test successful authentication configuration."""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
|
||||
config = AuthConfig()
|
||||
|
||||
assert config.api_url == "http://gitea.example.com/api/v1"
|
||||
assert config.api_token == "test_token_123"
|
||||
|
||||
|
||||
def test_auth_config_removes_trailing_slash(monkeypatch):
|
||||
"""Test that trailing slash is removed from URL."""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1/")
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
|
||||
config = AuthConfig()
|
||||
|
||||
assert config.api_url == "http://gitea.example.com/api/v1"
|
||||
|
||||
|
||||
def test_auth_config_missing_url(monkeypatch):
|
||||
"""Test error when GITEA_API_URL is missing."""
|
||||
monkeypatch.delenv("GITEA_API_URL", raising=False)
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
|
||||
with pytest.raises(ValueError, match="GITEA_API_URL"):
|
||||
AuthConfig()
|
||||
|
||||
|
||||
def test_auth_config_missing_token(monkeypatch):
|
||||
"""Test error when GITEA_API_TOKEN is missing."""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
||||
monkeypatch.delenv("GITEA_API_TOKEN", raising=False)
|
||||
|
||||
with pytest.raises(ValueError, match="GITEA_API_TOKEN"):
|
||||
AuthConfig()
|
||||
|
||||
|
||||
def test_get_auth_headers(monkeypatch):
|
||||
"""Test authentication headers generation."""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
|
||||
config = AuthConfig()
|
||||
headers = config.get_auth_headers()
|
||||
|
||||
assert headers["Authorization"] == "token test_token_123"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
@@ -1,93 +0,0 @@
|
||||
"""Tests for Gitea API client."""
|
||||
|
||||
import pytest
|
||||
import httpx
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
from gitea_mcp.auth import AuthConfig
|
||||
from gitea_mcp.client import (
|
||||
GiteaClient,
|
||||
GiteaClientError,
|
||||
GiteaAuthError,
|
||||
GiteaNotFoundError,
|
||||
GiteaServerError,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config(monkeypatch):
|
||||
"""Create mock authentication config."""
|
||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
||||
return AuthConfig()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_initialization(mock_config):
|
||||
"""Test client initialization."""
|
||||
client = GiteaClient(mock_config, timeout=10.0)
|
||||
assert client.config == mock_config
|
||||
assert client.timeout == 10.0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_context_manager(mock_config):
|
||||
"""Test client as async context manager."""
|
||||
async with GiteaClient(mock_config) as client:
|
||||
assert client._client is not None
|
||||
assert isinstance(client._client, httpx.AsyncClient)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_client_without_context_manager_raises_error(mock_config):
|
||||
"""Test that using client without context manager raises error."""
|
||||
client = GiteaClient(mock_config)
|
||||
|
||||
with pytest.raises(GiteaClientError, match="not initialized"):
|
||||
await client.get("/test")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_error_401(mock_config):
|
||||
"""Test handling 401 authentication error."""
|
||||
response = MagicMock()
|
||||
response.status_code = 401
|
||||
|
||||
async with GiteaClient(mock_config) as client:
|
||||
with pytest.raises(GiteaAuthError, match="Authentication failed"):
|
||||
client._handle_error(response)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_error_403(mock_config):
|
||||
"""Test handling 403 forbidden error."""
|
||||
response = MagicMock()
|
||||
response.status_code = 403
|
||||
|
||||
async with GiteaClient(mock_config) as client:
|
||||
with pytest.raises(GiteaAuthError, match="Access forbidden"):
|
||||
client._handle_error(response)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_error_404(mock_config):
|
||||
"""Test handling 404 not found error."""
|
||||
response = MagicMock()
|
||||
response.status_code = 404
|
||||
response.request = MagicMock()
|
||||
response.request.url = "http://gitea.example.com/api/v1/test"
|
||||
|
||||
async with GiteaClient(mock_config) as client:
|
||||
with pytest.raises(GiteaNotFoundError, match="not found"):
|
||||
client._handle_error(response)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_error_500(mock_config):
|
||||
"""Test handling 500 server error."""
|
||||
response = MagicMock()
|
||||
response.status_code = 500
|
||||
response.text = "Internal Server Error"
|
||||
|
||||
async with GiteaClient(mock_config) as client:
|
||||
with pytest.raises(GiteaServerError, match="server error"):
|
||||
client._handle_error(response)
|
||||
@@ -1,372 +0,0 @@
|
||||
"""Tests for issue operations tools."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import Tool, TextContent
|
||||
from gitea_mcp.tools.issues import (
|
||||
get_issue_tools,
|
||||
handle_issue_tool,
|
||||
_list_issues,
|
||||
_get_issue,
|
||||
_create_issue,
|
||||
_update_issue,
|
||||
)
|
||||
from gitea_mcp.client import GiteaClientError
|
||||
|
||||
|
||||
class TestIssueToolDefinitions:
|
||||
"""Test issue tool schema definitions."""
|
||||
|
||||
def test_get_issue_tools_returns_list(self):
|
||||
"""Test that get_issue_tools returns a list of Tool objects."""
|
||||
tools = get_issue_tools()
|
||||
assert isinstance(tools, list)
|
||||
assert len(tools) == 4
|
||||
assert all(isinstance(tool, Tool) for tool in tools)
|
||||
|
||||
def test_list_issues_tool_schema(self):
|
||||
"""Test gitea_list_issues tool has correct schema."""
|
||||
tools = get_issue_tools()
|
||||
list_tool = next(t for t in tools if t.name == "gitea_list_issues")
|
||||
|
||||
assert list_tool.name == "gitea_list_issues"
|
||||
assert "list issues" in list_tool.description.lower()
|
||||
|
||||
schema = list_tool.inputSchema
|
||||
assert schema["type"] == "object"
|
||||
assert set(schema["required"]) == {"owner", "repo"}
|
||||
assert "owner" in schema["properties"]
|
||||
assert "repo" in schema["properties"]
|
||||
assert "state" in schema["properties"]
|
||||
assert "labels" in schema["properties"]
|
||||
assert "milestone" in schema["properties"]
|
||||
|
||||
def test_get_issue_tool_schema(self):
|
||||
"""Test gitea_get_issue tool has correct schema."""
|
||||
tools = get_issue_tools()
|
||||
get_tool = next(t for t in tools if t.name == "gitea_get_issue")
|
||||
|
||||
assert get_tool.name == "gitea_get_issue"
|
||||
assert "get details" in get_tool.description.lower()
|
||||
|
||||
schema = get_tool.inputSchema
|
||||
assert set(schema["required"]) == {"owner", "repo", "index"}
|
||||
assert "index" in schema["properties"]
|
||||
|
||||
def test_create_issue_tool_schema(self):
|
||||
"""Test gitea_create_issue tool has correct schema."""
|
||||
tools = get_issue_tools()
|
||||
create_tool = next(t for t in tools if t.name == "gitea_create_issue")
|
||||
|
||||
assert create_tool.name == "gitea_create_issue"
|
||||
assert "create" in create_tool.description.lower()
|
||||
|
||||
schema = create_tool.inputSchema
|
||||
assert set(schema["required"]) == {"owner", "repo", "title"}
|
||||
assert "body" in schema["properties"]
|
||||
assert "labels" in schema["properties"]
|
||||
assert "assignees" in schema["properties"]
|
||||
|
||||
def test_update_issue_tool_schema(self):
|
||||
"""Test gitea_update_issue tool has correct schema."""
|
||||
tools = get_issue_tools()
|
||||
update_tool = next(t for t in tools if t.name == "gitea_update_issue")
|
||||
|
||||
assert update_tool.name == "gitea_update_issue"
|
||||
assert "update" in update_tool.description.lower()
|
||||
|
||||
schema = update_tool.inputSchema
|
||||
assert set(schema["required"]) == {"owner", "repo", "index"}
|
||||
assert "title" in schema["properties"]
|
||||
assert "state" in schema["properties"]
|
||||
|
||||
|
||||
class TestListIssues:
|
||||
"""Test _list_issues function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_issues_success(self, mock_client, sample_issue):
|
||||
"""Test successful issue listing."""
|
||||
mock_client.get.return_value = [sample_issue]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "open",
|
||||
}
|
||||
|
||||
result = await _list_issues(arguments, mock_client)
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert "Test Issue" in result[0].text
|
||||
assert "#42" in result[0].text
|
||||
assert "bug" in result[0].text
|
||||
|
||||
mock_client.get.assert_called_once()
|
||||
call_args = mock_client.get.call_args
|
||||
assert "/repos/testowner/testrepo/issues" in call_args[0][0]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_issues_empty(self, mock_client):
|
||||
"""Test listing when no issues found."""
|
||||
mock_client.get.return_value = []
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "closed",
|
||||
}
|
||||
|
||||
result = await _list_issues(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "No closed issues found" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_issues_with_filters(self, mock_client, sample_issue):
|
||||
"""Test listing with label and milestone filters."""
|
||||
mock_client.get.return_value = [sample_issue]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "all",
|
||||
"labels": "bug,priority-high",
|
||||
"milestone": "v1.0",
|
||||
"page": 2,
|
||||
"limit": 50,
|
||||
}
|
||||
|
||||
result = await _list_issues(arguments, mock_client)
|
||||
|
||||
mock_client.get.assert_called_once()
|
||||
call_args = mock_client.get.call_args
|
||||
params = call_args[1]["params"]
|
||||
assert params["labels"] == "bug,priority-high"
|
||||
assert params["milestone"] == "v1.0"
|
||||
assert params["page"] == 2
|
||||
assert params["limit"] == 50
|
||||
|
||||
|
||||
class TestGetIssue:
|
||||
"""Test _get_issue function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_issue_success(self, mock_client, sample_issue):
|
||||
"""Test successful issue retrieval."""
|
||||
mock_client.get.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"index": 42,
|
||||
}
|
||||
|
||||
result = await _get_issue(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Issue #42" in result[0].text
|
||||
assert "Test Issue" in result[0].text
|
||||
assert "This is a test issue" in result[0].text
|
||||
assert "bug" in result[0].text
|
||||
assert "v1.0" in result[0].text
|
||||
assert "testuser" in result[0].text
|
||||
|
||||
mock_client.get.assert_called_once()
|
||||
assert "/repos/testowner/testrepo/issues/42" in mock_client.get.call_args[0][0]
|
||||
|
||||
|
||||
class TestCreateIssue:
|
||||
"""Test _create_issue function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_minimal(self, mock_client, sample_issue):
|
||||
"""Test creating issue with minimal required fields."""
|
||||
mock_client.post.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "New Issue",
|
||||
}
|
||||
|
||||
result = await _create_issue(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created issue #42" in result[0].text
|
||||
assert "Test Issue" in result[0].text
|
||||
|
||||
mock_client.post.assert_called_once()
|
||||
call_args = mock_client.post.call_args
|
||||
assert "/repos/testowner/testrepo/issues" in call_args[0][0]
|
||||
assert call_args[1]["json"]["title"] == "New Issue"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_full(self, mock_client, sample_issue):
|
||||
"""Test creating issue with all optional fields."""
|
||||
mock_client.post.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "New Issue",
|
||||
"body": "Detailed description",
|
||||
"labels": [1, 2],
|
||||
"milestone": 10,
|
||||
"assignees": ["user1", "user2"],
|
||||
}
|
||||
|
||||
result = await _create_issue(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
json_data = call_args[1]["json"]
|
||||
assert json_data["title"] == "New Issue"
|
||||
assert json_data["body"] == "Detailed description"
|
||||
assert json_data["labels"] == [1, 2]
|
||||
assert json_data["milestone"] == 10
|
||||
assert json_data["assignees"] == ["user1", "user2"]
|
||||
|
||||
|
||||
class TestUpdateIssue:
|
||||
"""Test _update_issue function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_issue_title(self, mock_client, sample_issue):
|
||||
"""Test updating issue title."""
|
||||
updated_issue = sample_issue.copy()
|
||||
updated_issue["title"] = "Updated Title"
|
||||
mock_client.patch.return_value = updated_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"index": 42,
|
||||
"title": "Updated Title",
|
||||
}
|
||||
|
||||
result = await _update_issue(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Updated issue #42" in result[0].text
|
||||
|
||||
mock_client.patch.assert_called_once()
|
||||
call_args = mock_client.patch.call_args
|
||||
assert "/repos/testowner/testrepo/issues/42" in call_args[0][0]
|
||||
assert call_args[1]["json"]["title"] == "Updated Title"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_issue_state(self, mock_client, sample_issue):
|
||||
"""Test closing an issue."""
|
||||
closed_issue = sample_issue.copy()
|
||||
closed_issue["state"] = "closed"
|
||||
mock_client.patch.return_value = closed_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"index": 42,
|
||||
"state": "closed",
|
||||
}
|
||||
|
||||
result = await _update_issue(arguments, mock_client)
|
||||
|
||||
assert "State: closed" in result[0].text
|
||||
|
||||
call_args = mock_client.patch.call_args
|
||||
assert call_args[1]["json"]["state"] == "closed"
|
||||
|
||||
|
||||
class TestIssueToolHandler:
|
||||
"""Test handle_issue_tool function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_list_issues(self, mock_client, sample_issue):
|
||||
"""Test handler routes to _list_issues."""
|
||||
mock_client.get.return_value = [sample_issue]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_issue_tool("gitea_list_issues", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
mock_client.get.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_get_issue(self, mock_client, sample_issue):
|
||||
"""Test handler routes to _get_issue."""
|
||||
mock_client.get.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"index": 42,
|
||||
}
|
||||
|
||||
result = await handle_issue_tool("gitea_get_issue", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Issue #42" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_create_issue(self, mock_client, sample_issue):
|
||||
"""Test handler routes to _create_issue."""
|
||||
mock_client.post.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "New Issue",
|
||||
}
|
||||
|
||||
result = await handle_issue_tool("gitea_create_issue", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created issue" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_update_issue(self, mock_client, sample_issue):
|
||||
"""Test handler routes to _update_issue."""
|
||||
mock_client.patch.return_value = sample_issue
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"index": 42,
|
||||
"title": "Updated",
|
||||
}
|
||||
|
||||
result = await handle_issue_tool("gitea_update_issue", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Updated issue" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_unknown_tool(self, mock_client):
|
||||
"""Test handler with unknown tool name."""
|
||||
result = await handle_issue_tool("gitea_unknown_tool", {}, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Unknown issue tool" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_client_error(self, mock_client):
|
||||
"""Test handler gracefully handles GiteaClientError."""
|
||||
mock_client.get.side_effect = GiteaClientError("API Error")
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_issue_tool("gitea_list_issues", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Error:" in result[0].text
|
||||
assert "API Error" in result[0].text
|
||||
@@ -1,256 +0,0 @@
|
||||
"""Tests for label operations tools."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import Tool, TextContent
|
||||
from gitea_mcp.tools.labels import (
|
||||
get_label_tools,
|
||||
handle_label_tool,
|
||||
_list_labels,
|
||||
_create_label,
|
||||
)
|
||||
from gitea_mcp.client import GiteaClientError
|
||||
|
||||
|
||||
class TestLabelToolDefinitions:
|
||||
"""Test label tool schema definitions."""
|
||||
|
||||
def test_get_label_tools_returns_list(self):
|
||||
"""Test that get_label_tools returns a list of Tool objects."""
|
||||
tools = get_label_tools()
|
||||
assert isinstance(tools, list)
|
||||
assert len(tools) == 2
|
||||
assert all(isinstance(tool, Tool) for tool in tools)
|
||||
|
||||
def test_list_labels_tool_schema(self):
|
||||
"""Test gitea_list_labels tool has correct schema."""
|
||||
tools = get_label_tools()
|
||||
list_tool = next(t for t in tools if t.name == "gitea_list_labels")
|
||||
|
||||
assert list_tool.name == "gitea_list_labels"
|
||||
assert "list" in list_tool.description.lower()
|
||||
assert "label" in list_tool.description.lower()
|
||||
|
||||
schema = list_tool.inputSchema
|
||||
assert schema["type"] == "object"
|
||||
assert set(schema["required"]) == {"owner", "repo"}
|
||||
assert "owner" in schema["properties"]
|
||||
assert "repo" in schema["properties"]
|
||||
|
||||
def test_create_label_tool_schema(self):
|
||||
"""Test gitea_create_label tool has correct schema."""
|
||||
tools = get_label_tools()
|
||||
create_tool = next(t for t in tools if t.name == "gitea_create_label")
|
||||
|
||||
assert create_tool.name == "gitea_create_label"
|
||||
assert "create" in create_tool.description.lower()
|
||||
|
||||
schema = create_tool.inputSchema
|
||||
assert set(schema["required"]) == {"owner", "repo", "name", "color"}
|
||||
assert "name" in schema["properties"]
|
||||
assert "color" in schema["properties"]
|
||||
assert "description" in schema["properties"]
|
||||
assert "hex" in schema["properties"]["color"]["description"].lower()
|
||||
|
||||
|
||||
class TestListLabels:
|
||||
"""Test _list_labels function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_labels_success(self, mock_client, sample_label):
|
||||
"""Test successful label listing."""
|
||||
additional_label = {
|
||||
"id": 2,
|
||||
"name": "enhancement",
|
||||
"color": "00ff00",
|
||||
"description": "New feature or request",
|
||||
}
|
||||
mock_client.get.return_value = [sample_label, additional_label]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_labels(mock_client, arguments)
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert "bug" in result[0].text
|
||||
assert "enhancement" in result[0].text
|
||||
assert "#ff0000" in result[0].text
|
||||
assert "#00ff00" in result[0].text
|
||||
assert "Something isn't working" in result[0].text
|
||||
assert "New feature" in result[0].text
|
||||
|
||||
mock_client.get.assert_called_once_with("/repos/testowner/testrepo/labels")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_labels_empty(self, mock_client):
|
||||
"""Test listing when no labels found."""
|
||||
mock_client.get.return_value = []
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_labels(mock_client, arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "No labels found" in result[0].text
|
||||
assert "testowner/testrepo" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_labels_without_description(self, mock_client):
|
||||
"""Test listing labels without descriptions."""
|
||||
label_no_desc = {
|
||||
"id": 1,
|
||||
"name": "bug",
|
||||
"color": "ff0000",
|
||||
}
|
||||
mock_client.get.return_value = [label_no_desc]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_labels(mock_client, arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "bug" in result[0].text
|
||||
assert "#ff0000" in result[0].text
|
||||
|
||||
|
||||
class TestCreateLabel:
|
||||
"""Test _create_label function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_label_minimal(self, mock_client, sample_label):
|
||||
"""Test creating label with minimal required fields."""
|
||||
mock_client.post.return_value = sample_label
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"name": "bug",
|
||||
"color": "ff0000",
|
||||
}
|
||||
|
||||
result = await _create_label(mock_client, arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created label 'bug'" in result[0].text
|
||||
assert "#ff0000" in result[0].text
|
||||
assert "testowner/testrepo" in result[0].text
|
||||
|
||||
mock_client.post.assert_called_once()
|
||||
call_args = mock_client.post.call_args
|
||||
assert call_args[0][0] == "/repos/testowner/testrepo/labels"
|
||||
payload = call_args[0][1]
|
||||
assert payload["name"] == "bug"
|
||||
assert payload["color"] == "ff0000"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_label_with_description(self, mock_client, sample_label):
|
||||
"""Test creating label with description."""
|
||||
mock_client.post.return_value = sample_label
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"name": "bug",
|
||||
"color": "ff0000",
|
||||
"description": "Something isn't working",
|
||||
}
|
||||
|
||||
result = await _create_label(mock_client, arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
payload = call_args[0][1]
|
||||
assert payload["description"] == "Something isn't working"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_label_strips_hash(self, mock_client, sample_label):
|
||||
"""Test creating label with # prefix is handled correctly."""
|
||||
mock_client.post.return_value = sample_label
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"name": "bug",
|
||||
"color": "#ff0000",
|
||||
}
|
||||
|
||||
result = await _create_label(mock_client, arguments)
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
payload = call_args[0][1]
|
||||
assert payload["color"] == "ff0000"
|
||||
assert not payload["color"].startswith("#")
|
||||
|
||||
|
||||
class TestLabelToolHandler:
|
||||
"""Test handle_label_tool function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_list_labels(self, mock_client, sample_label):
|
||||
"""Test handler routes to _list_labels."""
|
||||
mock_client.get.return_value = [sample_label]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_label_tool(mock_client, "gitea_list_labels", arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert "bug" in result[0].text
|
||||
mock_client.get.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_create_label(self, mock_client, sample_label):
|
||||
"""Test handler routes to _create_label."""
|
||||
mock_client.post.return_value = sample_label
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"name": "bug",
|
||||
"color": "ff0000",
|
||||
}
|
||||
|
||||
result = await handle_label_tool(mock_client, "gitea_create_label", arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created label" in result[0].text
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_unknown_tool(self, mock_client):
|
||||
"""Test handler with unknown tool name."""
|
||||
result = await handle_label_tool(mock_client, "gitea_unknown_tool", {})
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Unknown label tool" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_client_error(self, mock_client):
|
||||
"""Test handler gracefully handles GiteaClientError."""
|
||||
mock_client.get.side_effect = GiteaClientError("API Error")
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_label_tool(mock_client, "gitea_list_labels", arguments)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Gitea API error:" in result[0].text
|
||||
assert "API Error" in result[0].text
|
||||
@@ -1,334 +0,0 @@
|
||||
"""Tests for milestone operations tools."""
|
||||
|
||||
import pytest
|
||||
from mcp.types import Tool, TextContent
|
||||
from gitea_mcp.tools.milestones import (
|
||||
get_milestone_tools,
|
||||
handle_milestone_tool,
|
||||
_list_milestones,
|
||||
_create_milestone,
|
||||
)
|
||||
from gitea_mcp.client import GiteaClientError
|
||||
|
||||
|
||||
class TestMilestoneToolDefinitions:
|
||||
"""Test milestone tool schema definitions."""
|
||||
|
||||
def test_get_milestone_tools_returns_list(self):
|
||||
"""Test that get_milestone_tools returns a list of Tool objects."""
|
||||
tools = get_milestone_tools()
|
||||
assert isinstance(tools, list)
|
||||
assert len(tools) == 2
|
||||
assert all(isinstance(tool, Tool) for tool in tools)
|
||||
|
||||
def test_list_milestones_tool_schema(self):
|
||||
"""Test gitea_list_milestones tool has correct schema."""
|
||||
tools = get_milestone_tools()
|
||||
list_tool = next(t for t in tools if t.name == "gitea_list_milestones")
|
||||
|
||||
assert list_tool.name == "gitea_list_milestones"
|
||||
assert "list" in list_tool.description.lower()
|
||||
assert "milestone" in list_tool.description.lower()
|
||||
|
||||
schema = list_tool.inputSchema
|
||||
assert schema["type"] == "object"
|
||||
assert set(schema["required"]) == {"owner", "repo"}
|
||||
assert "owner" in schema["properties"]
|
||||
assert "repo" in schema["properties"]
|
||||
assert "state" in schema["properties"]
|
||||
assert schema["properties"]["state"]["enum"] == ["open", "closed", "all"]
|
||||
|
||||
def test_create_milestone_tool_schema(self):
|
||||
"""Test gitea_create_milestone tool has correct schema."""
|
||||
tools = get_milestone_tools()
|
||||
create_tool = next(t for t in tools if t.name == "gitea_create_milestone")
|
||||
|
||||
assert create_tool.name == "gitea_create_milestone"
|
||||
assert "create" in create_tool.description.lower()
|
||||
|
||||
schema = create_tool.inputSchema
|
||||
assert set(schema["required"]) == {"owner", "repo", "title"}
|
||||
assert "title" in schema["properties"]
|
||||
assert "description" in schema["properties"]
|
||||
assert "due_on" in schema["properties"]
|
||||
assert "iso 8601" in schema["properties"]["due_on"]["description"].lower()
|
||||
|
||||
|
||||
class TestListMilestones:
|
||||
"""Test _list_milestones function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_success(self, mock_client, sample_milestone):
|
||||
"""Test successful milestone listing."""
|
||||
additional_milestone = {
|
||||
"id": 11,
|
||||
"title": "v2.0",
|
||||
"description": "Second major release",
|
||||
"state": "open",
|
||||
"open_issues": 10,
|
||||
"closed_issues": 0,
|
||||
"created_at": "2024-02-01T00:00:00Z",
|
||||
"updated_at": "2024-02-01T00:00:00Z",
|
||||
}
|
||||
mock_client.get.return_value = [sample_milestone, additional_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "open",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
assert isinstance(result, list)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert "v1.0" in result[0].text
|
||||
assert "v2.0" in result[0].text
|
||||
assert "First major release" in result[0].text
|
||||
assert "Open Issues: 5" in result[0].text
|
||||
assert "Closed Issues: 15" in result[0].text
|
||||
|
||||
mock_client.get.assert_called_once()
|
||||
call_args = mock_client.get.call_args
|
||||
assert "/repos/testowner/testrepo/milestones" in call_args[0][0]
|
||||
assert call_args[1]["params"]["state"] == "open"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_empty(self, mock_client):
|
||||
"""Test listing when no milestones found."""
|
||||
mock_client.get.return_value = []
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "closed",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "No closed milestones found" in result[0].text
|
||||
assert "testowner/testrepo" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_default_state(self, mock_client, sample_milestone):
|
||||
"""Test listing milestones with default state."""
|
||||
mock_client.get.return_value = [sample_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
call_args = mock_client.get.call_args
|
||||
assert call_args[1]["params"]["state"] == "open"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_all_states(self, mock_client, sample_milestone):
|
||||
"""Test listing milestones with state=all."""
|
||||
mock_client.get.return_value = [sample_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"state": "all",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
assert "Found 1 all milestone(s)" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_with_due_date(self, mock_client, sample_milestone):
|
||||
"""Test milestone listing includes due date."""
|
||||
mock_client.get.return_value = [sample_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
assert "Due: 2024-12-31T23:59:59Z" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_milestones_without_optional_fields(self, mock_client):
|
||||
"""Test milestone listing without description and due date."""
|
||||
minimal_milestone = {
|
||||
"id": 10,
|
||||
"title": "v1.0",
|
||||
"state": "open",
|
||||
"open_issues": 5,
|
||||
"closed_issues": 15,
|
||||
"created_at": "2024-01-01T00:00:00Z",
|
||||
}
|
||||
mock_client.get.return_value = [minimal_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await _list_milestones(arguments, mock_client)
|
||||
|
||||
assert "v1.0" in result[0].text
|
||||
assert "State: open" in result[0].text
|
||||
assert "Created: 2024-01-01T00:00:00Z" in result[0].text
|
||||
|
||||
|
||||
class TestCreateMilestone:
|
||||
"""Test _create_milestone function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_milestone_minimal(self, mock_client, sample_milestone):
|
||||
"""Test creating milestone with minimal required fields."""
|
||||
mock_client.post.return_value = sample_milestone
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "v1.0",
|
||||
}
|
||||
|
||||
result = await _create_milestone(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created milestone: v1.0" in result[0].text
|
||||
|
||||
mock_client.post.assert_called_once()
|
||||
call_args = mock_client.post.call_args
|
||||
assert "/repos/testowner/testrepo/milestones" in call_args[0][0]
|
||||
json_data = call_args[1]["json"]
|
||||
assert json_data["title"] == "v1.0"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_milestone_with_description(self, mock_client, sample_milestone):
|
||||
"""Test creating milestone with description."""
|
||||
mock_client.post.return_value = sample_milestone
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "v1.0",
|
||||
"description": "First major release",
|
||||
}
|
||||
|
||||
result = await _create_milestone(arguments, mock_client)
|
||||
|
||||
assert "Description: First major release" in result[0].text
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
json_data = call_args[1]["json"]
|
||||
assert json_data["description"] == "First major release"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_milestone_with_due_date(self, mock_client, sample_milestone):
|
||||
"""Test creating milestone with due date."""
|
||||
mock_client.post.return_value = sample_milestone
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "v1.0",
|
||||
"due_on": "2024-12-31T23:59:59Z",
|
||||
}
|
||||
|
||||
result = await _create_milestone(arguments, mock_client)
|
||||
|
||||
assert "Due: 2024-12-31T23:59:59Z" in result[0].text
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
json_data = call_args[1]["json"]
|
||||
assert json_data["due_on"] == "2024-12-31T23:59:59Z"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_milestone_full(self, mock_client, sample_milestone):
|
||||
"""Test creating milestone with all optional fields."""
|
||||
mock_client.post.return_value = sample_milestone
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "v1.0",
|
||||
"description": "First major release",
|
||||
"due_on": "2024-12-31T23:59:59Z",
|
||||
}
|
||||
|
||||
result = await _create_milestone(arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created milestone: v1.0" in result[0].text
|
||||
assert "Description: First major release" in result[0].text
|
||||
assert "Due: 2024-12-31T23:59:59Z" in result[0].text
|
||||
|
||||
call_args = mock_client.post.call_args
|
||||
json_data = call_args[1]["json"]
|
||||
assert json_data["title"] == "v1.0"
|
||||
assert json_data["description"] == "First major release"
|
||||
assert json_data["due_on"] == "2024-12-31T23:59:59Z"
|
||||
|
||||
|
||||
class TestMilestoneToolHandler:
|
||||
"""Test handle_milestone_tool function."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_list_milestones(self, mock_client, sample_milestone):
|
||||
"""Test handler routes to _list_milestones."""
|
||||
mock_client.get.return_value = [sample_milestone]
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_milestone_tool("gitea_list_milestones", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], TextContent)
|
||||
assert "v1.0" in result[0].text
|
||||
mock_client.get.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_create_milestone(self, mock_client, sample_milestone):
|
||||
"""Test handler routes to _create_milestone."""
|
||||
mock_client.post.return_value = sample_milestone
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
"title": "v1.0",
|
||||
}
|
||||
|
||||
result = await handle_milestone_tool("gitea_create_milestone", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Created milestone" in result[0].text
|
||||
mock_client.post.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_unknown_tool(self, mock_client):
|
||||
"""Test handler with unknown tool name."""
|
||||
result = await handle_milestone_tool("gitea_unknown_tool", {}, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Unknown milestone tool" in result[0].text
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_client_error(self, mock_client):
|
||||
"""Test handler gracefully handles GiteaClientError."""
|
||||
mock_client.get.side_effect = GiteaClientError("API Error")
|
||||
|
||||
arguments = {
|
||||
"owner": "testowner",
|
||||
"repo": "testrepo",
|
||||
}
|
||||
|
||||
result = await handle_milestone_tool("gitea_list_milestones", arguments, mock_client)
|
||||
|
||||
assert len(result) == 1
|
||||
assert "Error:" in result[0].text
|
||||
assert "API Error" in result[0].text
|
||||
Reference in New Issue
Block a user