Files
leo-claude-mktplace/docs/references/MCP-GITEA.md
lmiranda ba599e342e refactor: update repository URL and rebrand to Bandit Labs
- Update git remote to new Tailscale hostname
- Replace old organization name (hhl-infra) with bandit
- Replace old repository name (claude-code-hhl-toolkit) with support-claude-mktplace
- Update all documentation references to use generic gitea.example.com
- Rebrand from HyperHive Labs to Bandit Labs across all files
- Rename workspace file to match new repository name

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-09 11:51:13 -05:00

1269 lines
37 KiB
Markdown

# Gitea MCP Server Reference
## Overview
The Gitea MCP Server provides integration with Gitea for issue management, label operations, and repository tracking. It's shared by both `projman` and `projman-pmo` plugins, detecting its operating mode based on environment variables.
**Location:** `mcp-servers/gitea/` (repository root)
**Key Features:**
- Issue CRUD operations
- Label taxonomy management (43-label system)
- Mode detection (project-scoped vs company-wide)
- Hybrid configuration (system + project level)
- Python 3.11+ implementation
---
## Architecture
### Mode Detection
The MCP server operates in two modes based on environment variables:
**Project Mode (projman):**
- When `GITEA_REPO` is present
- Operates on single repository
- Used by projman plugin
**Company Mode (pmo):**
- When `GITEA_REPO` is absent
- Operates on all repositories in organization
- Used by projman-pmo plugin
```python
# mcp-servers/gitea/mcp_server/config.py
def load(self):
# ... load configs ...
self.repo = os.getenv('GITEA_REPO') # Optional
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)")
return {
'api_url': self.api_url,
'api_token': self.api_token,
'owner': self.owner,
'repo': self.repo,
'mode': self.mode
}
```
---
## Configuration
### System-Level Configuration
**File:** `~/.config/claude/gitea.env`
```bash
GITEA_API_URL=https://gitea.example.com/api/v1
GITEA_API_TOKEN=your_gitea_token
GITEA_OWNER=bandit
```
**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. Token configuration:
- **Token Name:** `claude-code-mcp`
- **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. **Important:** Copy the token immediately (shown only once)
7. Add to configuration file (see setup below)
**Setup:**
```bash
# Create config directory
mkdir -p ~/.config/claude
# Create gitea.env
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
# Secure the file (important!)
chmod 600 ~/.config/claude/gitea.env
# Verify setup
cat ~/.config/claude/gitea.env
```
### Project-Level Configuration
**File:** `project-root/.env`
```bash
# Repository name (project mode only)
GITEA_REPO=cuisineflow
```
**Setup:**
```bash
# In each project root
echo "GITEA_REPO=cuisineflow" > .env
# Add to .gitignore
echo ".env" >> .gitignore
```
### Configuration Loading Strategy
```python
# mcp-servers/gitea/mcp_server/config.py
from pathlib import Path
from dotenv import load_dotenv
import os
from typing import Dict, Optional
class GiteaConfig:
"""Hybrid configuration loader"""
def __init__(self):
self.api_url: Optional[str] = None
self.api_token: Optional[str] = None
self.owner: Optional[str] = None
self.repo: Optional[str] = None
self.mode: str = 'project'
def load(self) -> Dict[str, str]:
"""
Load configuration from system and project levels.
Project-level configuration overrides system-level.
"""
# Load system config
system_config = Path.home() / '.config' / 'claude' / 'gitea.env'
if system_config.exists():
load_dotenv(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)
# Extract values
self.api_url = os.getenv('GITEA_API_URL')
self.api_token = os.getenv('GITEA_API_TOKEN')
self.owner = os.getenv('GITEA_OWNER')
self.repo = os.getenv('GITEA_REPO') # Optional for PMO
# Detect mode
self.mode = 'project' if self.repo else 'company'
# Validate required variables
self._validate()
return {
'api_url': self.api_url,
'api_token': self.api_token,
'owner': self.owner,
'repo': self.repo,
'mode': self.mode
}
def _validate(self) -> None:
"""Validate that required configuration is present"""
required = {
'GITEA_API_URL': self.api_url,
'GITEA_API_TOKEN': self.api_token,
'GITEA_OWNER': self.owner
}
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"
)
```
---
## Directory Structure
```
mcp-servers/gitea/
├── .venv/ # Python virtual environment
├── requirements.txt # Python dependencies
├── .env.example # Configuration template
├── 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 CRUD tools
│ └── labels.py # Label management tools
└── tests/
├── test_config.py
├── test_gitea_client.py
└── test_tools.py
```
---
## Dependencies
**File:** `mcp-servers/gitea/requirements.txt`
```txt
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
```
**Python Version:** 3.10+ required
**Installation:**
```bash
cd mcp-servers/gitea
python -m venv .venv
source .venv/bin/activate # Linux/Mac
# or .venv\Scripts\activate # Windows
pip install -r requirements.txt
```
---
## Gitea API Client
```python
# mcp-servers/gitea/mcp_server/gitea_client.py
import requests
from typing import List, Dict, Optional
from .config import GiteaConfig
class GiteaClient:
"""Client for interacting with Gitea API"""
def __init__(self):
config = GiteaConfig()
config_dict = config.load()
self.base_url = config_dict['api_url']
self.token = config_dict['api_token']
self.owner = config_dict['owner']
self.repo = config_dict.get('repo') # Optional for PMO
self.mode = config_dict['mode']
self.session = requests.Session()
self.session.headers.update({
'Authorization': f'token {self.token}',
'Content-Type': 'application/json'
})
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: Override configured repo (for PMO multi-repo)
"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues"
params = {'state': state}
if labels:
params['labels'] = ','.join(labels)
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"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues/{issue_number}"
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"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues"
data = {
'title': title,
'body': body
}
if labels:
data['labels'] = labels
response = self.session.post(url, json=data)
response.raise_for_status()
return response.json()
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"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.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
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"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues/{issue_number}/comments"
data = {'body': comment}
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"""
target_repo = repo or self.repo
if not target_repo:
raise ValueError("Repository not specified")
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/labels"
response = self.session.get(url)
response.raise_for_status()
return response.json()
def get_org_labels(self) -> List[Dict]:
"""Get organization-level labels"""
url = f"{self.base_url}/orgs/{self.owner}/labels"
response = self.session.get(url)
response.raise_for_status()
return response.json()
# PMO-specific methods
def list_repos(self) -> List[Dict]:
"""List all repositories in organization (PMO mode)"""
url = f"{self.base_url}/orgs/{self.owner}/repos"
response = self.session.get(url)
response.raise_for_status()
return response.json()
def aggregate_issues(
self,
state: str = 'open',
labels: Optional[List[str]] = None
) -> Dict[str, List[Dict]]:
"""
Fetch issues across all repositories (PMO mode).
Returns dict keyed by repository name.
"""
repos = self.list_repos()
aggregated = {}
for repo in repos:
repo_name = repo['name']
try:
issues = self.list_issues(
state=state,
labels=labels,
repo=repo_name
)
if issues:
aggregated[repo_name] = issues
except Exception as e:
# Log error but continue with other repos
print(f"Error fetching issues from {repo_name}: {e}")
return aggregated
```
---
## MCP Server Implementation
### Overview
The MCP (Model Context Protocol) server exposes Gitea tools to Claude Code via JSON-RPC 2.0 over stdio.
**Communication Flow:**
```
Claude Code <--> stdio <--> MCP Server <--> Gitea API
```
### Server Entry Point
**File:** `mcp-servers/gitea/mcp_server/server.py`
```python
"""
MCP Server entry point for Gitea integration.
"""
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
logging.basicConfig(level=logging.INFO)
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"""
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"
},
"labels": {
"type": "array",
"items": {"type": "string"}
},
"repo": {"type": "string"}
}
}
),
Tool(
name="get_issue",
description="Get specific issue details",
inputSchema={
"type": "object",
"properties": {
"issue_number": {"type": "integer"},
"repo": {"type": "string"}
},
"required": ["issue_number"]
}
),
Tool(
name="create_issue",
description="Create a new issue in Gitea",
inputSchema={
"type": "object",
"properties": {
"title": {"type": "string"},
"body": {"type": "string"},
"labels": {
"type": "array",
"items": {"type": "string"}
},
"repo": {"type": "string"}
},
"required": ["title", "body"]
}
),
Tool(
name="update_issue",
description="Update existing issue",
inputSchema={
"type": "object",
"properties": {
"issue_number": {"type": "integer"},
"title": {"type": "string"},
"body": {"type": "string"},
"state": {
"type": "string",
"enum": ["open", "closed"]
},
"labels": {
"type": "array",
"items": {"type": "string"}
},
"repo": {"type": "string"}
},
"required": ["issue_number"]
}
),
Tool(
name="add_comment",
description="Add comment to issue",
inputSchema={
"type": "object",
"properties": {
"issue_number": {"type": "integer"},
"comment": {"type": "string"},
"repo": {"type": "string"}
},
"required": ["issue_number", "comment"]
}
),
Tool(
name="get_labels",
description="Get all available labels (org + repo)",
inputSchema={
"type": "object",
"properties": {
"repo": {"type": "string"}
}
}
),
Tool(
name="suggest_labels",
description="Analyze context and suggest appropriate labels",
inputSchema={
"type": "object",
"properties": {
"context": {"type": "string"}
},
"required": ["context"]
}
),
Tool(
name="aggregate_issues",
description="Fetch issues across all repositories (PMO mode)",
inputSchema={
"type": "object",
"properties": {
"state": {
"type": "string",
"enum": ["open", "closed", "all"],
"default": "open"
},
"labels": {
"type": "array",
"items": {"type": "string"}
}
}
}
)
]
@self.server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Handle tool invocation"""
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())
```
### Tool Implementation with Async Wrappers
Since the Gitea client uses `requests` (synchronous), but MCP tools must be async:
```python
# mcp-servers/gitea/mcp_server/tools/issues.py
import asyncio
class IssueTools:
def __init__(self, gitea_client):
self.gitea = gitea_client
async def list_issues(self, state='open', labels=None, repo=None):
"""
List issues from repository.
Wraps sync Gitea client method in executor for async compatibility.
"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
self.gitea.list_issues,
state,
labels,
repo
)
async def create_issue(self, title, body, labels=None, repo=None):
"""Create new issue"""
loop = asyncio.get_event_loop()
return await loop.run_in_executor(
None,
self.gitea.create_issue,
title,
body,
labels,
repo
)
# Other methods follow same pattern
```
### Branch Detection in MCP Tools
Implement branch-aware security at the MCP level:
```python
# mcp-servers/gitea/mcp_server/tools/issues.py
import subprocess
class IssueTools:
def _get_current_branch(self) -> str:
"""Get current git branch"""
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"""
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 create_issue(self, title, body, labels=None, repo=None):
"""Create issue with branch check"""
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,
self.gitea.create_issue,
title,
body,
labels,
repo
)
```
### Testing the MCP Server
**Manual testing:**
```bash
cd mcp-servers/gitea
source .venv/bin/activate
python -m mcp_server.server
```
**Unit tests with mocks:**
```python
# tests/test_server.py
import pytest
from unittest.mock import Mock, patch
@pytest.mark.asyncio
async def test_server_initialization():
"""Test server initializes correctly"""
from mcp_server.server import GiteaMCPServer
with patch('mcp_server.server.GiteaClient'):
server = GiteaMCPServer()
await server.initialize()
assert server.client is not None
assert server.config is not None
```
**Integration tests with real API:**
```python
# tests/test_integration.py
import pytest
@pytest.mark.integration
@pytest.mark.asyncio
async def test_list_issues_real_api():
"""Test against real Gitea instance"""
from mcp_server.server import GiteaMCPServer
server = GiteaMCPServer()
await server.initialize()
result = await server.issue_tools.list_issues(state='open')
assert isinstance(result, list)
```
**Run tests:**
```bash
# Unit tests only (with mocks)
pytest tests/ -m "not integration"
# Integration tests (requires Gitea access)
pytest tests/ -m integration
# All tests
pytest tests/
```
---
## Label Taxonomy System
### Dynamic Label System
The label taxonomy is **fetched dynamically from Gitea** at runtime. The current system has 44 labels:
**Organization Labels (28):**
- Agent/2
- Complexity/3
- Efforts/5
- Priority/4
- Risk/3
- Source/4
- Type/6 (includes Type/Bug, Type/Feature, Type/Refactor, Type/Documentation, Type/Test, Type/Chore)
**Repository Labels (16):**
- Component/9 (Backend, Frontend, API, Database, Auth, Deploy, Testing, Docs, Infra)
- Tech/7 (Python, JavaScript, Docker, PostgreSQL, Redis, Vue, FastAPI)
**Note:** Label count and composition may change. The `/labels-sync` command fetches the latest labels from Gitea and updates suggestion logic accordingly.
### Label Suggestion Logic
```python
# mcp-servers/gitea/mcp_server/tools/labels.py
from typing import List, Dict
import re
class LabelTools:
def __init__(self, gitea_client):
self.gitea = gitea_client
async def get_labels(self, repo: str = None) -> List[Dict]:
"""Get all labels (org + repo)"""
org_labels = self.gitea.get_org_labels()
if repo or self.gitea.repo:
target_repo = repo or self.gitea.repo
repo_labels = self.gitea.get_labels(target_repo)
return org_labels + repo_labels
return org_labels
async def suggest_labels(self, context: str) -> List[str]:
"""
Analyze context and suggest appropriate labels.
Args:
context: Issue title + description or sprint context
"""
suggested = []
context_lower = context.lower()
# Type detection (exclusive - only one)
if any(word in context_lower for word in ['bug', 'error', 'fix', 'broken']):
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']):
suggested.append('Type/Feature')
elif any(word in context_lower for word in ['docs', 'documentation', 'readme']):
suggested.append('Type/Documentation')
elif any(word in context_lower for word in ['test', 'testing', 'spec']):
suggested.append('Type/Test')
elif any(word in context_lower for word in ['chore', 'maintenance', 'update']):
suggested.append('Type/Chore')
# Priority detection
if any(word in context_lower for word in ['critical', 'urgent', 'blocker', 'blocking']):
suggested.append('Priority/Critical')
elif any(word in context_lower for word in ['high', 'important', 'asap']):
suggested.append('Priority/High')
elif any(word in context_lower for word in ['low', 'nice-to-have', 'optional']):
suggested.append('Priority/Low')
else:
suggested.append('Priority/Medium')
# Component detection (based on keywords)
component_keywords = {
'Component/Backend': ['backend', 'server', 'api', 'database', 'service'],
'Component/Frontend': ['frontend', 'ui', 'interface', 'react', 'vue'],
'Component/API': ['api', 'endpoint', 'rest', 'graphql'],
'Component/Database': ['database', 'db', 'sql', 'migration', 'schema'],
'Component/Auth': ['auth', 'authentication', 'login', 'oauth', 'token'],
'Component/Deploy': ['deploy', 'deployment', 'docker', 'kubernetes'],
'Component/Testing': ['test', 'testing', 'spec', 'jest', 'pytest'],
'Component/Docs': ['docs', 'documentation', 'readme', 'guide']
}
for label, keywords in component_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:
suggested.append('Source/Staging')
elif 'production' in context_lower or 'prod' in context_lower:
suggested.append('Source/Production')
return suggested
```
---
## MCP Tools
### Issue Tools
```python
# mcp-servers/gitea/mcp_server/tools/issues.py
class IssueTools:
def __init__(self, gitea_client):
self.gitea = gitea_client
async def list_issues(self, state='open', labels=None, repo=None):
"""List issues in repository"""
return self.gitea.list_issues(state=state, labels=labels, repo=repo)
async def get_issue(self, issue_number, repo=None):
"""Get specific issue details"""
return self.gitea.get_issue(issue_number, repo=repo)
async def create_issue(self, title, body, labels=None, repo=None):
"""Create new issue"""
return self.gitea.create_issue(title, body, labels=labels, repo=repo)
async def update_issue(self, issue_number, title=None, body=None, state=None, labels=None, repo=None):
"""Update existing issue"""
return self.gitea.update_issue(
issue_number,
title=title,
body=body,
state=state,
labels=labels,
repo=repo
)
async def add_comment(self, issue_number, comment, repo=None):
"""Add comment to issue"""
return self.gitea.add_comment(issue_number, comment, repo=repo)
# PMO-specific
async def aggregate_issues(self, state='open', labels=None):
"""Aggregate issues across all repositories (PMO mode)"""
if self.gitea.mode != 'company':
raise ValueError("aggregate_issues only available in company mode")
return self.gitea.aggregate_issues(state=state, labels=labels)
```
---
## Testing
### Unit Tests
```python
# tests/test_config.py
import pytest
from pathlib import Path
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))
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
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_required_config():
"""Test error handling for missing configuration"""
with pytest.raises(FileNotFoundError):
config = GiteaConfig()
config.load()
```
### Integration Tests
```python
# tests/test_gitea_client.py
import pytest
from mcp_server.gitea_client import GiteaClient
@pytest.fixture
def gitea_client():
"""Fixture providing configured Gitea client"""
return GiteaClient()
def test_list_issues(gitea_client):
"""Test listing issues from Gitea"""
issues = gitea_client.list_issues(state='open')
assert isinstance(issues, list)
def test_create_issue(gitea_client):
"""Test creating an issue in Gitea"""
issue = gitea_client.create_issue(
title="Test Issue",
body="Test body",
labels=["Type/Bug", "Priority/Low"]
)
assert issue['title'] == "Test Issue"
assert any(label['name'] == "Type/Bug" for label in issue['labels'])
def test_get_labels(gitea_client):
"""Test fetching labels"""
labels = gitea_client.get_labels()
assert isinstance(labels, list)
assert len(labels) > 0
def test_pmo_mode_aggregate_issues():
"""Test PMO mode aggregation (no repo specified)"""
# Set up client without repo
client = GiteaClient() # Should detect company mode
if client.mode == 'company':
aggregated = client.aggregate_issues(state='open')
assert isinstance(aggregated, dict)
# Each key should be a repo name
for repo_name, issues in aggregated.items():
assert isinstance(issues, list)
```
### Running Tests
```bash
# Activate virtual environment
source .venv/bin/activate
# Run all tests
pytest
# Run with coverage
pytest --cov=mcp_server --cov-report=html
# Run specific test file
pytest tests/test_config.py
# Run with verbose output
pytest -v
```
---
## Performance Optimization
### Caching
```python
from functools import lru_cache
@lru_cache(maxsize=128)
def get_labels_cached(self, repo: str) -> List[Dict]:
"""Cached label retrieval to reduce API calls"""
return self.get_labels(repo)
```
### Retry Logic
```python
import time
from typing import Callable
def retry_on_failure(max_retries=3, delay=1):
"""Decorator for retrying failed API calls"""
def decorator(func: Callable):
def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return func(*args, **kwargs)
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise
time.sleep(delay * (attempt + 1))
return wrapper
return decorator
@retry_on_failure(max_retries=3)
def list_issues(self, state='open', labels=None, repo=None):
"""List issues with automatic retry"""
# Implementation
```
---
## Troubleshooting
### Common Issues
**Issue:** Module not found
```bash
# Solution: Ensure PYTHONPATH is set in .mcp.json
"env": {
"PYTHONPATH": "${CLAUDE_PLUGIN_ROOT}/../mcp-servers/gitea"
}
```
**Issue:** Configuration not loading
```bash
# Solution: Check file permissions
chmod 600 ~/.config/claude/gitea.env
# Verify file exists
cat ~/.config/claude/gitea.env
```
**Issue:** API authentication failing
```bash
# Solution: Test token manually
curl -H "Authorization: token YOUR_TOKEN" \
https://your-gitea.com/api/v1/user
```
**Issue:** PMO mode not working
```bash
# Solution: Ensure GITEA_REPO is NOT set
# Check environment variables
env | grep GITEA
```
---
## Security
### Best Practices
1. **Token Storage:**
- Store tokens in `~/.config/claude/gitea.env`
- Set file permissions to 600
- Never commit tokens to git
2. **Input Validation:**
- Validate all user input before API calls
- Sanitize issue titles and descriptions
- Prevent injection attacks
3. **Error Handling:**
- Don't leak tokens in error messages
- Log errors without sensitive data
- Provide user-friendly error messages
4. **API Rate Limiting:**
- Implement exponential backoff
- Cache frequently accessed data
- Batch requests where possible
---
## Next Steps
1. **Set up system configuration**
2. **Create virtual environment**
3. **Install dependencies**
4. **Run tests against actual Gitea instance**
5. **Integrate with projman plugin**