diff --git a/src/gitea_mcp/__init__.py b/src/gitea_mcp/__init__.py deleted file mode 100644 index 7a9bc41..0000000 --- a/src/gitea_mcp/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -"""Gitea MCP Server - MCP server for Gitea API integration.""" - -__version__ = "0.1.0" diff --git a/src/gitea_mcp/auth.py b/src/gitea_mcp/auth.py deleted file mode 100644 index ebffeac..0000000 --- a/src/gitea_mcp/auth.py +++ /dev/null @@ -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", - } diff --git a/src/gitea_mcp/client.py b/src/gitea_mcp/client.py deleted file mode 100644 index 2c09ad8..0000000 --- a/src/gitea_mcp/client.py +++ /dev/null @@ -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 diff --git a/src/gitea_mcp/server.py b/src/gitea_mcp/server.py deleted file mode 100644 index 63f7755..0000000 --- a/src/gitea_mcp/server.py +++ /dev/null @@ -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() diff --git a/src/gitea_mcp/tools/__init__.py b/src/gitea_mcp/tools/__init__.py deleted file mode 100644 index f071371..0000000 --- a/src/gitea_mcp/tools/__init__.py +++ /dev/null @@ -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", -] diff --git a/src/gitea_mcp/tools/issues.py b/src/gitea_mcp/tools/issues.py deleted file mode 100644 index a742109..0000000 --- a/src/gitea_mcp/tools/issues.py +++ /dev/null @@ -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)] diff --git a/src/gitea_mcp/tools/labels.py b/src/gitea_mcp/tools/labels.py deleted file mode 100644 index 0f8a286..0000000 --- a/src/gitea_mcp/tools/labels.py +++ /dev/null @@ -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}", - ) - ] diff --git a/src/gitea_mcp/tools/milestones.py b/src/gitea_mcp/tools/milestones.py deleted file mode 100644 index ab49a72..0000000 --- a/src/gitea_mcp/tools/milestones.py +++ /dev/null @@ -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)] diff --git a/tests/__init__.py b/tests/__init__.py deleted file mode 100644 index f1a4707..0000000 --- a/tests/__init__.py +++ /dev/null @@ -1 +0,0 @@ -"""Tests for Gitea MCP server.""" diff --git a/tests/conftest.py b/tests/conftest.py deleted file mode 100644 index 024709b..0000000 --- a/tests/conftest.py +++ /dev/null @@ -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", - } diff --git a/tests/test_auth.py b/tests/test_auth.py deleted file mode 100644 index 4bde5b5..0000000 --- a/tests/test_auth.py +++ /dev/null @@ -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" diff --git a/tests/test_client.py b/tests/test_client.py deleted file mode 100644 index 2ea1011..0000000 --- a/tests/test_client.py +++ /dev/null @@ -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) diff --git a/tests/test_issues.py b/tests/test_issues.py deleted file mode 100644 index 521dddc..0000000 --- a/tests/test_issues.py +++ /dev/null @@ -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 diff --git a/tests/test_labels.py b/tests/test_labels.py deleted file mode 100644 index 5dbbd99..0000000 --- a/tests/test_labels.py +++ /dev/null @@ -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 diff --git a/tests/test_milestones.py b/tests/test_milestones.py deleted file mode 100644 index ba2ea40..0000000 --- a/tests/test_milestones.py +++ /dev/null @@ -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