generated from personal-projects/leo-claude-mktplace
Compare commits
10 Commits
95ed3d81cf
...
b94dcebfc7
| Author | SHA1 | Date | |
|---|---|---|---|
| b94dcebfc7 | |||
| 201cc680ca | |||
| 0653a4f70e | |||
| 13ffd8a543 | |||
| 2230bceb51 | |||
| ab8c9069da | |||
| 7ffc0f9ce4 | |||
| 38dd315dd5 | |||
| 694406941c | |||
| 1e0d896d87 |
368
README.md
368
README.md
@@ -1,27 +1,35 @@
|
||||
# Gitea MCP Remote
|
||||
# Gitea MCP Server
|
||||
|
||||
MCP server for Gitea API integration.
|
||||
A Model Context Protocol (MCP) server that enables AI assistants like Claude to interact with Gitea repositories through its API. This server provides tools for managing issues, labels, and milestones in your Gitea instance.
|
||||
|
||||
## Overview
|
||||
## Features
|
||||
|
||||
This project provides a Model Context Protocol (MCP) server that enables AI assistants to interact with Gitea through its API.
|
||||
|
||||
## Project Status
|
||||
|
||||
Currently in initial development. Project structure has been initialized.
|
||||
- **Issue Operations**: List, get, create, and update issues with full support for labels, milestones, and assignees
|
||||
- **Label Management**: List and create labels with custom colors and descriptions
|
||||
- **Milestone Management**: List and create milestones with due dates and descriptions
|
||||
- **Async API**: Built on modern async Python for efficient operations
|
||||
- **Type Safety**: Full type hints for better IDE support and code quality
|
||||
|
||||
## Requirements
|
||||
|
||||
- Python >= 3.10
|
||||
- Gitea instance with API access
|
||||
- Gitea API token with appropriate permissions
|
||||
|
||||
## Installation
|
||||
|
||||
### From Source
|
||||
|
||||
```bash
|
||||
# Clone the repository
|
||||
git clone https://github.com/lmiranda/gitea-mcp-remote.git
|
||||
cd gitea-mcp-remote
|
||||
|
||||
# Install the package
|
||||
pip install -e .
|
||||
```
|
||||
|
||||
## Development
|
||||
### For Development
|
||||
|
||||
Install with development dependencies:
|
||||
|
||||
@@ -29,12 +37,350 @@ Install with development dependencies:
|
||||
pip install -e ".[dev]"
|
||||
```
|
||||
|
||||
Run tests:
|
||||
## Configuration
|
||||
|
||||
The server requires two environment variables to connect to your Gitea instance:
|
||||
|
||||
- `GITEA_API_URL`: Base URL of your Gitea instance (e.g., `https://gitea.example.com/api/v1`)
|
||||
- `GITEA_API_TOKEN`: Personal access token for authentication
|
||||
|
||||
### Creating a .env File
|
||||
|
||||
Create a `.env` file in your project directory:
|
||||
|
||||
```bash
|
||||
GITEA_API_URL=https://gitea.example.com/api/v1
|
||||
GITEA_API_TOKEN=your_gitea_token_here
|
||||
```
|
||||
|
||||
### Getting a Gitea API Token
|
||||
|
||||
1. Log into your Gitea instance
|
||||
2. Navigate to Settings > Applications
|
||||
3. Under "Generate New Token", enter a name (e.g., "MCP Server")
|
||||
4. Select appropriate permissions (minimum: read/write for repositories)
|
||||
5. Click "Generate Token" and copy the token
|
||||
6. Add the token to your `.env` file
|
||||
|
||||
## Usage with Claude Desktop
|
||||
|
||||
Add this configuration to your Claude Desktop config file:
|
||||
|
||||
**Location:**
|
||||
- macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
|
||||
- Windows: `%APPDATA%\Claude\claude_desktop_config.json`
|
||||
- Linux: `~/.config/Claude/claude_desktop_config.json`
|
||||
|
||||
**Configuration:**
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"gitea": {
|
||||
"command": "python",
|
||||
"args": ["-m", "gitea_mcp.server"],
|
||||
"env": {
|
||||
"GITEA_API_URL": "https://gitea.example.com/api/v1",
|
||||
"GITEA_API_TOKEN": "your_gitea_token_here"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Or, if you prefer using a .env file:
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"gitea": {
|
||||
"command": "python",
|
||||
"args": ["-m", "gitea_mcp.server"],
|
||||
"cwd": "/path/to/gitea-mcp-remote"
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Usage with Claude Code
|
||||
|
||||
Add to your MCP settings file:
|
||||
|
||||
**Location:** `~/.config/claude-code/mcp.json` (or your configured MCP settings path)
|
||||
|
||||
**Configuration:**
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"gitea": {
|
||||
"command": "python",
|
||||
"args": ["-m", "gitea_mcp.server"],
|
||||
"env": {
|
||||
"GITEA_API_URL": "https://gitea.example.com/api/v1",
|
||||
"GITEA_API_TOKEN": "your_gitea_token_here"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Available Tools
|
||||
|
||||
### Issue Tools
|
||||
|
||||
#### gitea_list_issues
|
||||
|
||||
List issues in a repository with optional filters.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner (username or organization)
|
||||
- `repo` (required): Repository name
|
||||
- `state` (optional): Filter by state - `open`, `closed`, or `all` (default: `open`)
|
||||
- `labels` (optional): Comma-separated list of label names to filter by
|
||||
- `milestone` (optional): Milestone name to filter by
|
||||
- `page` (optional): Page number for pagination (default: 1)
|
||||
- `limit` (optional): Number of issues per page (default: 30)
|
||||
|
||||
**Example:**
|
||||
```
|
||||
List all open issues in myorg/myrepo
|
||||
```
|
||||
|
||||
#### gitea_get_issue
|
||||
|
||||
Get details of a specific issue by number.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `index` (required): Issue number
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Get details of issue #42 in myorg/myrepo
|
||||
```
|
||||
|
||||
#### gitea_create_issue
|
||||
|
||||
Create a new issue in a repository.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `title` (required): Issue title
|
||||
- `body` (optional): Issue description/body
|
||||
- `labels` (optional): Array of label IDs to assign
|
||||
- `milestone` (optional): Milestone ID to assign
|
||||
- `assignees` (optional): Array of usernames to assign
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Create an issue in myorg/myrepo with title "Bug: Login fails" and body "Users cannot log in with special characters in password"
|
||||
```
|
||||
|
||||
#### gitea_update_issue
|
||||
|
||||
Update an existing issue.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `index` (required): Issue number
|
||||
- `title` (optional): New issue title
|
||||
- `body` (optional): New issue body
|
||||
- `state` (optional): Issue state - `open` or `closed`
|
||||
- `labels` (optional): Array of label IDs (replaces existing)
|
||||
- `milestone` (optional): Milestone ID to assign
|
||||
- `assignees` (optional): Array of usernames (replaces existing)
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Close issue #42 in myorg/myrepo
|
||||
```
|
||||
|
||||
### Label Tools
|
||||
|
||||
#### gitea_list_labels
|
||||
|
||||
List all labels in a repository.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
|
||||
**Example:**
|
||||
```
|
||||
List all labels in myorg/myrepo
|
||||
```
|
||||
|
||||
#### gitea_create_label
|
||||
|
||||
Create a new label in a repository.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `name` (required): Label name
|
||||
- `color` (required): Label color (hex without #, e.g., `ff0000` for red)
|
||||
- `description` (optional): Label description
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Create a label "bug" with red color (ff0000) in myorg/myrepo
|
||||
```
|
||||
|
||||
### Milestone Tools
|
||||
|
||||
#### gitea_list_milestones
|
||||
|
||||
List milestones in a repository.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `state` (optional): Filter by state - `open`, `closed`, or `all` (default: `open`)
|
||||
|
||||
**Example:**
|
||||
```
|
||||
List all milestones in myorg/myrepo
|
||||
```
|
||||
|
||||
#### gitea_create_milestone
|
||||
|
||||
Create a new milestone in a repository.
|
||||
|
||||
**Parameters:**
|
||||
- `owner` (required): Repository owner
|
||||
- `repo` (required): Repository name
|
||||
- `title` (required): Milestone title
|
||||
- `description` (optional): Milestone description
|
||||
- `due_on` (optional): Due date in ISO 8601 format (e.g., `2024-12-31T23:59:59Z`)
|
||||
|
||||
**Example:**
|
||||
```
|
||||
Create a milestone "v1.0 Release" with due date 2024-12-31 in myorg/myrepo
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
### Core Components
|
||||
|
||||
#### GiteaClient
|
||||
|
||||
HTTP client for Gitea API interactions.
|
||||
|
||||
**Methods:**
|
||||
- `get(endpoint, params)`: GET request
|
||||
- `post(endpoint, json)`: POST request
|
||||
- `patch(endpoint, json)`: PATCH request
|
||||
|
||||
#### AuthConfig
|
||||
|
||||
Configuration manager for API authentication.
|
||||
|
||||
**Environment Variables:**
|
||||
- `GITEA_API_URL`: Gitea API base URL
|
||||
- `GITEA_API_TOKEN`: Authentication token
|
||||
|
||||
**Methods:**
|
||||
- `get_auth_headers()`: Returns authentication headers
|
||||
|
||||
### Tool Modules
|
||||
|
||||
- `gitea_mcp.tools.issues`: Issue operation tools and handlers
|
||||
- `gitea_mcp.tools.labels`: Label operation tools and handlers
|
||||
- `gitea_mcp.tools.milestones`: Milestone operation tools and handlers
|
||||
|
||||
## Development
|
||||
|
||||
### Setup Development Environment
|
||||
|
||||
```bash
|
||||
# Install with development dependencies
|
||||
pip install -e ".[dev]"
|
||||
```
|
||||
|
||||
### Running Tests
|
||||
|
||||
```bash
|
||||
pytest
|
||||
```
|
||||
|
||||
### Project Structure
|
||||
|
||||
```
|
||||
gitea-mcp-remote/
|
||||
├── src/
|
||||
│ └── gitea_mcp/
|
||||
│ ├── __init__.py
|
||||
│ ├── server.py # MCP server implementation
|
||||
│ ├── auth.py # Authentication config
|
||||
│ ├── client.py # Gitea API client
|
||||
│ └── tools/ # Tool implementations
|
||||
│ ├── __init__.py
|
||||
│ ├── issues.py # Issue tools
|
||||
│ ├── labels.py # Label tools
|
||||
│ └── milestones.py # Milestone tools
|
||||
├── tests/ # Test suite
|
||||
├── pyproject.toml # Project configuration
|
||||
└── README.md # This file
|
||||
```
|
||||
|
||||
### Code Quality
|
||||
|
||||
This project uses:
|
||||
- Type hints throughout the codebase
|
||||
- Async/await for all I/O operations
|
||||
- Comprehensive error handling
|
||||
- Structured logging
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Authentication Errors
|
||||
|
||||
If you receive authentication errors:
|
||||
|
||||
1. Verify your `GITEA_API_TOKEN` is correct
|
||||
2. Check that the token has appropriate permissions
|
||||
3. Ensure your `GITEA_API_URL` includes `/api/v1` at the end
|
||||
4. Verify the Gitea instance is accessible from your network
|
||||
|
||||
### Connection Errors
|
||||
|
||||
If you cannot connect to Gitea:
|
||||
|
||||
1. Check that `GITEA_API_URL` is correct and accessible
|
||||
2. Verify network connectivity to the Gitea instance
|
||||
3. Check for firewalls or proxies blocking the connection
|
||||
|
||||
### Tool Not Found
|
||||
|
||||
If Claude cannot find the tools:
|
||||
|
||||
1. Restart Claude Desktop/Code after updating the configuration
|
||||
2. Verify the configuration file syntax is valid JSON
|
||||
3. Check that Python is in your PATH
|
||||
4. Ensure the package is properly installed (`pip list | grep gitea-mcp`)
|
||||
|
||||
## Contributing
|
||||
|
||||
Contributions are welcome! Please feel free to submit issues or pull requests.
|
||||
|
||||
## License
|
||||
|
||||
MIT
|
||||
MIT License - see LICENSE file for details
|
||||
|
||||
## Version
|
||||
|
||||
Current version: 0.1.0
|
||||
|
||||
## Author
|
||||
|
||||
Leo Miranda
|
||||
|
||||
## Links
|
||||
|
||||
- Repository: https://github.com/lmiranda/gitea-mcp-remote
|
||||
- Issues: https://github.com/lmiranda/gitea-mcp-remote/issues
|
||||
- MCP Documentation: https://modelcontextprotocol.io
|
||||
|
||||
@@ -33,8 +33,12 @@ dependencies = [
|
||||
dev = [
|
||||
"pytest>=7.0.0",
|
||||
"pytest-asyncio>=0.21.0",
|
||||
"pytest-cov>=4.0.0",
|
||||
]
|
||||
|
||||
[project.scripts]
|
||||
gitea-mcp = "gitea_mcp.server:main"
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/lmiranda/gitea-mcp-remote"
|
||||
Repository = "https://github.com/lmiranda/gitea-mcp-remote"
|
||||
|
||||
21
run_tests.sh
Executable file
21
run_tests.sh
Executable file
@@ -0,0 +1,21 @@
|
||||
#!/bin/bash
|
||||
# Test runner script for gitea-mcp-remote
|
||||
|
||||
set -e
|
||||
|
||||
echo "Running gitea-mcp-remote test suite..."
|
||||
echo
|
||||
|
||||
# Install dev dependencies if not already installed
|
||||
if ! command -v pytest &> /dev/null; then
|
||||
echo "Installing dev dependencies..."
|
||||
pip install -e ".[dev]"
|
||||
echo
|
||||
fi
|
||||
|
||||
# Run tests with coverage
|
||||
echo "Running tests with coverage..."
|
||||
python -m pytest tests/ -v --cov=src/gitea_mcp --cov-report=term-missing --cov-report=html
|
||||
|
||||
echo
|
||||
echo "Coverage report saved to htmlcov/index.html"
|
||||
51
src/gitea_mcp/auth.py
Normal file
51
src/gitea_mcp/auth.py
Normal file
@@ -0,0 +1,51 @@
|
||||
"""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",
|
||||
}
|
||||
216
src/gitea_mcp/client.py
Normal file
216
src/gitea_mcp/client.py
Normal file
@@ -0,0 +1,216 @@
|
||||
"""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,7 +1,173 @@
|
||||
"""MCP server implementation for Gitea API integration.
|
||||
"""MCP server implementation for Gitea API integration."""
|
||||
|
||||
This module will contain the main MCP server implementation.
|
||||
"""
|
||||
import asyncio
|
||||
import argparse
|
||||
from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
# Placeholder for MCP server implementation
|
||||
# TODO: Implement MCP server in future issues
|
||||
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 +1,14 @@
|
||||
"""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",
|
||||
]
|
||||
|
||||
392
src/gitea_mcp/tools/issues.py
Normal file
392
src/gitea_mcp/tools/issues.py
Normal file
@@ -0,0 +1,392 @@
|
||||
"""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)]
|
||||
170
src/gitea_mcp/tools/labels.py
Normal file
170
src/gitea_mcp/tools/labels.py
Normal file
@@ -0,0 +1,170 @@
|
||||
"""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}",
|
||||
)
|
||||
]
|
||||
190
src/gitea_mcp/tools/milestones.py
Normal file
190
src/gitea_mcp/tools/milestones.py
Normal file
@@ -0,0 +1,190 @@
|
||||
"""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)]
|
||||
103
tests/conftest.py
Normal file
103
tests/conftest.py
Normal file
@@ -0,0 +1,103 @@
|
||||
"""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",
|
||||
}
|
||||
56
tests/test_auth.py
Normal file
56
tests/test_auth.py
Normal file
@@ -0,0 +1,56 @@
|
||||
"""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"
|
||||
93
tests/test_client.py
Normal file
93
tests/test_client.py
Normal file
@@ -0,0 +1,93 @@
|
||||
"""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)
|
||||
372
tests/test_issues.py
Normal file
372
tests/test_issues.py
Normal file
@@ -0,0 +1,372 @@
|
||||
"""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
|
||||
256
tests/test_labels.py
Normal file
256
tests/test_labels.py
Normal file
@@ -0,0 +1,256 @@
|
||||
"""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
|
||||
334
tests/test_milestones.py
Normal file
334
tests/test_milestones.py
Normal file
@@ -0,0 +1,334 @@
|
||||
"""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