refactor: bundle MCP servers inside plugins for cache compatibility
Claude Code only caches the plugin directory when installed from a
marketplace, not parent directories. This broke the shared mcp-servers/
architecture because relative paths like ../../mcp-servers/ resolved
to non-existent locations in the cache.
Changes:
- Move gitea and wikijs MCP servers into plugins/projman/mcp-servers/
- Move netbox MCP server into plugins/cmdb-assistant/mcp-servers/
- Update .mcp.json files to use ${CLAUDE_PLUGIN_ROOT}/mcp-servers/
- Update setup.sh to handle new bundled structure
- Add netbox.env config template to setup.sh
- Update CLAUDE.md and CANONICAL-PATHS.md documentation
This ensures plugins work correctly when installed and cached.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -1,413 +0,0 @@
|
||||
# Gitea MCP Server
|
||||
|
||||
Model Context Protocol (MCP) server for Gitea integration with Claude Code.
|
||||
|
||||
## Overview
|
||||
|
||||
The Gitea MCP Server provides Claude Code with direct access to Gitea for issue management, label operations, and repository tracking. It supports both single-repository (project mode) and multi-repository (company/PMO mode) operations.
|
||||
|
||||
**Status**: ✅ Phase 1 Complete - Fully functional and tested
|
||||
|
||||
## Features
|
||||
|
||||
### Core Functionality
|
||||
|
||||
- **Issue Management**: CRUD operations for Gitea issues
|
||||
- **Label Taxonomy**: Dynamic 44-label system with intelligent suggestions
|
||||
- **Mode Detection**: Automatic project vs company-wide mode detection
|
||||
- **Branch-Aware Security**: Prevents accidental changes on production branches
|
||||
- **Hybrid Configuration**: System-level credentials + project-level paths
|
||||
- **PMO Support**: Multi-repository aggregation for organization-wide views
|
||||
|
||||
### Tools Provided
|
||||
|
||||
| Tool | Description | Mode |
|
||||
|------|-------------|------|
|
||||
| `list_issues` | List issues from repository | Both |
|
||||
| `get_issue` | Get specific issue details | Both |
|
||||
| `create_issue` | Create new issue with labels | Both |
|
||||
| `update_issue` | Update existing issue | Both |
|
||||
| `add_comment` | Add comment to issue | Both |
|
||||
| `get_labels` | Get all labels (org + repo) | Both |
|
||||
| `suggest_labels` | Intelligent label suggestion | Both |
|
||||
| `aggregate_issues` | Cross-repository issue aggregation | PMO Only |
|
||||
|
||||
## Architecture
|
||||
|
||||
### Directory Structure
|
||||
|
||||
```
|
||||
mcp-servers/gitea/
|
||||
├── .venv/ # Python virtual environment
|
||||
├── requirements.txt # Python dependencies
|
||||
├── mcp_server/
|
||||
│ ├── __init__.py
|
||||
│ ├── server.py # MCP server entry point
|
||||
│ ├── config.py # Configuration loader
|
||||
│ ├── gitea_client.py # Gitea API client
|
||||
│ └── tools/
|
||||
│ ├── __init__.py
|
||||
│ ├── issues.py # Issue tools
|
||||
│ └── labels.py # Label tools
|
||||
├── tests/
|
||||
│ ├── __init__.py
|
||||
│ ├── test_config.py
|
||||
│ ├── test_gitea_client.py
|
||||
│ ├── test_issues.py
|
||||
│ └── test_labels.py
|
||||
├── README.md # This file
|
||||
└── TESTING.md # Testing instructions
|
||||
```
|
||||
|
||||
### Mode Detection
|
||||
|
||||
The server operates in two modes based on environment variables:
|
||||
|
||||
**Project Mode** (Single Repository):
|
||||
- When `GITEA_REPO` is set
|
||||
- Operates on single repository
|
||||
- Used by `projman` plugin
|
||||
|
||||
**Company Mode** (Multi-Repository / PMO):
|
||||
- When `GITEA_REPO` is NOT set
|
||||
- Operates on all repositories in organization
|
||||
- Used by `projman-pmo` plugin
|
||||
|
||||
### Branch-Aware Security
|
||||
|
||||
Operations are restricted based on the current Git branch:
|
||||
|
||||
| Branch | Read | Create Issue | Update/Comment |
|
||||
|--------|------|--------------|----------------|
|
||||
| `main`, `master`, `prod/*` | ✅ | ❌ | ❌ |
|
||||
| `staging`, `stage/*` | ✅ | ✅ | ❌ |
|
||||
| `development`, `develop`, `feat/*`, `dev/*` | ✅ | ✅ | ✅ |
|
||||
|
||||
## Installation
|
||||
|
||||
### Prerequisites
|
||||
|
||||
- Python 3.10 or higher
|
||||
- Git repository (for branch detection)
|
||||
- Access to Gitea instance with API token
|
||||
|
||||
### Step 1: Install Dependencies
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
python3 -m venv .venv
|
||||
source .venv/bin/activate # Linux/Mac
|
||||
# or .venv\Scripts\activate # Windows
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
### Step 2: Configure System-Level Settings
|
||||
|
||||
Create `~/.config/claude/gitea.env`:
|
||||
|
||||
```bash
|
||||
mkdir -p ~/.config/claude
|
||||
|
||||
cat > ~/.config/claude/gitea.env << EOF
|
||||
GITEA_API_URL=https://gitea.example.com/api/v1
|
||||
GITEA_API_TOKEN=your_gitea_token_here
|
||||
GITEA_OWNER=bandit
|
||||
EOF
|
||||
|
||||
chmod 600 ~/.config/claude/gitea.env
|
||||
```
|
||||
|
||||
### Step 3: Configure Project-Level Settings (Optional)
|
||||
|
||||
For project mode, create `.env` in your project root:
|
||||
|
||||
```bash
|
||||
echo "GITEA_REPO=your-repo-name" > .env
|
||||
echo ".env" >> .gitignore
|
||||
```
|
||||
|
||||
For company/PMO mode, omit the `.env` file or don't set `GITEA_REPO`.
|
||||
|
||||
## Configuration
|
||||
|
||||
### System-Level Configuration
|
||||
|
||||
**File**: `~/.config/claude/gitea.env`
|
||||
|
||||
**Required Variables**:
|
||||
- `GITEA_API_URL` - Gitea API endpoint (e.g., `https://gitea.example.com/api/v1`)
|
||||
- `GITEA_API_TOKEN` - Personal access token with repo permissions
|
||||
- `GITEA_OWNER` - Organization or user name (e.g., `bandit`)
|
||||
|
||||
### Project-Level Configuration
|
||||
|
||||
**File**: `<project-root>/.env`
|
||||
|
||||
**Optional Variables**:
|
||||
- `GITEA_REPO` - Repository name (enables project mode)
|
||||
|
||||
### Generating Gitea API Token
|
||||
|
||||
1. Log into Gitea: https://gitea.example.com
|
||||
2. Navigate to: **Settings** → **Applications** → **Manage Access Tokens**
|
||||
3. Click **Generate New Token**
|
||||
4. Configure token:
|
||||
- **Token Name**: `claude-code-mcp`
|
||||
- **Permissions**:
|
||||
- ✅ `repo` (all) - Read/write repositories, issues, labels
|
||||
- ✅ `read:org` - Read organization information and labels
|
||||
- ✅ `read:user` - Read user information
|
||||
5. Click **Generate Token**
|
||||
6. Copy token immediately (shown only once)
|
||||
7. Add to `~/.config/claude/gitea.env`
|
||||
|
||||
## Usage
|
||||
|
||||
### Running the MCP Server
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate
|
||||
python -m mcp_server.server
|
||||
```
|
||||
|
||||
The server communicates via JSON-RPC 2.0 over stdio.
|
||||
|
||||
### Integration with Claude Code Plugins
|
||||
|
||||
The MCP server is designed to be used by Claude Code plugins via `.mcp.json` configuration:
|
||||
|
||||
```json
|
||||
{
|
||||
"mcpServers": {
|
||||
"gitea": {
|
||||
"command": "python",
|
||||
"args": ["-m", "mcp_server.server"],
|
||||
"cwd": "${CLAUDE_PLUGIN_ROOT}/../mcp-servers/gitea",
|
||||
"env": {
|
||||
"PYTHONPATH": "${CLAUDE_PLUGIN_ROOT}/../mcp-servers/gitea"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Example Tool Calls
|
||||
|
||||
**List Issues**:
|
||||
```python
|
||||
from mcp_server.tools.issues import IssueTools
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
client = GiteaClient()
|
||||
issue_tools = IssueTools(client)
|
||||
|
||||
issues = await issue_tools.list_issues(state='open', labels=['Type/Bug'])
|
||||
```
|
||||
|
||||
**Suggest Labels**:
|
||||
```python
|
||||
from mcp_server.tools.labels import LabelTools
|
||||
|
||||
label_tools = LabelTools(client)
|
||||
|
||||
context = "Fix critical authentication bug in production API"
|
||||
suggestions = await label_tools.suggest_labels(context)
|
||||
# Returns: ['Type/Bug', 'Priority/Critical', 'Component/Auth', 'Component/API', ...]
|
||||
```
|
||||
|
||||
## Testing
|
||||
|
||||
### Unit Tests
|
||||
|
||||
Run all 42 unit tests with mocks:
|
||||
|
||||
```bash
|
||||
pytest tests/ -v
|
||||
```
|
||||
|
||||
Expected: `42 passed in 0.57s`
|
||||
|
||||
### Integration Tests
|
||||
|
||||
Test with real Gitea instance:
|
||||
|
||||
```bash
|
||||
python -c "
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
client = GiteaClient()
|
||||
issues = client.list_issues(state='open')
|
||||
print(f'Found {len(issues)} open issues')
|
||||
"
|
||||
```
|
||||
|
||||
### Full Testing Guide
|
||||
|
||||
See [TESTING.md](./TESTING.md) for comprehensive testing instructions.
|
||||
|
||||
## Label Taxonomy System
|
||||
|
||||
The system supports a dynamic 44-label taxonomy (28 org + 16 repo):
|
||||
|
||||
**Organization Labels (28)**:
|
||||
- `Agent/*` (2) - Agent/Human, Agent/Claude
|
||||
- `Complexity/*` (3) - Simple, Medium, Complex
|
||||
- `Efforts/*` (5) - XS, S, M, L, XL
|
||||
- `Priority/*` (4) - Low, Medium, High, Critical
|
||||
- `Risk/*` (3) - Low, Medium, High
|
||||
- `Source/*` (4) - Development, Staging, Production, Customer
|
||||
- `Type/*` (6) - Bug, Feature, Refactor, Documentation, Test, Chore
|
||||
|
||||
**Repository Labels (16)**:
|
||||
- `Component/*` (9) - Backend, Frontend, API, Database, Auth, Deploy, Testing, Docs, Infra
|
||||
- `Tech/*` (7) - Python, JavaScript, Docker, PostgreSQL, Redis, Vue, FastAPI
|
||||
|
||||
Labels are fetched dynamically from Gitea and suggestions adapt to the current taxonomy.
|
||||
|
||||
## Security
|
||||
|
||||
### Token Storage
|
||||
|
||||
- Store tokens in `~/.config/claude/gitea.env`
|
||||
- Set file permissions to `600` (read/write owner only)
|
||||
- Never commit tokens to Git
|
||||
- Use separate tokens for development and production
|
||||
|
||||
### Branch Detection
|
||||
|
||||
The MCP server implements defense-in-depth branch detection:
|
||||
|
||||
1. **MCP Tools**: Check branch before operations
|
||||
2. **Agent Prompts**: Warn users about branch restrictions
|
||||
3. **CLAUDE.md**: Provides additional context
|
||||
|
||||
### Input Validation
|
||||
|
||||
- All user input is validated before API calls
|
||||
- Issue titles and descriptions are sanitized
|
||||
- Label names are checked against taxonomy
|
||||
- Repository names are validated
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
**Module not found**:
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate
|
||||
```
|
||||
|
||||
**Configuration not found**:
|
||||
```bash
|
||||
ls -la ~/.config/claude/gitea.env
|
||||
# If missing, create it following installation steps
|
||||
```
|
||||
|
||||
**Authentication failed**:
|
||||
```bash
|
||||
# Test token manually
|
||||
curl -H "Authorization: token YOUR_TOKEN" \
|
||||
https://gitea.example.com/api/v1/user
|
||||
```
|
||||
|
||||
**Permission denied on branch**:
|
||||
```bash
|
||||
# Check current branch
|
||||
git branch --show-current
|
||||
|
||||
# Switch to development branch
|
||||
git checkout development
|
||||
```
|
||||
|
||||
See [TESTING.md](./TESTING.md#troubleshooting) for more details.
|
||||
|
||||
## Development
|
||||
|
||||
### Project Structure
|
||||
|
||||
- `config.py` - Hybrid configuration loader with mode detection
|
||||
- `gitea_client.py` - Synchronous Gitea API client using requests
|
||||
- `tools/issues.py` - Async wrappers with branch detection
|
||||
- `tools/labels.py` - Label management and suggestion
|
||||
- `server.py` - MCP server with JSON-RPC 2.0 over stdio
|
||||
|
||||
### Adding New Tools
|
||||
|
||||
1. Add method to `GiteaClient` (sync)
|
||||
2. Add async wrapper to appropriate tool class
|
||||
3. Register tool in `server.py` `setup_tools()`
|
||||
4. Add unit tests
|
||||
5. Update documentation
|
||||
|
||||
### Testing Philosophy
|
||||
|
||||
- **Unit tests**: Use mocks for fast feedback
|
||||
- **Integration tests**: Use real Gitea API for validation
|
||||
- **Branch detection**: Test all branch types
|
||||
- **Mode detection**: Test both project and company modes
|
||||
|
||||
## Performance
|
||||
|
||||
### Caching
|
||||
|
||||
Labels are cached to reduce API calls:
|
||||
|
||||
```python
|
||||
from functools import lru_cache
|
||||
|
||||
@lru_cache(maxsize=128)
|
||||
def get_labels_cached(self, repo: str):
|
||||
return self.get_labels(repo)
|
||||
```
|
||||
|
||||
### Retry Logic
|
||||
|
||||
API calls include automatic retry with exponential backoff:
|
||||
|
||||
```python
|
||||
@retry_on_failure(max_retries=3, delay=1)
|
||||
def list_issues(self, state='open', labels=None, repo=None):
|
||||
# Implementation
|
||||
```
|
||||
|
||||
## Changelog
|
||||
|
||||
### v1.0.0 (2025-01-06) - Phase 1 Complete
|
||||
|
||||
✅ Initial implementation:
|
||||
- Configuration management (hybrid system + project)
|
||||
- Gitea API client with all CRUD operations
|
||||
- MCP server with 8 tools
|
||||
- Issue tools with branch detection
|
||||
- Label tools with intelligent suggestions
|
||||
- Mode detection (project vs company)
|
||||
- Branch-aware security model
|
||||
- 42 unit tests (100% passing)
|
||||
- Comprehensive documentation
|
||||
|
||||
## License
|
||||
|
||||
Part of the Bandit Labs Claude Code Plugins project.
|
||||
|
||||
## Related Documentation
|
||||
|
||||
- **MCP Specification**: `docs/references/MCP-GITEA.md`
|
||||
- **Project Summary**: `docs/references/PROJECT-SUMMARY.md`
|
||||
- **Implementation Plan**: `docs/reference-material/projman-implementation-plan.md`
|
||||
- **Testing Guide**: `TESTING.md`
|
||||
|
||||
## Support
|
||||
|
||||
For issues or questions:
|
||||
1. Check [TESTING.md](./TESTING.md) troubleshooting section
|
||||
2. Review [MCP-GITEA.md](../../docs/references/MCP-GITEA.md) specification
|
||||
3. Create an issue in the project repository
|
||||
|
||||
---
|
||||
|
||||
**Built for**: Bandit Labs Project Management Plugins
|
||||
**Phase**: 1 (Complete)
|
||||
**Status**: ✅ Production Ready
|
||||
**Last Updated**: 2025-01-06
|
||||
@@ -1,582 +0,0 @@
|
||||
# Gitea MCP Server - Testing Guide
|
||||
|
||||
This document provides comprehensive testing instructions for the Gitea MCP Server implementation.
|
||||
|
||||
## Table of Contents
|
||||
|
||||
1. [Unit Tests](#unit-tests)
|
||||
2. [Manual MCP Server Testing](#manual-mcp-server-testing)
|
||||
3. [Integration Testing](#integration-testing)
|
||||
4. [Configuration Setup for Testing](#configuration-setup-for-testing)
|
||||
5. [Troubleshooting](#troubleshooting)
|
||||
|
||||
---
|
||||
|
||||
## Unit Tests
|
||||
|
||||
Unit tests use mocks to test all modules without requiring a real Gitea instance.
|
||||
|
||||
### Prerequisites
|
||||
|
||||
Ensure the virtual environment is activated and dependencies are installed:
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate # Linux/Mac
|
||||
# or .venv\Scripts\activate # Windows
|
||||
```
|
||||
|
||||
### Running All Tests
|
||||
|
||||
Run all 42 unit tests:
|
||||
|
||||
```bash
|
||||
pytest tests/ -v
|
||||
```
|
||||
|
||||
Expected output:
|
||||
```
|
||||
============================== 42 passed in 0.57s ==============================
|
||||
```
|
||||
|
||||
### Running Specific Test Files
|
||||
|
||||
Run tests for a specific module:
|
||||
|
||||
```bash
|
||||
# Configuration tests
|
||||
pytest tests/test_config.py -v
|
||||
|
||||
# Gitea client tests
|
||||
pytest tests/test_gitea_client.py -v
|
||||
|
||||
# Issue tools tests
|
||||
pytest tests/test_issues.py -v
|
||||
|
||||
# Label tools tests
|
||||
pytest tests/test_labels.py -v
|
||||
```
|
||||
|
||||
### Running Specific Tests
|
||||
|
||||
Run a single test:
|
||||
|
||||
```bash
|
||||
pytest tests/test_config.py::test_load_system_config -v
|
||||
```
|
||||
|
||||
### Test Coverage
|
||||
|
||||
Generate coverage report:
|
||||
|
||||
```bash
|
||||
pytest --cov=mcp_server --cov-report=html tests/
|
||||
|
||||
# View coverage report
|
||||
# Open htmlcov/index.html in your browser
|
||||
```
|
||||
|
||||
Expected coverage: >80% for all modules
|
||||
|
||||
### Test Organization
|
||||
|
||||
**Configuration Tests** (`test_config.py`):
|
||||
- System-level configuration loading
|
||||
- Project-level configuration override
|
||||
- Mode detection (project vs company)
|
||||
- Missing configuration handling
|
||||
|
||||
**Gitea Client Tests** (`test_gitea_client.py`):
|
||||
- API client initialization
|
||||
- Issue CRUD operations
|
||||
- Label retrieval
|
||||
- PMO multi-repo operations
|
||||
|
||||
**Issue Tools Tests** (`test_issues.py`):
|
||||
- Branch-aware security checks
|
||||
- Async wrappers for sync client
|
||||
- Permission enforcement
|
||||
- PMO aggregation mode
|
||||
|
||||
**Label Tools Tests** (`test_labels.py`):
|
||||
- Label retrieval (org + repo)
|
||||
- Intelligent label suggestion
|
||||
- Multi-category detection
|
||||
|
||||
---
|
||||
|
||||
## Manual MCP Server Testing
|
||||
|
||||
Test the MCP server manually using stdio communication.
|
||||
|
||||
### Step 1: Start the MCP Server
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate
|
||||
python -m mcp_server.server
|
||||
```
|
||||
|
||||
The server will start and wait for JSON-RPC 2.0 messages on stdin.
|
||||
|
||||
### Step 2: Test Tool Listing
|
||||
|
||||
In another terminal, send a tool listing request:
|
||||
|
||||
```bash
|
||||
echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' | python -m mcp_server.server
|
||||
```
|
||||
|
||||
Expected response:
|
||||
```json
|
||||
{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 1,
|
||||
"result": {
|
||||
"tools": [
|
||||
{"name": "list_issues", "description": "List issues from Gitea repository", ...},
|
||||
{"name": "get_issue", "description": "Get specific issue details", ...},
|
||||
{"name": "create_issue", "description": "Create a new issue in Gitea", ...},
|
||||
...
|
||||
]
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Step 3: Test Tool Invocation
|
||||
|
||||
**Note:** Manual tool invocation requires proper configuration. See [Configuration Setup](#configuration-setup-for-testing).
|
||||
|
||||
Example: List issues
|
||||
```bash
|
||||
echo '{
|
||||
"jsonrpc": "2.0",
|
||||
"id": 2,
|
||||
"method": "tools/call",
|
||||
"params": {
|
||||
"name": "list_issues",
|
||||
"arguments": {
|
||||
"state": "open"
|
||||
}
|
||||
}
|
||||
}' | python -m mcp_server.server
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Integration Testing
|
||||
|
||||
Test the MCP server with a real Gitea instance.
|
||||
|
||||
### Prerequisites
|
||||
|
||||
1. **Gitea Instance**: Access to https://gitea.example.com (or your Gitea instance)
|
||||
2. **API Token**: Personal access token with required permissions
|
||||
3. **Configuration**: Properly configured system and project configs
|
||||
|
||||
### Step 1: Configuration Setup
|
||||
|
||||
Create system-level configuration:
|
||||
|
||||
```bash
|
||||
mkdir -p ~/.config/claude
|
||||
|
||||
cat > ~/.config/claude/gitea.env << EOF
|
||||
GITEA_API_URL=https://gitea.example.com/api/v1
|
||||
GITEA_API_TOKEN=your_gitea_token_here
|
||||
GITEA_OWNER=bandit
|
||||
EOF
|
||||
|
||||
chmod 600 ~/.config/claude/gitea.env
|
||||
```
|
||||
|
||||
Create project-level configuration (for project mode testing):
|
||||
|
||||
```bash
|
||||
cd /path/to/test/project
|
||||
|
||||
cat > .env << EOF
|
||||
GITEA_REPO=test-repo
|
||||
EOF
|
||||
|
||||
# Add to .gitignore
|
||||
echo ".env" >> .gitignore
|
||||
```
|
||||
|
||||
### Step 2: Generate Gitea API Token
|
||||
|
||||
1. Log into Gitea: https://gitea.example.com
|
||||
2. Navigate to: **Settings** → **Applications** → **Manage Access Tokens**
|
||||
3. Click **Generate New Token**
|
||||
4. Token configuration:
|
||||
- **Token Name:** `mcp-integration-test`
|
||||
- **Required Permissions:**
|
||||
- ✅ `repo` (all) - Read/write access to repositories, issues, labels
|
||||
- ✅ `read:org` - Read organization information and labels
|
||||
- ✅ `read:user` - Read user information
|
||||
5. Click **Generate Token**
|
||||
6. Copy the token immediately (shown only once)
|
||||
7. Add to `~/.config/claude/gitea.env`
|
||||
|
||||
### Step 3: Verify Configuration
|
||||
|
||||
Test configuration loading:
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate
|
||||
python -c "
|
||||
from mcp_server.config import GiteaConfig
|
||||
config = GiteaConfig()
|
||||
result = config.load()
|
||||
print(f'API URL: {result[\"api_url\"]}')
|
||||
print(f'Owner: {result[\"owner\"]}')
|
||||
print(f'Repo: {result[\"repo\"]}')
|
||||
print(f'Mode: {result[\"mode\"]}')
|
||||
"
|
||||
```
|
||||
|
||||
Expected output:
|
||||
```
|
||||
API URL: https://gitea.example.com/api/v1
|
||||
Owner: bandit
|
||||
Repo: test-repo (or None for company mode)
|
||||
Mode: project (or company)
|
||||
```
|
||||
|
||||
### Step 4: Test Gitea Client
|
||||
|
||||
Test basic Gitea API operations:
|
||||
|
||||
```bash
|
||||
python -c "
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
client = GiteaClient()
|
||||
|
||||
# Test listing issues
|
||||
print('Testing list_issues...')
|
||||
issues = client.list_issues(state='open')
|
||||
print(f'Found {len(issues)} open issues')
|
||||
|
||||
# Test getting labels
|
||||
print('\\nTesting get_labels...')
|
||||
labels = client.get_labels()
|
||||
print(f'Found {len(labels)} repository labels')
|
||||
|
||||
# Test getting org labels
|
||||
print('\\nTesting get_org_labels...')
|
||||
org_labels = client.get_org_labels()
|
||||
print(f'Found {len(org_labels)} organization labels')
|
||||
|
||||
print('\\n✅ All integration tests passed!')
|
||||
"
|
||||
```
|
||||
|
||||
### Step 5: Test Issue Creation (Optional)
|
||||
|
||||
**Warning:** This creates a real issue in Gitea. Use a test repository.
|
||||
|
||||
```bash
|
||||
python -c "
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
client = GiteaClient()
|
||||
|
||||
# Create test issue
|
||||
print('Creating test issue...')
|
||||
issue = client.create_issue(
|
||||
title='[TEST] MCP Server Integration Test',
|
||||
body='This is a test issue created by the Gitea MCP Server integration tests.',
|
||||
labels=['Type/Test']
|
||||
)
|
||||
print(f'Created issue #{issue[\"number\"]}: {issue[\"title\"]}')
|
||||
|
||||
# Clean up: Close the issue
|
||||
print('\\nClosing test issue...')
|
||||
client.update_issue(issue['number'], state='closed')
|
||||
print('✅ Test issue closed')
|
||||
"
|
||||
```
|
||||
|
||||
### Step 6: Test MCP Server with Real API
|
||||
|
||||
Start the MCP server and test with real Gitea API:
|
||||
|
||||
```bash
|
||||
cd mcp-servers/gitea
|
||||
source .venv/bin/activate
|
||||
|
||||
# Run server with test script
|
||||
python << 'EOF'
|
||||
import asyncio
|
||||
import json
|
||||
from mcp_server.server import GiteaMCPServer
|
||||
|
||||
async def test_server():
|
||||
server = GiteaMCPServer()
|
||||
await server.initialize()
|
||||
|
||||
# Test list_issues
|
||||
result = await server.issue_tools.list_issues(state='open')
|
||||
print(f'Found {len(result)} open issues')
|
||||
|
||||
# Test get_labels
|
||||
labels = await server.label_tools.get_labels()
|
||||
print(f'Found {labels["total_count"]} total labels')
|
||||
|
||||
# Test suggest_labels
|
||||
suggestions = await server.label_tools.suggest_labels(
|
||||
"Fix critical bug in authentication"
|
||||
)
|
||||
print(f'Suggested labels: {", ".join(suggestions)}')
|
||||
|
||||
print('✅ All MCP server integration tests passed!')
|
||||
|
||||
asyncio.run(test_server())
|
||||
EOF
|
||||
```
|
||||
|
||||
### Step 7: Test PMO Mode (Optional)
|
||||
|
||||
Test company-wide mode (no GITEA_REPO):
|
||||
|
||||
```bash
|
||||
# Temporarily remove GITEA_REPO
|
||||
unset GITEA_REPO
|
||||
|
||||
python -c "
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
client = GiteaClient()
|
||||
|
||||
print(f'Running in {client.mode} mode')
|
||||
|
||||
# Test list_repos
|
||||
print('\\nTesting list_repos...')
|
||||
repos = client.list_repos()
|
||||
print(f'Found {len(repos)} repositories')
|
||||
|
||||
# Test aggregate_issues
|
||||
print('\\nTesting aggregate_issues...')
|
||||
aggregated = client.aggregate_issues(state='open')
|
||||
for repo_name, issues in aggregated.items():
|
||||
print(f' {repo_name}: {len(issues)} open issues')
|
||||
|
||||
print('\\n✅ PMO mode tests passed!')
|
||||
"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Configuration Setup for Testing
|
||||
|
||||
### Minimal Configuration
|
||||
|
||||
**System-level** (`~/.config/claude/gitea.env`):
|
||||
```bash
|
||||
GITEA_API_URL=https://gitea.example.com/api/v1
|
||||
GITEA_API_TOKEN=your_token_here
|
||||
GITEA_OWNER=bandit
|
||||
```
|
||||
|
||||
**Project-level** (`.env` in project root):
|
||||
```bash
|
||||
# For project mode
|
||||
GITEA_REPO=test-repo
|
||||
|
||||
# For company mode (PMO), omit GITEA_REPO
|
||||
```
|
||||
|
||||
### Verification
|
||||
|
||||
Verify configuration is correct:
|
||||
|
||||
```bash
|
||||
# Check system config exists
|
||||
ls -la ~/.config/claude/gitea.env
|
||||
|
||||
# Check permissions (should be 600)
|
||||
stat -c "%a %n" ~/.config/claude/gitea.env
|
||||
|
||||
# Check content (without exposing token)
|
||||
grep -v TOKEN ~/.config/claude/gitea.env
|
||||
|
||||
# Check project config (if using project mode)
|
||||
cat .env
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Common Issues
|
||||
|
||||
#### 1. Import Errors
|
||||
|
||||
**Error:**
|
||||
```
|
||||
ModuleNotFoundError: No module named 'mcp_server'
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Ensure you're in the correct directory
|
||||
cd mcp-servers/gitea
|
||||
|
||||
# Activate virtual environment
|
||||
source .venv/bin/activate
|
||||
|
||||
# Verify installation
|
||||
pip list | grep mcp
|
||||
```
|
||||
|
||||
#### 2. Configuration Not Found
|
||||
|
||||
**Error:**
|
||||
```
|
||||
FileNotFoundError: System config not found: /home/user/.config/claude/gitea.env
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Create system config
|
||||
mkdir -p ~/.config/claude
|
||||
cat > ~/.config/claude/gitea.env << EOF
|
||||
GITEA_API_URL=https://gitea.example.com/api/v1
|
||||
GITEA_API_TOKEN=your_token_here
|
||||
GITEA_OWNER=bandit
|
||||
EOF
|
||||
|
||||
chmod 600 ~/.config/claude/gitea.env
|
||||
```
|
||||
|
||||
#### 3. Missing Required Configuration
|
||||
|
||||
**Error:**
|
||||
```
|
||||
ValueError: Missing required configuration: GITEA_API_TOKEN, GITEA_OWNER
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Check configuration file
|
||||
cat ~/.config/claude/gitea.env
|
||||
|
||||
# Ensure all required variables are present:
|
||||
# - GITEA_API_URL
|
||||
# - GITEA_API_TOKEN
|
||||
# - GITEA_OWNER
|
||||
```
|
||||
|
||||
#### 4. API Authentication Failed
|
||||
|
||||
**Error:**
|
||||
```
|
||||
requests.exceptions.HTTPError: 401 Client Error: Unauthorized
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Test token manually
|
||||
curl -H "Authorization: token YOUR_TOKEN" \
|
||||
https://gitea.example.com/api/v1/user
|
||||
|
||||
# If fails, regenerate token in Gitea settings
|
||||
```
|
||||
|
||||
#### 5. Permission Errors (Branch Detection)
|
||||
|
||||
**Error:**
|
||||
```
|
||||
PermissionError: Cannot create issues on branch 'main'
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Check current branch
|
||||
git branch --show-current
|
||||
|
||||
# Switch to development branch
|
||||
git checkout development
|
||||
# or
|
||||
git checkout -b feat/test-feature
|
||||
```
|
||||
|
||||
#### 6. Repository Not Specified
|
||||
|
||||
**Error:**
|
||||
```
|
||||
ValueError: Repository not specified
|
||||
```
|
||||
|
||||
**Solution:**
|
||||
```bash
|
||||
# Add GITEA_REPO to project config
|
||||
echo "GITEA_REPO=your-repo-name" >> .env
|
||||
|
||||
# Or specify repo in tool call
|
||||
# (for PMO mode multi-repo operations)
|
||||
```
|
||||
|
||||
### Debug Mode
|
||||
|
||||
Enable debug logging:
|
||||
|
||||
```bash
|
||||
export LOG_LEVEL=DEBUG
|
||||
python -m mcp_server.server
|
||||
```
|
||||
|
||||
### Test Summary
|
||||
|
||||
After completing all tests, verify:
|
||||
|
||||
- ✅ All 42 unit tests pass
|
||||
- ✅ MCP server starts without errors
|
||||
- ✅ Configuration loads correctly
|
||||
- ✅ Gitea API client connects successfully
|
||||
- ✅ Issues can be listed from Gitea
|
||||
- ✅ Labels can be retrieved
|
||||
- ✅ Label suggestions work correctly
|
||||
- ✅ Branch detection blocks writes on main/staging
|
||||
- ✅ Mode detection works (project vs company)
|
||||
|
||||
---
|
||||
|
||||
## Success Criteria
|
||||
|
||||
Phase 1 is complete when:
|
||||
|
||||
1. **All unit tests pass** (42/42)
|
||||
2. **MCP server starts without errors**
|
||||
3. **Can list issues from Gitea**
|
||||
4. **Can create issues with labels** (in development mode)
|
||||
5. **Mode detection works** (project vs company)
|
||||
6. **Branch detection prevents writes on main/staging**
|
||||
7. **Configuration properly merges** system + project levels
|
||||
|
||||
---
|
||||
|
||||
## Next Steps
|
||||
|
||||
After completing testing:
|
||||
|
||||
1. **Document any issues** found during testing
|
||||
2. **Create integration with projman plugin** (Phase 2)
|
||||
3. **Test in real project workflow** (Phase 5)
|
||||
4. **Performance optimization** (if needed)
|
||||
5. **Production hardening** (Phase 8)
|
||||
|
||||
---
|
||||
|
||||
## Additional Resources
|
||||
|
||||
- **MCP Documentation**: https://docs.anthropic.com/claude/docs/mcp
|
||||
- **Gitea API Documentation**: https://docs.gitea.io/en-us/api-usage/
|
||||
- **Project Documentation**: `docs/references/MCP-GITEA.md`
|
||||
- **Implementation Plan**: `docs/references/PROJECT-SUMMARY.md`
|
||||
|
||||
---
|
||||
|
||||
**Last Updated**: 2025-01-06 (Phase 1 Implementation)
|
||||
@@ -1,98 +0,0 @@
|
||||
"""
|
||||
Configuration loader for Gitea MCP Server.
|
||||
|
||||
Implements hybrid configuration system:
|
||||
- System-level: ~/.config/claude/gitea.env (credentials)
|
||||
- Project-level: .env (repository specification)
|
||||
"""
|
||||
from pathlib import Path
|
||||
from dotenv import load_dotenv
|
||||
import os
|
||||
import logging
|
||||
from typing import Dict, Optional
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GiteaConfig:
|
||||
"""Hybrid configuration loader with mode detection"""
|
||||
|
||||
def __init__(self):
|
||||
self.api_url: Optional[str] = None
|
||||
self.api_token: Optional[str] = None
|
||||
self.repo: Optional[str] = None
|
||||
self.mode: str = 'project'
|
||||
|
||||
def load(self) -> Dict[str, Optional[str]]:
|
||||
"""
|
||||
Load configuration from system and project levels.
|
||||
Project-level configuration overrides system-level.
|
||||
|
||||
Returns:
|
||||
Dict containing api_url, api_token, repo, mode
|
||||
|
||||
Raises:
|
||||
FileNotFoundError: If system config is missing
|
||||
ValueError: If required configuration is missing
|
||||
"""
|
||||
# Load system config
|
||||
system_config = Path.home() / '.config' / 'claude' / 'gitea.env'
|
||||
if system_config.exists():
|
||||
load_dotenv(system_config)
|
||||
logger.info(f"Loaded system configuration from {system_config}")
|
||||
else:
|
||||
raise FileNotFoundError(
|
||||
f"System config not found: {system_config}\n"
|
||||
"Create it with: mkdir -p ~/.config/claude && "
|
||||
"cat > ~/.config/claude/gitea.env"
|
||||
)
|
||||
|
||||
# Load project config (overrides system)
|
||||
project_config = Path.cwd() / '.env'
|
||||
if project_config.exists():
|
||||
load_dotenv(project_config, override=True)
|
||||
logger.info(f"Loaded project configuration from {project_config}")
|
||||
|
||||
# Extract values
|
||||
self.api_url = os.getenv('GITEA_API_URL')
|
||||
self.api_token = os.getenv('GITEA_API_TOKEN')
|
||||
self.repo = os.getenv('GITEA_REPO') # Optional, must be owner/repo format
|
||||
|
||||
# Detect mode
|
||||
if self.repo:
|
||||
self.mode = 'project'
|
||||
logger.info(f"Running in project mode: {self.repo}")
|
||||
else:
|
||||
self.mode = 'company'
|
||||
logger.info("Running in company-wide mode (PMO)")
|
||||
|
||||
# Validate required variables
|
||||
self._validate()
|
||||
|
||||
return {
|
||||
'api_url': self.api_url,
|
||||
'api_token': self.api_token,
|
||||
'repo': self.repo,
|
||||
'mode': self.mode
|
||||
}
|
||||
|
||||
def _validate(self) -> None:
|
||||
"""
|
||||
Validate that required configuration is present.
|
||||
|
||||
Raises:
|
||||
ValueError: If required configuration is missing
|
||||
"""
|
||||
required = {
|
||||
'GITEA_API_URL': self.api_url,
|
||||
'GITEA_API_TOKEN': self.api_token
|
||||
}
|
||||
|
||||
missing = [key for key, value in required.items() if not value]
|
||||
|
||||
if missing:
|
||||
raise ValueError(
|
||||
f"Missing required configuration: {', '.join(missing)}\n"
|
||||
"Check your ~/.config/claude/gitea.env file"
|
||||
)
|
||||
@@ -1,211 +0,0 @@
|
||||
"""
|
||||
Gitea API client for interacting with Gitea API.
|
||||
|
||||
Provides synchronous methods for:
|
||||
- Issue CRUD operations
|
||||
- Label management
|
||||
- Repository operations
|
||||
- PMO multi-repo aggregation
|
||||
"""
|
||||
import requests
|
||||
import logging
|
||||
from typing import List, Dict, Optional
|
||||
from .config import GiteaConfig
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GiteaClient:
|
||||
"""Client for interacting with Gitea API"""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize Gitea client with configuration"""
|
||||
config = GiteaConfig()
|
||||
config_dict = config.load()
|
||||
|
||||
self.base_url = config_dict['api_url']
|
||||
self.token = config_dict['api_token']
|
||||
self.repo = config_dict.get('repo') # Optional default repo in owner/repo format
|
||||
self.mode = config_dict['mode']
|
||||
|
||||
self.session = requests.Session()
|
||||
self.session.headers.update({
|
||||
'Authorization': f'token {self.token}',
|
||||
'Content-Type': 'application/json'
|
||||
})
|
||||
|
||||
logger.info(f"Gitea client initialized in {self.mode} mode")
|
||||
|
||||
def _parse_repo(self, repo: Optional[str] = None) -> tuple:
|
||||
"""Parse owner/repo from input. Always requires 'owner/repo' format."""
|
||||
target = repo or self.repo
|
||||
if not target or '/' not in target:
|
||||
raise ValueError("Use 'owner/repo' format (e.g. 'bandit/support-claude-mktplace')")
|
||||
parts = target.split('/', 1)
|
||||
return parts[0], parts[1]
|
||||
|
||||
def list_issues(
|
||||
self,
|
||||
state: str = 'open',
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
List issues from Gitea repository.
|
||||
|
||||
Args:
|
||||
state: Issue state (open, closed, all)
|
||||
labels: Filter by labels
|
||||
repo: Repository in 'owner/repo' format
|
||||
|
||||
Returns:
|
||||
List of issue dictionaries
|
||||
"""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/issues"
|
||||
params = {'state': state}
|
||||
if labels:
|
||||
params['labels'] = ','.join(labels)
|
||||
logger.info(f"Listing issues from {owner}/{target_repo} with state={state}")
|
||||
response = self.session.get(url, params=params)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_issue(
|
||||
self,
|
||||
issue_number: int,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Get specific issue details."""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/issues/{issue_number}"
|
||||
logger.info(f"Getting issue #{issue_number} from {owner}/{target_repo}")
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def create_issue(
|
||||
self,
|
||||
title: str,
|
||||
body: str,
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Create a new issue in Gitea."""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/issues"
|
||||
data = {'title': title, 'body': body}
|
||||
if labels:
|
||||
label_ids = self._resolve_label_ids(labels, owner, target_repo)
|
||||
data['labels'] = label_ids
|
||||
logger.info(f"Creating issue in {owner}/{target_repo}: {title}")
|
||||
response = self.session.post(url, json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def _resolve_label_ids(self, label_names: List[str], owner: str, repo: str) -> List[int]:
|
||||
"""Convert label names to label IDs."""
|
||||
org_labels = self.get_org_labels(owner)
|
||||
repo_labels = self.get_labels(f"{owner}/{repo}")
|
||||
all_labels = org_labels + repo_labels
|
||||
label_map = {label['name']: label['id'] for label in all_labels}
|
||||
label_ids = []
|
||||
for name in label_names:
|
||||
if name in label_map:
|
||||
label_ids.append(label_map[name])
|
||||
else:
|
||||
logger.warning(f"Label '{name}' not found, skipping")
|
||||
return label_ids
|
||||
|
||||
def update_issue(
|
||||
self,
|
||||
issue_number: int,
|
||||
title: Optional[str] = None,
|
||||
body: Optional[str] = None,
|
||||
state: Optional[str] = None,
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Update existing issue. Repo must be 'owner/repo' format."""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/issues/{issue_number}"
|
||||
data = {}
|
||||
if title is not None:
|
||||
data['title'] = title
|
||||
if body is not None:
|
||||
data['body'] = body
|
||||
if state is not None:
|
||||
data['state'] = state
|
||||
if labels is not None:
|
||||
data['labels'] = labels
|
||||
logger.info(f"Updating issue #{issue_number} in {owner}/{target_repo}")
|
||||
response = self.session.patch(url, json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def add_comment(
|
||||
self,
|
||||
issue_number: int,
|
||||
comment: str,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""Add comment to issue. Repo must be 'owner/repo' format."""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/issues/{issue_number}/comments"
|
||||
data = {'body': comment}
|
||||
logger.info(f"Adding comment to issue #{issue_number} in {owner}/{target_repo}")
|
||||
response = self.session.post(url, json=data)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_labels(self, repo: Optional[str] = None) -> List[Dict]:
|
||||
"""Get all labels from repository. Repo must be 'owner/repo' format."""
|
||||
owner, target_repo = self._parse_repo(repo)
|
||||
url = f"{self.base_url}/repos/{owner}/{target_repo}/labels"
|
||||
logger.info(f"Getting labels from {owner}/{target_repo}")
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def get_org_labels(self, org: str) -> List[Dict]:
|
||||
"""Get organization-level labels. Org is the organization name."""
|
||||
url = f"{self.base_url}/orgs/{org}/labels"
|
||||
logger.info(f"Getting organization labels for {org}")
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def list_repos(self, org: str) -> List[Dict]:
|
||||
"""List all repositories in organization. Org is the organization name."""
|
||||
url = f"{self.base_url}/orgs/{org}/repos"
|
||||
logger.info(f"Listing all repositories for organization {org}")
|
||||
response = self.session.get(url)
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
|
||||
def aggregate_issues(
|
||||
self,
|
||||
org: str,
|
||||
state: str = 'open',
|
||||
labels: Optional[List[str]] = None
|
||||
) -> Dict[str, List[Dict]]:
|
||||
"""Fetch issues across all repositories in org."""
|
||||
repos = self.list_repos(org)
|
||||
aggregated = {}
|
||||
logger.info(f"Aggregating issues across {len(repos)} repositories")
|
||||
for repo in repos:
|
||||
repo_name = repo['name']
|
||||
try:
|
||||
issues = self.list_issues(
|
||||
state=state,
|
||||
labels=labels,
|
||||
repo=f"{org}/{repo_name}"
|
||||
)
|
||||
if issues:
|
||||
aggregated[repo_name] = issues
|
||||
logger.info(f"Found {len(issues)} issues in {repo_name}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error fetching issues from {repo_name}: {e}")
|
||||
|
||||
return aggregated
|
||||
@@ -1,308 +0,0 @@
|
||||
"""
|
||||
MCP Server entry point for Gitea integration.
|
||||
|
||||
Provides Gitea tools to Claude Code via JSON-RPC 2.0 over stdio.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
import json
|
||||
from mcp.server import Server
|
||||
from mcp.server.stdio import stdio_server
|
||||
from mcp.types import Tool, TextContent
|
||||
|
||||
from .config import GiteaConfig
|
||||
from .gitea_client import GiteaClient
|
||||
from .tools.issues import IssueTools
|
||||
from .tools.labels import LabelTools
|
||||
|
||||
# Suppress noisy MCP validation warnings on stderr
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logging.getLogger("root").setLevel(logging.ERROR)
|
||||
logging.getLogger("mcp").setLevel(logging.ERROR)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class GiteaMCPServer:
|
||||
"""MCP Server for Gitea integration"""
|
||||
|
||||
def __init__(self):
|
||||
self.server = Server("gitea-mcp")
|
||||
self.config = None
|
||||
self.client = None
|
||||
self.issue_tools = None
|
||||
self.label_tools = None
|
||||
|
||||
async def initialize(self):
|
||||
"""
|
||||
Initialize server and load configuration.
|
||||
|
||||
Raises:
|
||||
Exception: If initialization fails
|
||||
"""
|
||||
try:
|
||||
config_loader = GiteaConfig()
|
||||
self.config = config_loader.load()
|
||||
|
||||
self.client = GiteaClient()
|
||||
self.issue_tools = IssueTools(self.client)
|
||||
self.label_tools = LabelTools(self.client)
|
||||
|
||||
logger.info(f"Gitea MCP Server initialized in {self.config['mode']} mode")
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to initialize: {e}")
|
||||
raise
|
||||
|
||||
def setup_tools(self):
|
||||
"""Register all available tools with the MCP server"""
|
||||
|
||||
@self.server.list_tools()
|
||||
async def list_tools() -> list[Tool]:
|
||||
"""Return list of available tools"""
|
||||
return [
|
||||
Tool(
|
||||
name="list_issues",
|
||||
description="List issues from Gitea repository",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"state": {
|
||||
"type": "string",
|
||||
"enum": ["open", "closed", "all"],
|
||||
"default": "open",
|
||||
"description": "Issue state filter"
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Filter by labels"
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
}
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="get_issue",
|
||||
description="Get specific issue details",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"issue_number": {
|
||||
"type": "integer",
|
||||
"description": "Issue number"
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
},
|
||||
"required": ["issue_number"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="create_issue",
|
||||
description="Create a new issue in Gitea",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Issue title"
|
||||
},
|
||||
"body": {
|
||||
"type": "string",
|
||||
"description": "Issue description"
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of label names"
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
},
|
||||
"required": ["title", "body"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="update_issue",
|
||||
description="Update existing issue",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"issue_number": {
|
||||
"type": "integer",
|
||||
"description": "Issue number"
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "New title"
|
||||
},
|
||||
"body": {
|
||||
"type": "string",
|
||||
"description": "New body"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"enum": ["open", "closed"],
|
||||
"description": "New state"
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "New labels"
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
},
|
||||
"required": ["issue_number"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="add_comment",
|
||||
description="Add comment to issue",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"issue_number": {
|
||||
"type": "integer",
|
||||
"description": "Issue number"
|
||||
},
|
||||
"comment": {
|
||||
"type": "string",
|
||||
"description": "Comment text"
|
||||
},
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
},
|
||||
"required": ["issue_number", "comment"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="get_labels",
|
||||
description="Get all available labels (org + repo)",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"repo": {
|
||||
"type": "string",
|
||||
"description": "Repository name (for PMO mode)"
|
||||
}
|
||||
}
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="suggest_labels",
|
||||
description="Analyze context and suggest appropriate labels",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"context": {
|
||||
"type": "string",
|
||||
"description": "Issue title + description or sprint context"
|
||||
}
|
||||
},
|
||||
"required": ["context"]
|
||||
}
|
||||
),
|
||||
Tool(
|
||||
name="aggregate_issues",
|
||||
description="Fetch issues across all repositories (PMO mode)",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"org": {
|
||||
"type": "string",
|
||||
"description": "Organization name (e.g. 'bandit')"
|
||||
},
|
||||
"state": {
|
||||
"type": "string",
|
||||
"enum": ["open", "closed", "all"],
|
||||
"default": "open",
|
||||
"description": "Issue state filter"
|
||||
},
|
||||
"labels": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Filter by labels"
|
||||
}
|
||||
},
|
||||
"required": ["org"]
|
||||
}
|
||||
)
|
||||
]
|
||||
|
||||
@self.server.call_tool()
|
||||
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
||||
"""
|
||||
Handle tool invocation.
|
||||
|
||||
Args:
|
||||
name: Tool name
|
||||
arguments: Tool arguments
|
||||
|
||||
Returns:
|
||||
List of TextContent with results
|
||||
"""
|
||||
try:
|
||||
# Route to appropriate tool handler
|
||||
if name == "list_issues":
|
||||
result = await self.issue_tools.list_issues(**arguments)
|
||||
elif name == "get_issue":
|
||||
result = await self.issue_tools.get_issue(**arguments)
|
||||
elif name == "create_issue":
|
||||
result = await self.issue_tools.create_issue(**arguments)
|
||||
elif name == "update_issue":
|
||||
result = await self.issue_tools.update_issue(**arguments)
|
||||
elif name == "add_comment":
|
||||
result = await self.issue_tools.add_comment(**arguments)
|
||||
elif name == "get_labels":
|
||||
result = await self.label_tools.get_labels(**arguments)
|
||||
elif name == "suggest_labels":
|
||||
result = await self.label_tools.suggest_labels(**arguments)
|
||||
elif name == "aggregate_issues":
|
||||
result = await self.issue_tools.aggregate_issues(**arguments)
|
||||
else:
|
||||
raise ValueError(f"Unknown tool: {name}")
|
||||
|
||||
return [TextContent(
|
||||
type="text",
|
||||
text=json.dumps(result, indent=2)
|
||||
)]
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Tool {name} failed: {e}")
|
||||
return [TextContent(
|
||||
type="text",
|
||||
text=f"Error: {str(e)}"
|
||||
)]
|
||||
|
||||
async def run(self):
|
||||
"""Run the MCP server"""
|
||||
await self.initialize()
|
||||
self.setup_tools()
|
||||
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await self.server.run(
|
||||
read_stream,
|
||||
write_stream,
|
||||
self.server.create_initialization_options()
|
||||
)
|
||||
|
||||
|
||||
async def main():
|
||||
"""Main entry point"""
|
||||
server = GiteaMCPServer()
|
||||
await server.run()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
@@ -1,7 +0,0 @@
|
||||
"""
|
||||
MCP tools for Gitea integration.
|
||||
|
||||
This package provides MCP tool implementations for:
|
||||
- Issue operations (issues.py)
|
||||
- Label management (labels.py)
|
||||
"""
|
||||
@@ -1,261 +0,0 @@
|
||||
"""
|
||||
Issue management tools for MCP server.
|
||||
|
||||
Provides async wrappers for issue CRUD operations with:
|
||||
- Branch-aware security
|
||||
- PMO multi-repo support
|
||||
- Comprehensive error handling
|
||||
"""
|
||||
import asyncio
|
||||
import subprocess
|
||||
import logging
|
||||
from typing import List, Dict, Optional
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IssueTools:
|
||||
"""Async wrappers for Gitea issue operations with branch detection"""
|
||||
|
||||
def __init__(self, gitea_client):
|
||||
"""
|
||||
Initialize issue tools.
|
||||
|
||||
Args:
|
||||
gitea_client: GiteaClient instance
|
||||
"""
|
||||
self.gitea = gitea_client
|
||||
|
||||
def _get_current_branch(self) -> str:
|
||||
"""
|
||||
Get current git branch.
|
||||
|
||||
Returns:
|
||||
Current branch name or 'unknown' if not in a git repo
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
check=True
|
||||
)
|
||||
return result.stdout.strip()
|
||||
except subprocess.CalledProcessError:
|
||||
return "unknown"
|
||||
|
||||
def _check_branch_permissions(self, operation: str) -> bool:
|
||||
"""
|
||||
Check if operation is allowed on current branch.
|
||||
|
||||
Args:
|
||||
operation: Operation name (list_issues, create_issue, etc.)
|
||||
|
||||
Returns:
|
||||
True if operation is allowed, False otherwise
|
||||
"""
|
||||
branch = self._get_current_branch()
|
||||
|
||||
# Production branches (read-only except incidents)
|
||||
if branch in ['main', 'master'] or branch.startswith('prod/'):
|
||||
return operation in ['list_issues', 'get_issue', 'get_labels']
|
||||
|
||||
# Staging branches (read-only for code)
|
||||
if branch == 'staging' or branch.startswith('stage/'):
|
||||
return operation in ['list_issues', 'get_issue', 'get_labels', 'create_issue']
|
||||
|
||||
# Development branches (full access)
|
||||
if branch in ['development', 'develop'] or branch.startswith(('feat/', 'feature/', 'dev/')):
|
||||
return True
|
||||
|
||||
# Unknown branch - be restrictive
|
||||
return False
|
||||
|
||||
async def list_issues(
|
||||
self,
|
||||
state: str = 'open',
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> List[Dict]:
|
||||
"""
|
||||
List issues from repository (async wrapper).
|
||||
|
||||
Args:
|
||||
state: Issue state (open, closed, all)
|
||||
labels: Filter by labels
|
||||
repo: Override configured repo (for PMO multi-repo)
|
||||
|
||||
Returns:
|
||||
List of issue dictionaries
|
||||
|
||||
Raises:
|
||||
PermissionError: If operation not allowed on current branch
|
||||
"""
|
||||
if not self._check_branch_permissions('list_issues'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(
|
||||
f"Cannot list issues on branch '{branch}'. "
|
||||
f"Switch to a development branch."
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.list_issues(state, labels, repo)
|
||||
)
|
||||
|
||||
async def get_issue(
|
||||
self,
|
||||
issue_number: int,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Get specific issue details (async wrapper).
|
||||
|
||||
Args:
|
||||
issue_number: Issue number
|
||||
repo: Override configured repo (for PMO multi-repo)
|
||||
|
||||
Returns:
|
||||
Issue dictionary
|
||||
|
||||
Raises:
|
||||
PermissionError: If operation not allowed on current branch
|
||||
"""
|
||||
if not self._check_branch_permissions('get_issue'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(
|
||||
f"Cannot get issue on branch '{branch}'. "
|
||||
f"Switch to a development branch."
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.get_issue(issue_number, repo)
|
||||
)
|
||||
|
||||
async def create_issue(
|
||||
self,
|
||||
title: str,
|
||||
body: str,
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Create new issue (async wrapper with branch check).
|
||||
|
||||
Args:
|
||||
title: Issue title
|
||||
body: Issue description
|
||||
labels: List of label names
|
||||
repo: Override configured repo (for PMO multi-repo)
|
||||
|
||||
Returns:
|
||||
Created issue dictionary
|
||||
|
||||
Raises:
|
||||
PermissionError: If operation not allowed on current branch
|
||||
"""
|
||||
if not self._check_branch_permissions('create_issue'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(
|
||||
f"Cannot create issues on branch '{branch}'. "
|
||||
f"Switch to a development branch to create issues."
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.create_issue(title, body, labels, repo)
|
||||
)
|
||||
|
||||
async def update_issue(
|
||||
self,
|
||||
issue_number: int,
|
||||
title: Optional[str] = None,
|
||||
body: Optional[str] = None,
|
||||
state: Optional[str] = None,
|
||||
labels: Optional[List[str]] = None,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Update existing issue (async wrapper with branch check).
|
||||
|
||||
Args:
|
||||
issue_number: Issue number
|
||||
title: New title (optional)
|
||||
body: New body (optional)
|
||||
state: New state - 'open' or 'closed' (optional)
|
||||
labels: New labels (optional)
|
||||
repo: Override configured repo (for PMO multi-repo)
|
||||
|
||||
Returns:
|
||||
Updated issue dictionary
|
||||
|
||||
Raises:
|
||||
PermissionError: If operation not allowed on current branch
|
||||
"""
|
||||
if not self._check_branch_permissions('update_issue'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(
|
||||
f"Cannot update issues on branch '{branch}'. "
|
||||
f"Switch to a development branch to update issues."
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.update_issue(issue_number, title, body, state, labels, repo)
|
||||
)
|
||||
|
||||
async def add_comment(
|
||||
self,
|
||||
issue_number: int,
|
||||
comment: str,
|
||||
repo: Optional[str] = None
|
||||
) -> Dict:
|
||||
"""
|
||||
Add comment to issue (async wrapper with branch check).
|
||||
|
||||
Args:
|
||||
issue_number: Issue number
|
||||
comment: Comment text
|
||||
repo: Override configured repo (for PMO multi-repo)
|
||||
|
||||
Returns:
|
||||
Created comment dictionary
|
||||
|
||||
Raises:
|
||||
PermissionError: If operation not allowed on current branch
|
||||
"""
|
||||
if not self._check_branch_permissions('add_comment'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(
|
||||
f"Cannot add comments on branch '{branch}'. "
|
||||
f"Switch to a development branch to add comments."
|
||||
)
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.add_comment(issue_number, comment, repo)
|
||||
)
|
||||
|
||||
async def aggregate_issues(
|
||||
self,
|
||||
org: str,
|
||||
state: str = 'open',
|
||||
labels: Optional[List[str]] = None
|
||||
) -> Dict[str, List[Dict]]:
|
||||
"""Aggregate issues across all repositories in org."""
|
||||
if not self._check_branch_permissions('aggregate_issues'):
|
||||
branch = self._get_current_branch()
|
||||
raise PermissionError(f"Cannot aggregate issues on branch '{branch}'.")
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
return await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.aggregate_issues(org, state, labels)
|
||||
)
|
||||
@@ -1,158 +0,0 @@
|
||||
"""
|
||||
Label management tools for MCP server.
|
||||
|
||||
Provides async wrappers for label operations with:
|
||||
- Label taxonomy retrieval
|
||||
- Intelligent label suggestion
|
||||
- Dynamic label detection
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import List, Dict, Optional
|
||||
|
||||
logging.basicConfig(level=logging.INFO)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class LabelTools:
|
||||
"""Async wrappers for Gitea label operations"""
|
||||
|
||||
def __init__(self, gitea_client):
|
||||
"""
|
||||
Initialize label tools.
|
||||
|
||||
Args:
|
||||
gitea_client: GiteaClient instance
|
||||
"""
|
||||
self.gitea = gitea_client
|
||||
|
||||
async def get_labels(self, repo: Optional[str] = None) -> Dict[str, List[Dict]]:
|
||||
"""Get all labels (org + repo). Repo must be 'owner/repo' format."""
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
target_repo = repo or self.gitea.repo
|
||||
if not target_repo or '/' not in target_repo:
|
||||
raise ValueError("Use 'owner/repo' format (e.g. 'bandit/support-claude-mktplace')")
|
||||
|
||||
org = target_repo.split('/')[0]
|
||||
|
||||
org_labels = await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.get_org_labels(org)
|
||||
)
|
||||
|
||||
repo_labels = await loop.run_in_executor(
|
||||
None,
|
||||
lambda: self.gitea.get_labels(target_repo)
|
||||
)
|
||||
|
||||
return {
|
||||
'organization': org_labels,
|
||||
'repository': repo_labels,
|
||||
'total_count': len(org_labels) + len(repo_labels)
|
||||
}
|
||||
|
||||
async def suggest_labels(self, context: str) -> List[str]:
|
||||
"""
|
||||
Analyze context and suggest appropriate labels.
|
||||
|
||||
Args:
|
||||
context: Issue title + description or sprint context
|
||||
|
||||
Returns:
|
||||
List of suggested label names
|
||||
"""
|
||||
suggested = []
|
||||
context_lower = context.lower()
|
||||
|
||||
# Type detection (exclusive - only one)
|
||||
if any(word in context_lower for word in ['bug', 'error', 'fix', 'broken', 'crash', 'fail']):
|
||||
suggested.append('Type/Bug')
|
||||
elif any(word in context_lower for word in ['refactor', 'extract', 'restructure', 'architecture', 'service extraction']):
|
||||
suggested.append('Type/Refactor')
|
||||
elif any(word in context_lower for word in ['feature', 'add', 'implement', 'new', 'create']):
|
||||
suggested.append('Type/Feature')
|
||||
elif any(word in context_lower for word in ['docs', 'documentation', 'readme', 'guide']):
|
||||
suggested.append('Type/Documentation')
|
||||
elif any(word in context_lower for word in ['test', 'testing', 'spec', 'coverage']):
|
||||
suggested.append('Type/Test')
|
||||
elif any(word in context_lower for word in ['chore', 'maintenance', 'update', 'upgrade']):
|
||||
suggested.append('Type/Chore')
|
||||
|
||||
# Priority detection
|
||||
if any(word in context_lower for word in ['critical', 'urgent', 'blocker', 'blocking', 'emergency']):
|
||||
suggested.append('Priority/Critical')
|
||||
elif any(word in context_lower for word in ['high', 'important', 'asap', 'soon']):
|
||||
suggested.append('Priority/High')
|
||||
elif any(word in context_lower for word in ['low', 'nice-to-have', 'optional', 'later']):
|
||||
suggested.append('Priority/Low')
|
||||
else:
|
||||
suggested.append('Priority/Medium')
|
||||
|
||||
# Complexity detection
|
||||
if any(word in context_lower for word in ['simple', 'trivial', 'easy', 'quick']):
|
||||
suggested.append('Complexity/Simple')
|
||||
elif any(word in context_lower for word in ['complex', 'difficult', 'challenging', 'intricate']):
|
||||
suggested.append('Complexity/Complex')
|
||||
else:
|
||||
suggested.append('Complexity/Medium')
|
||||
|
||||
# Efforts detection
|
||||
if any(word in context_lower for word in ['xs', 'tiny', '1 hour', '2 hours']):
|
||||
suggested.append('Efforts/XS')
|
||||
elif any(word in context_lower for word in ['small', 's ', '1 day', 'half day']):
|
||||
suggested.append('Efforts/S')
|
||||
elif any(word in context_lower for word in ['medium', 'm ', '2 days', '3 days']):
|
||||
suggested.append('Efforts/M')
|
||||
elif any(word in context_lower for word in ['large', 'l ', '1 week', '5 days']):
|
||||
suggested.append('Efforts/L')
|
||||
elif any(word in context_lower for word in ['xl', 'extra large', '2 weeks', 'sprint']):
|
||||
suggested.append('Efforts/XL')
|
||||
|
||||
# Component detection (based on keywords)
|
||||
component_keywords = {
|
||||
'Component/Backend': ['backend', 'server', 'api', 'database', 'service'],
|
||||
'Component/Frontend': ['frontend', 'ui', 'interface', 'react', 'vue', 'component'],
|
||||
'Component/API': ['api', 'endpoint', 'rest', 'graphql', 'route'],
|
||||
'Component/Database': ['database', 'db', 'sql', 'migration', 'schema', 'postgres'],
|
||||
'Component/Auth': ['auth', 'authentication', 'login', 'oauth', 'token', 'session'],
|
||||
'Component/Deploy': ['deploy', 'deployment', 'docker', 'kubernetes', 'ci/cd'],
|
||||
'Component/Testing': ['test', 'testing', 'spec', 'jest', 'pytest', 'coverage'],
|
||||
'Component/Docs': ['docs', 'documentation', 'readme', 'guide', 'wiki']
|
||||
}
|
||||
|
||||
for label, keywords in component_keywords.items():
|
||||
if any(keyword in context_lower for keyword in keywords):
|
||||
suggested.append(label)
|
||||
|
||||
# Tech stack detection
|
||||
tech_keywords = {
|
||||
'Tech/Python': ['python', 'fastapi', 'django', 'flask', 'pytest'],
|
||||
'Tech/JavaScript': ['javascript', 'js', 'node', 'npm', 'yarn'],
|
||||
'Tech/Docker': ['docker', 'dockerfile', 'container', 'compose'],
|
||||
'Tech/PostgreSQL': ['postgres', 'postgresql', 'psql', 'sql'],
|
||||
'Tech/Redis': ['redis', 'cache', 'session store'],
|
||||
'Tech/Vue': ['vue', 'vuejs', 'nuxt'],
|
||||
'Tech/FastAPI': ['fastapi', 'pydantic', 'starlette']
|
||||
}
|
||||
|
||||
for label, keywords in tech_keywords.items():
|
||||
if any(keyword in context_lower for keyword in keywords):
|
||||
suggested.append(label)
|
||||
|
||||
# Source detection (based on git branch or context)
|
||||
if 'development' in context_lower or 'dev/' in context_lower:
|
||||
suggested.append('Source/Development')
|
||||
elif 'staging' in context_lower or 'stage/' in context_lower:
|
||||
suggested.append('Source/Staging')
|
||||
elif 'production' in context_lower or 'prod' in context_lower:
|
||||
suggested.append('Source/Production')
|
||||
|
||||
# Risk detection
|
||||
if any(word in context_lower for word in ['breaking', 'breaking change', 'major', 'risky']):
|
||||
suggested.append('Risk/High')
|
||||
elif any(word in context_lower for word in ['safe', 'low risk', 'minor']):
|
||||
suggested.append('Risk/Low')
|
||||
|
||||
logger.info(f"Suggested {len(suggested)} labels based on context")
|
||||
return suggested
|
||||
@@ -1,6 +0,0 @@
|
||||
mcp>=0.9.0 # MCP SDK from Anthropic
|
||||
python-dotenv>=1.0.0 # Environment variable loading
|
||||
requests>=2.31.0 # HTTP client for Gitea API
|
||||
pydantic>=2.5.0 # Data validation
|
||||
pytest>=7.4.3 # Testing framework
|
||||
pytest-asyncio>=0.23.0 # Async testing support
|
||||
@@ -1,151 +0,0 @@
|
||||
"""
|
||||
Unit tests for configuration loader.
|
||||
"""
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import os
|
||||
from mcp_server.config import GiteaConfig
|
||||
|
||||
|
||||
def test_load_system_config(tmp_path, monkeypatch):
|
||||
"""Test loading system-level configuration"""
|
||||
# Mock home directory
|
||||
config_dir = tmp_path / '.config' / 'claude'
|
||||
config_dir.mkdir(parents=True)
|
||||
|
||||
config_file = config_dir / 'gitea.env'
|
||||
config_file.write_text(
|
||||
"GITEA_API_URL=https://test.com/api/v1\n"
|
||||
"GITEA_API_TOKEN=test_token\n"
|
||||
"GITEA_OWNER=test_owner\n"
|
||||
)
|
||||
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
config = GiteaConfig()
|
||||
result = config.load()
|
||||
|
||||
assert result['api_url'] == 'https://test.com/api/v1'
|
||||
assert result['api_token'] == 'test_token'
|
||||
assert result['owner'] == 'test_owner'
|
||||
assert result['mode'] == 'company' # No repo specified
|
||||
assert result['repo'] is None
|
||||
|
||||
|
||||
def test_project_config_override(tmp_path, monkeypatch):
|
||||
"""Test that project config overrides system config"""
|
||||
# Set up system config
|
||||
system_config_dir = tmp_path / '.config' / 'claude'
|
||||
system_config_dir.mkdir(parents=True)
|
||||
|
||||
system_config = system_config_dir / 'gitea.env'
|
||||
system_config.write_text(
|
||||
"GITEA_API_URL=https://test.com/api/v1\n"
|
||||
"GITEA_API_TOKEN=test_token\n"
|
||||
"GITEA_OWNER=test_owner\n"
|
||||
)
|
||||
|
||||
# Set up project config
|
||||
project_dir = tmp_path / 'project'
|
||||
project_dir.mkdir()
|
||||
|
||||
project_config = project_dir / '.env'
|
||||
project_config.write_text("GITEA_REPO=test_repo\n")
|
||||
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(project_dir)
|
||||
|
||||
config = GiteaConfig()
|
||||
result = config.load()
|
||||
|
||||
assert result['repo'] == 'test_repo'
|
||||
assert result['mode'] == 'project'
|
||||
|
||||
|
||||
def test_missing_system_config(tmp_path, monkeypatch):
|
||||
"""Test error handling for missing system configuration"""
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
with pytest.raises(FileNotFoundError) as exc_info:
|
||||
config = GiteaConfig()
|
||||
config.load()
|
||||
|
||||
assert "System config not found" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_missing_required_config(tmp_path, monkeypatch):
|
||||
"""Test error handling for missing required variables"""
|
||||
# Clear environment variables
|
||||
for var in ['GITEA_API_URL', 'GITEA_API_TOKEN', 'GITEA_OWNER', 'GITEA_REPO']:
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
|
||||
# Create incomplete config
|
||||
config_dir = tmp_path / '.config' / 'claude'
|
||||
config_dir.mkdir(parents=True)
|
||||
|
||||
config_file = config_dir / 'gitea.env'
|
||||
config_file.write_text(
|
||||
"GITEA_API_URL=https://test.com/api/v1\n"
|
||||
# Missing GITEA_API_TOKEN and GITEA_OWNER
|
||||
)
|
||||
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
config = GiteaConfig()
|
||||
config.load()
|
||||
|
||||
assert "Missing required configuration" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_mode_detection_project(tmp_path, monkeypatch):
|
||||
"""Test mode detection for project mode"""
|
||||
config_dir = tmp_path / '.config' / 'claude'
|
||||
config_dir.mkdir(parents=True)
|
||||
|
||||
config_file = config_dir / 'gitea.env'
|
||||
config_file.write_text(
|
||||
"GITEA_API_URL=https://test.com/api/v1\n"
|
||||
"GITEA_API_TOKEN=test_token\n"
|
||||
"GITEA_OWNER=test_owner\n"
|
||||
"GITEA_REPO=test_repo\n"
|
||||
)
|
||||
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
config = GiteaConfig()
|
||||
result = config.load()
|
||||
|
||||
assert result['mode'] == 'project'
|
||||
assert result['repo'] == 'test_repo'
|
||||
|
||||
|
||||
def test_mode_detection_company(tmp_path, monkeypatch):
|
||||
"""Test mode detection for company mode (PMO)"""
|
||||
# Clear environment variables, especially GITEA_REPO
|
||||
for var in ['GITEA_API_URL', 'GITEA_API_TOKEN', 'GITEA_OWNER', 'GITEA_REPO']:
|
||||
monkeypatch.delenv(var, raising=False)
|
||||
|
||||
config_dir = tmp_path / '.config' / 'claude'
|
||||
config_dir.mkdir(parents=True)
|
||||
|
||||
config_file = config_dir / 'gitea.env'
|
||||
config_file.write_text(
|
||||
"GITEA_API_URL=https://test.com/api/v1\n"
|
||||
"GITEA_API_TOKEN=test_token\n"
|
||||
"GITEA_OWNER=test_owner\n"
|
||||
# No GITEA_REPO
|
||||
)
|
||||
|
||||
monkeypatch.setenv('HOME', str(tmp_path))
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
||||
config = GiteaConfig()
|
||||
result = config.load()
|
||||
|
||||
assert result['mode'] == 'company'
|
||||
assert result['repo'] is None
|
||||
@@ -1,224 +0,0 @@
|
||||
"""
|
||||
Unit tests for Gitea API client.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, MagicMock
|
||||
from mcp_server.gitea_client import GiteaClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config():
|
||||
"""Fixture providing mocked configuration"""
|
||||
with patch('mcp_server.gitea_client.GiteaConfig') as mock_cfg:
|
||||
mock_instance = mock_cfg.return_value
|
||||
mock_instance.load.return_value = {
|
||||
'api_url': 'https://test.com/api/v1',
|
||||
'api_token': 'test_token',
|
||||
'owner': 'test_owner',
|
||||
'repo': 'test_repo',
|
||||
'mode': 'project'
|
||||
}
|
||||
yield mock_cfg
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gitea_client(mock_config):
|
||||
"""Fixture providing GiteaClient instance with mocked config"""
|
||||
return GiteaClient()
|
||||
|
||||
|
||||
def test_client_initialization(gitea_client):
|
||||
"""Test client initializes with correct configuration"""
|
||||
assert gitea_client.base_url == 'https://test.com/api/v1'
|
||||
assert gitea_client.token == 'test_token'
|
||||
assert gitea_client.owner == 'test_owner'
|
||||
assert gitea_client.repo == 'test_repo'
|
||||
assert gitea_client.mode == 'project'
|
||||
assert 'Authorization' in gitea_client.session.headers
|
||||
assert gitea_client.session.headers['Authorization'] == 'token test_token'
|
||||
|
||||
|
||||
def test_list_issues(gitea_client):
|
||||
"""Test listing issues"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = [
|
||||
{'number': 1, 'title': 'Test Issue 1'},
|
||||
{'number': 2, 'title': 'Test Issue 2'}
|
||||
]
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
issues = gitea_client.list_issues(state='open')
|
||||
|
||||
assert len(issues) == 2
|
||||
assert issues[0]['title'] == 'Test Issue 1'
|
||||
gitea_client.session.get.assert_called_once()
|
||||
|
||||
|
||||
def test_list_issues_with_labels(gitea_client):
|
||||
"""Test listing issues with label filter"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = [{'number': 1, 'title': 'Bug Issue'}]
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
issues = gitea_client.list_issues(state='open', labels=['Type/Bug'])
|
||||
|
||||
gitea_client.session.get.assert_called_once()
|
||||
call_args = gitea_client.session.get.call_args
|
||||
assert call_args[1]['params']['labels'] == 'Type/Bug'
|
||||
|
||||
|
||||
def test_get_issue(gitea_client):
|
||||
"""Test getting specific issue"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {'number': 1, 'title': 'Test Issue'}
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
issue = gitea_client.get_issue(1)
|
||||
|
||||
assert issue['number'] == 1
|
||||
assert issue['title'] == 'Test Issue'
|
||||
|
||||
|
||||
def test_create_issue(gitea_client):
|
||||
"""Test creating new issue"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'number': 1,
|
||||
'title': 'New Issue',
|
||||
'body': 'Issue body'
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'post', return_value=mock_response):
|
||||
issue = gitea_client.create_issue(
|
||||
title='New Issue',
|
||||
body='Issue body',
|
||||
labels=['Type/Bug']
|
||||
)
|
||||
|
||||
assert issue['title'] == 'New Issue'
|
||||
gitea_client.session.post.assert_called_once()
|
||||
|
||||
|
||||
def test_update_issue(gitea_client):
|
||||
"""Test updating existing issue"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {
|
||||
'number': 1,
|
||||
'title': 'Updated Issue'
|
||||
}
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'patch', return_value=mock_response):
|
||||
issue = gitea_client.update_issue(
|
||||
issue_number=1,
|
||||
title='Updated Issue'
|
||||
)
|
||||
|
||||
assert issue['title'] == 'Updated Issue'
|
||||
gitea_client.session.patch.assert_called_once()
|
||||
|
||||
|
||||
def test_add_comment(gitea_client):
|
||||
"""Test adding comment to issue"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {'body': 'Test comment'}
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'post', return_value=mock_response):
|
||||
comment = gitea_client.add_comment(1, 'Test comment')
|
||||
|
||||
assert comment['body'] == 'Test comment'
|
||||
gitea_client.session.post.assert_called_once()
|
||||
|
||||
|
||||
def test_get_labels(gitea_client):
|
||||
"""Test getting repository labels"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = [
|
||||
{'name': 'Type/Bug'},
|
||||
{'name': 'Priority/High'}
|
||||
]
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
labels = gitea_client.get_labels()
|
||||
|
||||
assert len(labels) == 2
|
||||
assert labels[0]['name'] == 'Type/Bug'
|
||||
|
||||
|
||||
def test_get_org_labels(gitea_client):
|
||||
"""Test getting organization labels"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = [
|
||||
{'name': 'Type/Bug'},
|
||||
{'name': 'Type/Feature'}
|
||||
]
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
labels = gitea_client.get_org_labels()
|
||||
|
||||
assert len(labels) == 2
|
||||
|
||||
|
||||
def test_list_repos(gitea_client):
|
||||
"""Test listing organization repositories (PMO mode)"""
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = [
|
||||
{'name': 'repo1'},
|
||||
{'name': 'repo2'}
|
||||
]
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch.object(gitea_client.session, 'get', return_value=mock_response):
|
||||
repos = gitea_client.list_repos()
|
||||
|
||||
assert len(repos) == 2
|
||||
assert repos[0]['name'] == 'repo1'
|
||||
|
||||
|
||||
def test_aggregate_issues(gitea_client):
|
||||
"""Test aggregating issues across repositories (PMO mode)"""
|
||||
# Mock list_repos
|
||||
gitea_client.list_repos = Mock(return_value=[
|
||||
{'name': 'repo1'},
|
||||
{'name': 'repo2'}
|
||||
])
|
||||
|
||||
# Mock list_issues
|
||||
gitea_client.list_issues = Mock(side_effect=[
|
||||
[{'number': 1, 'title': 'Issue 1'}], # repo1
|
||||
[{'number': 2, 'title': 'Issue 2'}] # repo2
|
||||
])
|
||||
|
||||
aggregated = gitea_client.aggregate_issues(state='open')
|
||||
|
||||
assert 'repo1' in aggregated
|
||||
assert 'repo2' in aggregated
|
||||
assert len(aggregated['repo1']) == 1
|
||||
assert len(aggregated['repo2']) == 1
|
||||
|
||||
|
||||
def test_no_repo_specified_error(gitea_client):
|
||||
"""Test error when repository not specified"""
|
||||
# Create client without repo
|
||||
with patch('mcp_server.gitea_client.GiteaConfig') as mock_cfg:
|
||||
mock_instance = mock_cfg.return_value
|
||||
mock_instance.load.return_value = {
|
||||
'api_url': 'https://test.com/api/v1',
|
||||
'api_token': 'test_token',
|
||||
'owner': 'test_owner',
|
||||
'repo': None, # No repo
|
||||
'mode': 'company'
|
||||
}
|
||||
client = GiteaClient()
|
||||
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
client.list_issues()
|
||||
|
||||
assert "Repository not specified" in str(exc_info.value)
|
||||
@@ -1,159 +0,0 @@
|
||||
"""
|
||||
Unit tests for issue tools with branch detection.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch, AsyncMock
|
||||
from mcp_server.tools.issues import IssueTools
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_gitea_client():
|
||||
"""Fixture providing mocked Gitea client"""
|
||||
client = Mock()
|
||||
client.mode = 'project'
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def issue_tools(mock_gitea_client):
|
||||
"""Fixture providing IssueTools instance"""
|
||||
return IssueTools(mock_gitea_client)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_issues_development_branch(issue_tools):
|
||||
"""Test listing issues on development branch (allowed)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='feat/test-feature'):
|
||||
issue_tools.gitea.list_issues = Mock(return_value=[{'number': 1}])
|
||||
|
||||
issues = await issue_tools.list_issues(state='open')
|
||||
|
||||
assert len(issues) == 1
|
||||
issue_tools.gitea.list_issues.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_development_branch(issue_tools):
|
||||
"""Test creating issue on development branch (allowed)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='development'):
|
||||
issue_tools.gitea.create_issue = Mock(return_value={'number': 1})
|
||||
|
||||
issue = await issue_tools.create_issue('Test', 'Body')
|
||||
|
||||
assert issue['number'] == 1
|
||||
issue_tools.gitea.create_issue.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_main_branch_blocked(issue_tools):
|
||||
"""Test creating issue on main branch (blocked)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='main'):
|
||||
with pytest.raises(PermissionError) as exc_info:
|
||||
await issue_tools.create_issue('Test', 'Body')
|
||||
|
||||
assert "Cannot create issues on branch 'main'" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_issue_staging_branch_allowed(issue_tools):
|
||||
"""Test creating issue on staging branch (allowed for documentation)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='staging'):
|
||||
issue_tools.gitea.create_issue = Mock(return_value={'number': 1})
|
||||
|
||||
issue = await issue_tools.create_issue('Test', 'Body')
|
||||
|
||||
assert issue['number'] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_issue_main_branch_blocked(issue_tools):
|
||||
"""Test updating issue on main branch (blocked)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='main'):
|
||||
with pytest.raises(PermissionError) as exc_info:
|
||||
await issue_tools.update_issue(1, title='Updated')
|
||||
|
||||
assert "Cannot update issues on branch 'main'" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_issues_main_branch_allowed(issue_tools):
|
||||
"""Test listing issues on main branch (allowed - read-only)"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='main'):
|
||||
issue_tools.gitea.list_issues = Mock(return_value=[{'number': 1}])
|
||||
|
||||
issues = await issue_tools.list_issues(state='open')
|
||||
|
||||
assert len(issues) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_issue(issue_tools):
|
||||
"""Test getting specific issue"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='development'):
|
||||
issue_tools.gitea.get_issue = Mock(return_value={'number': 1, 'title': 'Test'})
|
||||
|
||||
issue = await issue_tools.get_issue(1)
|
||||
|
||||
assert issue['number'] == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_comment(issue_tools):
|
||||
"""Test adding comment to issue"""
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='development'):
|
||||
issue_tools.gitea.add_comment = Mock(return_value={'body': 'Test comment'})
|
||||
|
||||
comment = await issue_tools.add_comment(1, 'Test comment')
|
||||
|
||||
assert comment['body'] == 'Test comment'
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aggregate_issues_company_mode(issue_tools):
|
||||
"""Test aggregating issues in company mode"""
|
||||
issue_tools.gitea.mode = 'company'
|
||||
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='development'):
|
||||
issue_tools.gitea.aggregate_issues = Mock(return_value={
|
||||
'repo1': [{'number': 1}],
|
||||
'repo2': [{'number': 2}]
|
||||
})
|
||||
|
||||
aggregated = await issue_tools.aggregate_issues()
|
||||
|
||||
assert 'repo1' in aggregated
|
||||
assert 'repo2' in aggregated
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_aggregate_issues_project_mode_error(issue_tools):
|
||||
"""Test that aggregate_issues fails in project mode"""
|
||||
issue_tools.gitea.mode = 'project'
|
||||
|
||||
with patch.object(issue_tools, '_get_current_branch', return_value='development'):
|
||||
with pytest.raises(ValueError) as exc_info:
|
||||
await issue_tools.aggregate_issues()
|
||||
|
||||
assert "only available in company mode" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_branch_detection():
|
||||
"""Test branch detection logic"""
|
||||
tools = IssueTools(Mock())
|
||||
|
||||
# Test development branches
|
||||
with patch.object(tools, '_get_current_branch', return_value='development'):
|
||||
assert tools._check_branch_permissions('create_issue') is True
|
||||
|
||||
with patch.object(tools, '_get_current_branch', return_value='feat/new-feature'):
|
||||
assert tools._check_branch_permissions('create_issue') is True
|
||||
|
||||
# Test production branches
|
||||
with patch.object(tools, '_get_current_branch', return_value='main'):
|
||||
assert tools._check_branch_permissions('create_issue') is False
|
||||
assert tools._check_branch_permissions('list_issues') is True
|
||||
|
||||
# Test staging branches
|
||||
with patch.object(tools, '_get_current_branch', return_value='staging'):
|
||||
assert tools._check_branch_permissions('create_issue') is True
|
||||
assert tools._check_branch_permissions('update_issue') is False
|
||||
@@ -1,246 +0,0 @@
|
||||
"""
|
||||
Unit tests for label tools with suggestion logic.
|
||||
"""
|
||||
import pytest
|
||||
from unittest.mock import Mock, patch
|
||||
from mcp_server.tools.labels import LabelTools
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_gitea_client():
|
||||
"""Fixture providing mocked Gitea client"""
|
||||
client = Mock()
|
||||
client.repo = 'test_repo'
|
||||
return client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def label_tools(mock_gitea_client):
|
||||
"""Fixture providing LabelTools instance"""
|
||||
return LabelTools(mock_gitea_client)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_labels(label_tools):
|
||||
"""Test getting all labels (org + repo)"""
|
||||
label_tools.gitea.get_org_labels = Mock(return_value=[
|
||||
{'name': 'Type/Bug'},
|
||||
{'name': 'Type/Feature'}
|
||||
])
|
||||
label_tools.gitea.get_labels = Mock(return_value=[
|
||||
{'name': 'Component/Backend'},
|
||||
{'name': 'Component/Frontend'}
|
||||
])
|
||||
|
||||
result = await label_tools.get_labels()
|
||||
|
||||
assert len(result['organization']) == 2
|
||||
assert len(result['repository']) == 2
|
||||
assert result['total_count'] == 4
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_bug():
|
||||
"""Test label suggestion for bug context"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
context = "Fix critical bug in login authentication"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
|
||||
assert 'Type/Bug' in suggestions
|
||||
assert 'Priority/Critical' in suggestions
|
||||
assert 'Component/Auth' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_feature():
|
||||
"""Test label suggestion for feature context"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
context = "Add new feature to implement user dashboard"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
|
||||
assert 'Type/Feature' in suggestions
|
||||
assert any('Priority' in label for label in suggestions)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_refactor():
|
||||
"""Test label suggestion for refactor context"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
context = "Refactor architecture to extract service layer"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
|
||||
assert 'Type/Refactor' in suggestions
|
||||
assert 'Component/Backend' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_documentation():
|
||||
"""Test label suggestion for documentation context"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
context = "Update documentation for API endpoints"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
|
||||
assert 'Type/Documentation' in suggestions
|
||||
assert 'Component/API' in suggestions or 'Component/Docs' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_priority():
|
||||
"""Test priority detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# Critical priority
|
||||
context = "Urgent blocker in production"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Priority/Critical' in suggestions
|
||||
|
||||
# High priority
|
||||
context = "Important feature needed asap"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Priority/High' in suggestions
|
||||
|
||||
# Low priority
|
||||
context = "Nice-to-have optional improvement"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Priority/Low' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_complexity():
|
||||
"""Test complexity detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# Simple complexity
|
||||
context = "Simple quick fix for typo"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Complexity/Simple' in suggestions
|
||||
|
||||
# Complex complexity
|
||||
context = "Complex challenging architecture redesign"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Complexity/Complex' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_efforts():
|
||||
"""Test efforts detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# XS effort
|
||||
context = "Tiny fix that takes 1 hour"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Efforts/XS' in suggestions
|
||||
|
||||
# L effort
|
||||
context = "Large feature taking 1 week"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Efforts/L' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_components():
|
||||
"""Test component detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# Backend component
|
||||
context = "Update backend API service"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Component/Backend' in suggestions
|
||||
assert 'Component/API' in suggestions
|
||||
|
||||
# Frontend component
|
||||
context = "Fix frontend UI component"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Component/Frontend' in suggestions
|
||||
|
||||
# Database component
|
||||
context = "Add database migration for schema"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Component/Database' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_tech_stack():
|
||||
"""Test tech stack detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# Python
|
||||
context = "Update Python FastAPI endpoint"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Tech/Python' in suggestions
|
||||
assert 'Tech/FastAPI' in suggestions
|
||||
|
||||
# Docker
|
||||
context = "Fix Dockerfile configuration"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Tech/Docker' in suggestions
|
||||
|
||||
# PostgreSQL
|
||||
context = "Optimize PostgreSQL query"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Tech/PostgreSQL' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_source():
|
||||
"""Test source detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# Development
|
||||
context = "Issue found in development environment"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Source/Development' in suggestions
|
||||
|
||||
# Production
|
||||
context = "Critical production issue"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Source/Production' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_risk():
|
||||
"""Test risk detection in suggestions"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
# High risk
|
||||
context = "Breaking change to major API"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Risk/High' in suggestions
|
||||
|
||||
# Low risk
|
||||
context = "Safe minor update with low risk"
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
assert 'Risk/Low' in suggestions
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_suggest_labels_multiple_categories():
|
||||
"""Test that suggestions span multiple categories"""
|
||||
tools = LabelTools(Mock())
|
||||
|
||||
context = """
|
||||
Urgent critical bug in production backend API service.
|
||||
Need to fix broken authentication endpoint.
|
||||
This is a complex issue requiring FastAPI and PostgreSQL expertise.
|
||||
"""
|
||||
|
||||
suggestions = await tools.suggest_labels(context)
|
||||
|
||||
# Should have Type
|
||||
assert any('Type/' in label for label in suggestions)
|
||||
|
||||
# Should have Priority
|
||||
assert any('Priority/' in label for label in suggestions)
|
||||
|
||||
# Should have Component
|
||||
assert any('Component/' in label for label in suggestions)
|
||||
|
||||
# Should have Tech
|
||||
assert any('Tech/' in label for label in suggestions)
|
||||
|
||||
# Should have Source
|
||||
assert any('Source/' in label for label in suggestions)
|
||||
Reference in New Issue
Block a user