7 Commits

Author SHA1 Message Date
82f6ef794e chore: Update features and issue lib 2025-09-26 17:19:16 +02:00
6713768ea6 fix: Resolve failing tests after CLI and error handling refactoring
Fix all test failures introduced by recent architectural changes:

• Issue Creator Tests:
  - Fixed mock API responses to include required fields (created_at, updated_at, html_url)
  - Added input validation for empty titles back to issue creator
  - Updated test expectations to match new error handling patterns
  - Created helper function for complete mock responses

• Issue Fetcher Test:
  - Updated mock target from tddai.issue_fetcher.subprocess to gitea.http_client.subprocess
  - Fixed test assertions to match new error handling with specific exception chaining
  - Test now properly validates API error translation

• Makefile Integration Test:
  - Implemented lazy initialization in tddai_cli.py to prevent import-time configuration errors
  - Replaced eager CLI framework initialization with _get_cli() lazy pattern
  - Preserves normal CLI functionality while fixing test environment compatibility

• Result: All 171 tests now pass (169 passed, 2 skipped)
• Maintains backward compatibility of CLI interface
• Validates that refactored error handling works correctly

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 17:15:36 +02:00
235e6831ed docs: Add comprehensive error handling guidelines
Create ERROR_HANDLING_GUIDE.md with complete reference for maintaining
consistent error handling patterns across the codebase:

• Quick reference with DO/DON'T examples
• Complete exception hierarchy documentation
• Service layer and file operation patterns
• Exception chaining and logging integration rules
• Anti-patterns to avoid and testing guidelines
• Refactoring checklist with search patterns
• Migration templates for future cleanups

This guide ensures:
- Consistent error handling patterns
- Preserved debugging context
- User-friendly error messages
- No silent failures
- Easy future maintenance

Prevents codebase coherence loss over time by providing systematic
approach for identifying and fixing error handling issues.

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 16:54:44 +02:00
bbc6192fe1 refactor: Standardize error handling patterns across codebase
Comprehensive error handling improvements addressing inconsistent patterns:

• Created markitect/exceptions.py with complete domain-specific exception hierarchy
  - MarkitectError base class with context and cause chaining support
  - Specific exceptions for Document, AST, Cache, Database, Schema operations
  - Built-in logging and context preservation

• Fixed overly broad exception handling in tddai modules:
  - issue_fetcher.py: Replace generic Exception with specific Gitea errors
  - project_manager.py: Proper error translation with context preservation
  - coverage_analyzer.py: Replace silent suppression with logging

• Enhanced cache_service.py error handling:
  - Specific OSError/PermissionError handling for file operations
  - Logging integration for unexpected errors
  - Preserved error collection and reporting

• Implemented proper exception chaining patterns:
  - All error translations use `raise ... from e` for debugging
  - Preserved original exception context and stack traces
  - Added docstring declarations of raised exceptions

• Benefits:
  - Eliminates silent error suppression and debugging black holes
  - Provides specific, actionable error messages
  - Preserves full error context for troubleshooting
  - Establishes consistent patterns for future development

Resolves issue #21: Error handling standardization

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 16:35:13 +02:00
7f5309c4b0 refactor: Separate CLI presentation from core business logic
Complete architectural separation of concerns implementing clean layered design:

• Services Layer: Pure business logic isolated from presentation
  - WorkspaceService: TDD workspace operations
  - IssueService: Issue management and creation
  - ProjectService: Project management and milestones
  - ExportService: Unix-friendly data export

• CLI Layer: Clean presentation with command/presenter separation
  - Commands delegate to services for all business operations
  - Presenters handle formatted output and error messaging
  - Framework provides unified interface

• Benefits:
  - Eliminates mixed concerns in 943-line CLI monolith
  - Enables easier testing and maintenance
  - Preserves all existing functionality and Unix pipeline compatibility
  - Provides foundation for future CLI development

Resolves issue #20: CLI separation from core logic

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 15:08:54 +02:00
fd8f792f08 refactor: Factor out Gitea interfacing into clean facade pattern
- Create new gitea/ package with clean API facade
- Establish proper separation of concerns: tddai uses gitea, not vice versa
- Replace duplicate curl+subprocess patterns with unified HTTP client
- Add rich domain models with properties (issue.priority, issue.status)
- Maintain full backwards compatibility in tddai modules
- Reduce code complexity: -373 lines, +151 lines (net -222 lines)
- Improve testability and maintainability through clean interfaces

Architecture:
- gitea.client.GiteaClient - main facade with sub-clients
- gitea.api_client - high-level API with model conversion
- gitea.http_client - low-level HTTP operations
- gitea.models - rich domain objects (Issue, Milestone, Label)
- gitea.config - gitea-specific configuration
- gitea.exceptions - clean exception hierarchy

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 14:25:40 +02:00
b20b7003f5 feat: Add Unix-friendly issue index with multiple output formats
- Add issue-index command with TSV, CSV, JSON, and fields output formats
- Support sorting by number, title, priority, state, created, updated
- Add filtering by state (open/closed) and priority level
- Include proper data cleaning for Unix pipeline processing
- Add make targets: issues-get, issues-csv, issues-json, issues-high
- Optimize for awk, cut, grep, and other Unix text processing tools

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

Co-Authored-By: Claude <noreply@anthropic.com>
2025-09-26 11:24:25 +02:00
35 changed files with 3143 additions and 983 deletions

4
.gitignore vendored
View File

@@ -89,3 +89,7 @@ debug_*.py
# Claude Code local settings (user-specific permissions)
.claude/settings.local.json
.aider*
# TDDAI-specific ignores
ISSUES.index

385
ERROR_HANDLING_GUIDE.md Normal file
View File

@@ -0,0 +1,385 @@
# Error Handling Guidelines
**Version**: 1.0
**Last Updated**: 2025-09-26
**Purpose**: Maintain consistent, debuggable error handling across the codebase
## Quick Reference
### ✅ DO
```python
# Specific exception handling with chaining
try:
result = api_call()
except GiteaNotFoundError as e:
raise IssueError(f"Issue #{number} not found") from e
except GiteaAuthError as e:
raise IssueError(f"Authentication failed") from e
# Logging for unexpected errors
except Exception as e:
logger.error(f"Unexpected error in {operation}", exc_info=True)
raise DomainError(f"Operation failed: {operation}") from e
```
### ❌ DON'T
```python
# Overly broad exception handling
try:
result = api_call()
except Exception as e: # Too broad!
raise IssueError(f"Failed: {e}")
# Silent error suppression
try:
process_file()
except Exception:
continue # Never do this!
```
## 1. Exception Hierarchy
### Use Domain-Specific Exceptions
**Markitect Operations:**
```python
from markitect.exceptions import (
MarkitectError, # Base for all Markitect operations
DocumentError, # Document processing errors
ASTError, # AST parsing/processing errors
CacheError, # Cache operations errors
DatabaseError, # Database operation errors
SchemaError, # Schema validation/processing
ValidationError, # Document validation errors
GraphQLError, # GraphQL operations
ConfigurationError # Configuration/setup errors
)
```
**TDDAI Operations:**
```python
from tddai.exceptions import (
TddaiError, # Base for TDDAI operations
WorkspaceError, # Workspace management
IssueError, # Issue fetching/management
TestGenerationError, # Test generation
ConfigurationError # Configuration issues
)
```
**Gitea Operations:**
```python
from gitea.exceptions import (
GiteaError, # Base Gitea error
GiteaNotFoundError,# 404 responses
GiteaAuthError, # Authentication failures
GiteaApiError, # API errors with status codes
GiteaConfigError # Configuration issues
)
```
## 2. Exception Translation Patterns
### Service Layer Pattern
Services should translate external exceptions to domain exceptions:
```python
class IssueService:
def get_issue(self, issue_number: int) -> Issue:
"""Get issue by number.
Raises:
IssueError: When issue cannot be retrieved
"""
try:
return self.gitea_client.issues.get(issue_number)
except GiteaNotFoundError as e:
raise IssueError(f"Issue #{issue_number} not found") from e
except GiteaAuthError as e:
raise IssueError(f"Authentication failed") from e
except GiteaApiError as e:
raise IssueError(f"API error: {e}") from e
# Don't catch GiteaError - let specific exceptions handle it
```
### File Operations Pattern
```python
def read_config_file(file_path: Path) -> dict:
"""Read configuration file.
Raises:
ConfigurationError: When file cannot be read or parsed
"""
try:
content = file_path.read_text()
return json.loads(content)
except FileNotFoundError as e:
raise ConfigurationError(f"Config file not found: {file_path}") from e
except PermissionError as e:
raise ConfigurationError(f"Permission denied reading: {file_path}") from e
except json.JSONDecodeError as e:
raise ConfigurationError(f"Invalid JSON in {file_path}: {e}") from e
except Exception as e:
logger.error(f"Unexpected error reading {file_path}", exc_info=True)
raise ConfigurationError(f"Failed to read config: {file_path}") from e
```
## 3. Exception Chaining Rules
### Always Chain Exceptions
Use `raise ... from e` to preserve the original exception:
```python
# ✅ CORRECT - preserves debugging information
try:
dangerous_operation()
except SpecificError as e:
raise DomainError("User-friendly message") from e
# ❌ WRONG - loses original exception context
try:
dangerous_operation()
except SpecificError as e:
raise DomainError(f"Failed: {e}")
```
### Chain Standard Exceptions
```python
try:
data = json.loads(content)
except json.JSONDecodeError as e:
raise ValidationError(f"Invalid JSON format") from e
except (ValueError, TypeError) as e:
raise ValidationError(f"Data validation failed") from e
```
## 4. Logging Integration
### Log Before Re-raising
```python
import logging
logger = logging.getLogger(__name__)
try:
complex_operation()
except ExpectedError as e:
# Don't log expected errors - let caller decide
raise DomainError("Operation failed") from e
except Exception as e:
# Always log unexpected errors
logger.error(
"Unexpected error in complex_operation",
extra={'context': {'param1': value1}},
exc_info=True
)
raise DomainError("Unexpected failure") from e
```
### Logging Levels
- **ERROR**: Unexpected exceptions that indicate bugs
- **WARNING**: Expected exceptions that are concerning (file not found, permission denied)
- **INFO**: Normal error recovery (retries, fallbacks)
- **DEBUG**: Detailed error context for development
## 5. Error Messages
### User-Facing Messages
```python
# ✅ GOOD - actionable and specific
raise IssueError(f"Issue #{number} not found. Check the issue number and try again.")
# ✅ GOOD - includes context
raise ConfigurationError(f"Missing required setting 'api_token' in {config_file}")
# ❌ BAD - too technical
raise IssueError(f"HTTP 404 response from /api/v1/repos/owner/repo/issues/{number}")
# ❌ BAD - too vague
raise IssueError("Something went wrong")
```
### Include Context
```python
# Use MarkitectError's context parameter
raise DocumentError(
"Failed to parse document",
cause=original_error,
context={
'file_path': str(file_path),
'line_number': line_num,
'operation': 'parse_markdown'
}
)
```
## 6. CLI Error Handling
### Consistent CLI Pattern
```python
def cli_command():
"""CLI command that handles domain exceptions."""
try:
result = service.perform_operation()
show_success(result)
except DomainError as e:
OutputFormatter.exit_with_error(str(e))
# Don't catch Exception - let unexpected errors bubble up
```
### Exit Codes
- **0**: Success
- **1**: Expected failure (user error, missing resource)
- **2**: Configuration error
- **>2**: Unexpected error (let Python handle it)
## 7. Anti-Patterns to Avoid
### 1. Overly Broad Exception Handling
```python
# ❌ NEVER DO THIS
try:
operation()
except Exception as e:
raise DomainError(f"Failed: {e}")
```
### 2. Silent Error Suppression
```python
# ❌ NEVER DO THIS
try:
process_file(file)
except Exception:
continue # Silent failure!
# ✅ DO THIS INSTEAD
try:
process_file(file)
except (OSError, IOError) as e:
logger.warning(f"Could not process {file}: {e}")
continue
except Exception as e:
logger.error(f"Unexpected error processing {file}: {e}", exc_info=True)
continue
```
### 3. Exception Conversion Without Context
```python
# ❌ WRONG - loses information
try:
api_call()
except requests.RequestException as e:
raise IssueError("API failed")
# ✅ CORRECT - preserves context
try:
api_call()
except requests.ConnectionError as e:
raise IssueError("Cannot connect to Gitea server") from e
except requests.Timeout as e:
raise IssueError("Gitea server request timed out") from e
except requests.HTTPError as e:
raise IssueError(f"HTTP error: {e.response.status_code}") from e
```
## 8. Testing Error Handling
### Test Exception Translation
```python
def test_issue_not_found():
"""Test that GiteaNotFoundError is translated to IssueError."""
with mock.patch.object(gitea_client, 'get') as mock_get:
mock_get.side_effect = GiteaNotFoundError("Not found")
with pytest.raises(IssueError) as exc_info:
service.get_issue(123)
assert "Issue #123 not found" in str(exc_info.value)
assert exc_info.value.__cause__.__class__ == GiteaNotFoundError
```
### Test Error Messages
```python
def test_meaningful_error_messages():
"""Test that error messages are user-friendly."""
with pytest.raises(ConfigurationError) as exc_info:
service.load_config("nonexistent.json")
error_msg = str(exc_info.value)
assert "nonexistent.json" in error_msg
assert "not found" in error_msg.lower()
```
## 9. Refactoring Checklist
When refactoring error handling, use this checklist:
### 🔍 Identify Issues
- [ ] Search for `except Exception:` patterns
- [ ] Look for `continue` without logging in exception blocks
- [ ] Find missing exception chaining (`raise ... from e`)
- [ ] Check for generic error messages
### 🔧 Fix Patterns
- [ ] Replace broad exceptions with specific ones
- [ ] Add proper exception chaining
- [ ] Implement logging for unexpected errors
- [ ] Improve error message clarity
- [ ] Add exception documentation to functions
### ✅ Verify
- [ ] Test that CLI still works
- [ ] Verify error messages are user-friendly
- [ ] Check that debugging information is preserved
- [ ] Ensure no silent failures remain
### 📚 Document
- [ ] Update function docstrings with `Raises:` sections
- [ ] Add new exceptions to relevant `__init__.py` files
- [ ] Update this guide if new patterns emerge
## 10. Common Search Patterns
Use these patterns to find error handling issues:
```bash
# Find overly broad exception handling
rg "except Exception" --type py
# Find silent error suppression
rg "except.*:\s*continue" --type py
rg "except.*:\s*pass" --type py
# Find missing exception chaining
rg "raise.*Error.*:" --type py | grep -v "from"
# Find exception handling without logging
rg "except.*Exception.*:" -A 3 --type py | grep -v "log"
```
## 11. Quick Migration Template
Use this template for migrating old exception handling:
```python
# OLD PATTERN
try:
operation()
except Exception as e:
raise DomainError(f"Operation failed: {e}")
# NEW PATTERN
try:
operation()
except SpecificError1 as e:
raise DomainError(f"Specific failure case 1") from e
except SpecificError2 as e:
raise DomainError(f"Specific failure case 2") from e
except Exception as e:
logger.error("Unexpected error in operation", exc_info=True)
raise DomainError(f"Unexpected operation failure") from e
```
---
**Remember**: Good error handling makes debugging easier, provides better user experience, and prevents silent failures that hide bugs. When in doubt, be specific, preserve context, and log unexpected errors.

View File

@@ -195,4 +195,4 @@ AND json_extract(front_matter, '$.category') = 'technical';
---
*MarkiTect represents a paradigm shift from simple markdown processing to comprehensive document lifecycle management with performance guarantees and relational capabilities.*
*MarkiTect represents a paradigm shift from simple markdown processing to comprehensive document lifecycle management with performance guarantees and relational capabilities.*

View File

@@ -37,9 +37,13 @@ help:
@echo " add-diary-entry - Add new entry to ProjectDiary.md (requires Claude Code)"
@echo ""
@echo "Issue Management:"
@echo " list-issues - Show all gitea issues with status and priority"
@echo " list-open-issues - Show only open issues (active backlog)"
@echo " show-issue NUM=X - Show detailed view of specific issue"
@echo " list-issues - Show all gitea issues with status and priority"
@echo " list-open-issues - Show only open issues (active backlog)"
@echo " show-issue NUM=X - Show detailed view of specific issue"
@echo " issues-get - Export compact issue index to ISSUES.index"
@echo " issues-csv - Export issues as CSV for spreadsheet processing"
@echo " issues-json - Export issues as JSON for programmatic processing"
@echo " issues-high - Export only high/critical priority issues"
@echo ""
@echo "Test-Driven Development:"
@echo " test-from-issue NUM=X - Generate test skeleton from issue (requires Claude Code)"
@@ -247,6 +251,40 @@ show-issue: $(VENV)/bin/activate
list-open-issues: $(VENV)/bin/activate
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py list-open-issues
# Export compact issue index to ISSUES.index file (TSV format)
issues-get: $(VENV)/bin/activate
@echo "📋 Fetching issue index from gitea..."
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py issue-index --format tsv --sort number > ISSUES.index
@echo "✅ Issue index exported to ISSUES.index (TSV format)"
@echo "📄 File contents:"
@cat ISSUES.index
# Export issues as CSV for spreadsheet processing
issues-csv: $(VENV)/bin/activate
@echo "📊 Exporting issues as CSV..."
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py issue-index --format csv --sort priority --include-state > ISSUES.csv
@echo "✅ Issues exported to ISSUES.csv"
@wc -l ISSUES.csv | awk '{print "📄 Total entries:", $$1-1, "(excluding header)"}'
# Export issues as JSON for programmatic processing
issues-json: $(VENV)/bin/activate
@echo "🔧 Exporting issues as JSON..."
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py issue-index --format json --sort priority > ISSUES.json
@echo "✅ Issues exported to ISSUES.json"
@echo "📄 Sample entry:"
@head -20 ISSUES.json
# Export only high and critical priority issues
issues-high: $(VENV)/bin/activate
@echo "🚨 Exporting high priority issues..."
@echo "High priority issues:" > ISSUES.high.txt
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py issue-index --format tsv --filter-priority high --sort number >> ISSUES.high.txt
@echo "" >> ISSUES.high.txt
@echo "Critical priority issues:" >> ISSUES.high.txt
@PYTHONPATH=. $(VENV_PYTHON) tddai_cli.py issue-index --format tsv --filter-priority critical --sort number >> ISSUES.high.txt
@echo "✅ High priority issues exported to ISSUES.high.txt"
@cat ISSUES.high.txt
# Generate test skeleton from gitea issue (requires Claude Code)
test-from-issue:
@if [ -z "$(NUM)" ]; then \

20
cli/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
"""
CLI presentation layer.
This package handles all CLI-specific concerns:
- Argument parsing and validation
- Output formatting and presentation
- User interaction and feedback
- Error handling and display
The CLI layer delegates business logic to services and focuses purely on
presentation and user interface concerns.
"""
from .core import CLIFramework
from .presenters import *
from .commands import *
__all__ = [
'CLIFramework'
]

18
cli/commands/__init__.py Normal file
View File

@@ -0,0 +1,18 @@
"""
CLI command modules.
Commands handle argument parsing and delegation to services.
They contain no business logic, only CLI-specific concerns.
"""
from .workspace import WorkspaceCommands
from .issues import IssueCommands
from .project import ProjectCommands
from .export import ExportCommands
__all__ = [
'WorkspaceCommands',
'IssueCommands',
'ProjectCommands',
'ExportCommands'
]

46
cli/commands/export.py Normal file
View File

@@ -0,0 +1,46 @@
"""
Export and reporting CLI commands.
"""
import sys
from typing import Optional
from tddai import TddaiError
from services import ExportService
from cli.presenters import OutputFormatter
class ExportCommands:
"""Commands for data export and reporting."""
def __init__(self):
self.service = ExportService()
def issue_index(self, format_type: str = "tsv", sort_by: str = "number",
filter_state: Optional[str] = None, filter_priority: Optional[str] = None,
include_state: bool = False) -> None:
"""Output compact index of all issues for Unix processing.
Args:
format_type: Output format (tsv, csv, json, fields)
sort_by: Sort by field (number, title, priority, state, created, updated)
filter_state: Filter by state (open, closed)
filter_priority: Filter by priority (low, medium, high, critical, none)
include_state: Include state column in output
"""
try:
output = self.service.export_issues(
format_type=format_type,
sort_by=sort_by,
filter_state=filter_state,
filter_priority=filter_priority,
include_state=include_state
)
# Output directly to stdout for piping
print(output)
except TddaiError as e:
# Send error to stderr to avoid corrupting piped output
print(f"❌ Error: {e}", file=sys.stderr)
sys.exit(1)

114
cli/commands/issues.py Normal file
View File

@@ -0,0 +1,114 @@
"""
Issue CLI commands.
"""
from typing import List
from tddai import TddaiError
from services import IssueService
from cli.presenters import OutputFormatter, IssueView
class IssueCommands:
"""Commands for issue operations."""
def __init__(self):
self.service = IssueService()
def list_issues(self) -> None:
"""List all issues."""
try:
issues = self.service.list_issues()
IssueView.show_list(issues)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))
def list_open_issues(self) -> None:
"""List only open issues."""
try:
issues = self.service.list_open_issues()
IssueView.show_open_issues(issues)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))
def show_issue(self, issue_number: int) -> None:
"""Show detailed issue information."""
try:
issue_data = self.service.get_issue_details(issue_number)
IssueView.show_issue_details(issue_data)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))
def create_issue(self, title: str, body: str, issue_type: str = "enhancement") -> None:
"""Create a new issue."""
try:
OutputFormatter.info(f"Creating {issue_type} issue: {title}")
OutputFormatter.empty_line()
result = self.service.create_issue(title, body, labels=[issue_type])
IssueView.show_creation_success(result, issue_type)
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error creating issue: {e}")
def create_enhancement_issue(self, title: str, use_case: str,
technical_requirements: str = "",
acceptance_criteria: List[str] = None,
dependencies: List[str] = None,
priority: str = "Medium") -> None:
"""Create a structured enhancement issue."""
try:
OutputFormatter.info(f"Creating enhancement issue: {title}")
OutputFormatter.empty_line()
result = self.service.create_enhancement_issue(
title, use_case, technical_requirements,
acceptance_criteria, dependencies, priority
)
OutputFormatter.success("Enhancement issue created successfully!")
OutputFormatter.key_value("Number", f"#{result['number']}")
OutputFormatter.key_value("Title", result['title'])
OutputFormatter.key_value("Priority", priority)
if 'html_url' in result:
OutputFormatter.key_value("URL", result['html_url'])
OutputFormatter.empty_line()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error creating enhancement issue: {e}")
def create_from_template(self, template_file: str, **kwargs) -> None:
"""Create issue from template file."""
try:
OutputFormatter.info(f"Creating issue from template: {template_file}")
OutputFormatter.empty_line()
result = self.service.create_from_template(template_file, **kwargs)
OutputFormatter.success("Issue created from template successfully!")
OutputFormatter.key_value("Number", f"#{result['number']}")
OutputFormatter.key_value("Title", result['title'])
if 'html_url' in result:
OutputFormatter.key_value("URL", result['html_url'])
OutputFormatter.empty_line()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error creating issue from template: {e}")
def analyze_coverage(self, issue_number: int) -> None:
"""Analyze test coverage for a specific issue."""
try:
coverage_data = self.service.analyze_coverage(issue_number)
IssueView.show_coverage_analysis(coverage_data)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))

88
cli/commands/project.py Normal file
View File

@@ -0,0 +1,88 @@
"""
Project management CLI commands.
"""
from tddai import TddaiError
from services import ProjectService
from cli.presenters import OutputFormatter, ProjectView
class ProjectCommands:
"""Commands for project management operations."""
def __init__(self):
self.service = ProjectService()
def setup_project_management(self) -> None:
"""Setup project management labels and milestones."""
try:
OutputFormatter.info("Setting up project management system...")
self.service.setup_project_management()
ProjectView.show_setup_success()
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error setting up project management: {e}")
def move_issue_to_state(self, issue_number: int, state: str) -> None:
"""Move issue to a specific project state."""
try:
OutputFormatter.info(f"Moving issue #{issue_number} to {state} state...")
result = self.service.set_issue_state(issue_number, state)
# If moving to done, also close the issue
if state == 'done':
self.service.move_issue_to_done(issue_number)
OutputFormatter.success(f"Issue #{issue_number} moved to {state} and closed")
else:
OutputFormatter.success(f"Issue #{issue_number} moved to {state}")
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error moving issue to {state}: {e}")
def set_issue_priority(self, issue_number: int, priority: str) -> None:
"""Set issue priority."""
try:
OutputFormatter.info(f"Setting issue #{issue_number} priority to {priority}...")
result = self.service.set_issue_priority(issue_number, priority)
OutputFormatter.success(f"Issue #{issue_number} priority set to {priority}")
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error setting issue priority: {e}")
def create_milestone(self, title: str, description: str = "") -> None:
"""Create a new milestone (project)."""
try:
OutputFormatter.info(f"Creating milestone: {title}")
milestone = self.service.create_milestone(title, description)
OutputFormatter.success("Milestone created successfully!")
OutputFormatter.key_value("ID", milestone.id)
OutputFormatter.key_value("Title", milestone.title)
OutputFormatter.key_value("Description", milestone.description)
OutputFormatter.key_value("State", milestone.state)
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error creating milestone: {e}")
def list_milestones(self) -> None:
"""List all milestones."""
try:
milestones = self.service.list_milestones("all")
ProjectView.show_milestone_list(milestones)
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error listing milestones: {e}")
def assign_issue_to_milestone(self, issue_number: int, milestone_id: int) -> None:
"""Assign issue to a milestone."""
try:
OutputFormatter.info(f"Assigning issue #{issue_number} to milestone #{milestone_id}...")
result = self.service.assign_issue_to_milestone(issue_number, milestone_id)
OutputFormatter.success(f"Issue #{issue_number} assigned to milestone #{milestone_id}")
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error assigning issue to milestone: {e}")
def project_overview(self) -> None:
"""Show project management overview."""
try:
overview = self.service.get_project_overview()
ProjectView.show_overview(overview)
except TddaiError as e:
OutputFormatter.exit_with_error(f"Error getting project overview: {e}")

99
cli/commands/workspace.py Normal file
View File

@@ -0,0 +1,99 @@
"""
Workspace CLI commands.
"""
from tddai import TddaiError
from services import WorkspaceService
from cli.presenters import OutputFormatter, WorkspaceView
class WorkspaceCommands:
"""Commands for workspace operations."""
def __init__(self):
self.service = WorkspaceService()
def status(self) -> None:
"""Show current workspace status."""
try:
summary = self.service.get_workspace_summary()
WorkspaceView.show_status(summary)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))
def start_issue(self, issue_number: int) -> None:
"""Start working on an issue."""
try:
OutputFormatter.info(f"Starting work on issue #{issue_number}...")
OutputFormatter.info(f"Fetching issue #{issue_number} details...")
workspace_info = self.service.start_issue_workspace(issue_number)
summary = self.service.get_workspace_summary()
WorkspaceView.show_start_success(summary)
except TddaiError as e:
if "Already working on" in str(e):
OutputFormatter.warning(str(e))
print(" Run 'make tdd-finish' first or 'make tdd-status' to see details")
OutputFormatter.exit_with_error("Cannot start new workspace", 1)
else:
OutputFormatter.exit_with_error(str(e))
def finish_issue(self) -> None:
"""Finish current issue workspace."""
try:
issue_number = self.service.finish_current_workspace()
if issue_number is None:
OutputFormatter.error("No active issue workspace")
print(" Nothing to finish")
OutputFormatter.exit_with_error("", 1)
# Get test count before finishing
summary = self.service.get_workspace_summary()
test_count = summary.get('test_count', 0)
WorkspaceView.show_finish_success(issue_number, test_count)
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))
def add_test_guidance(self) -> None:
"""Show guidance for adding tests."""
try:
summary = self.service.get_workspace_summary()
if not summary['active']:
OutputFormatter.error("No active issue workspace")
print(" Run 'make tdd-start NUM=X' first")
OutputFormatter.exit_with_error("", 1)
issue_num = summary['issue_number']
issue_title = summary['issue_title']
workspace_dir = summary['workspace_dir']
print(f"🧪 Adding test to issue #{issue_num} workspace")
OutputFormatter.empty_line()
OutputFormatter.key_value("Issue", f"#{issue_num}: {issue_title}")
OutputFormatter.key_value("Workspace", f"{workspace_dir}/issue_{issue_num}/")
OutputFormatter.empty_line()
print("🤖 Please ask Claude Code to generate a test:")
OutputFormatter.empty_line()
print(" Command: 'Generate a test for the current workspace issue'")
OutputFormatter.empty_line()
print("📝 Test Requirements:")
print(f" - Save test in: {workspace_dir}/issue_{issue_num}/tests/")
print(f" - Name format: test_issue_{issue_num}_<scenario>.py")
print(f" - Include docstring referencing issue #{issue_num}")
print(" - Follow TDD principles (test should fail initially)")
print(" - Review requirements.md and test_plan.md for context")
OutputFormatter.empty_line()
print("📋 Issue Details:")
OutputFormatter.key_value("Title", issue_title)
# Note: Could fetch full issue details if needed
OutputFormatter.empty_line()
print("💡 After generation: Use 'make tdd-status' to see all tests")
except TddaiError as e:
OutputFormatter.exit_with_error(str(e))

78
cli/core.py Normal file
View File

@@ -0,0 +1,78 @@
"""
CLI framework core.
Provides the main CLI framework and command delegation.
"""
from .commands import WorkspaceCommands, IssueCommands, ProjectCommands, ExportCommands
class CLIFramework:
"""Main CLI framework that delegates to command classes."""
def __init__(self):
self.workspace = WorkspaceCommands()
self.issues = IssueCommands()
self.project = ProjectCommands()
self.export = ExportCommands()
# Workspace operations
def workspace_status(self):
return self.workspace.status()
def start_issue(self, issue_number: int):
return self.workspace.start_issue(issue_number)
def finish_issue(self):
return self.workspace.finish_issue()
def add_test_guidance(self):
return self.workspace.add_test_guidance()
# Issue operations
def list_issues(self):
return self.issues.list_issues()
def list_open_issues(self):
return self.issues.list_open_issues()
def show_issue(self, issue_number: int):
return self.issues.show_issue(issue_number)
def create_issue(self, title: str, body: str, issue_type: str = "enhancement"):
return self.issues.create_issue(title, body, issue_type)
def create_enhancement_issue(self, title: str, use_case: str, **kwargs):
return self.issues.create_enhancement_issue(title, use_case, **kwargs)
def create_from_template(self, template_file: str, **kwargs):
return self.issues.create_from_template(template_file, **kwargs)
def analyze_coverage(self, issue_number: int):
return self.issues.analyze_coverage(issue_number)
# Project management operations
def setup_project_management(self):
return self.project.setup_project_management()
def move_issue_to_state(self, issue_number: int, state: str):
return self.project.move_issue_to_state(issue_number, state)
def set_issue_priority(self, issue_number: int, priority: str):
return self.project.set_issue_priority(issue_number, priority)
def create_milestone(self, title: str, description: str = ""):
return self.project.create_milestone(title, description)
def list_milestones(self):
return self.project.list_milestones()
def assign_issue_to_milestone(self, issue_number: int, milestone_id: int):
return self.project.assign_issue_to_milestone(issue_number, milestone_id)
def project_overview(self):
return self.project.project_overview()
# Export operations
def issue_index(self, **kwargs):
return self.export.issue_index(**kwargs)

View File

@@ -0,0 +1,16 @@
"""
Presenters for CLI output formatting.
Presenters handle all output formatting and user feedback without
containing business logic.
"""
from .formatters import OutputFormatter
from .views import WorkspaceView, IssueView, ProjectView
__all__ = [
'OutputFormatter',
'WorkspaceView',
'IssueView',
'ProjectView'
]

View File

@@ -0,0 +1,85 @@
"""
Output formatting utilities for CLI presentation.
"""
import sys
from typing import Any, Dict, List
class OutputFormatter:
"""Handles output formatting and display."""
@staticmethod
def success(message: str) -> None:
"""Display success message."""
print(f"{message}")
@staticmethod
def info(message: str) -> None:
"""Display info message."""
print(f"📋 {message}")
@staticmethod
def warning(message: str) -> None:
"""Display warning message."""
print(f"⚠️ {message}")
@staticmethod
def error(message: str) -> None:
"""Display error message."""
print(f"{message}")
@staticmethod
def header(title: str, separator: str = "=") -> None:
"""Display section header."""
print(title)
print(separator * len(title))
print()
@staticmethod
def section(title: str) -> None:
"""Display section title."""
print(f"## {title}")
print()
@staticmethod
def bullet_point(text: str, indent: int = 0) -> None:
"""Display bullet point."""
prefix = " " * indent
print(f"{prefix}- {text}")
@staticmethod
def key_value(key: str, value: Any, indent: int = 0) -> None:
"""Display key-value pair."""
prefix = " " * indent
print(f"{prefix}{key}: {value}")
@staticmethod
def empty_line() -> None:
"""Display empty line."""
print()
@staticmethod
def exit_with_error(message: str, exit_code: int = 1) -> None:
"""Display error and exit."""
OutputFormatter.error(message)
sys.exit(exit_code)
@staticmethod
def format_file_list(files: List[str], title: str = "Files") -> None:
"""Format and display file list."""
print(f"📄 {title} ({len(files)}):")
if files:
for file in files:
print(f" - {file}")
else:
print(" - No files found")
print()
@staticmethod
def format_command_list(commands: List[str], title: str = "Commands") -> None:
"""Format and display command list."""
print(f"💡 {title}:")
for command in commands:
print(f" - {command}")
print()

323
cli/presenters/views.py Normal file
View File

@@ -0,0 +1,323 @@
"""
View models for displaying complex data structures.
"""
from typing import Dict, Any, List
from .formatters import OutputFormatter
class WorkspaceView:
"""View for workspace information display."""
@staticmethod
def show_status(summary: Dict[str, Any]) -> None:
"""Display workspace status."""
if summary['clean']:
OutputFormatter.info("No active issue workspace")
print(" Use 'make tdd-start NUM=X' to begin working on an issue")
return
if summary['dirty']:
OutputFormatter.warning("Workspace directory exists but no current issue file")
print(" Run 'make tdd-finish' to clean up or 'make tdd-start' to create new workspace")
return
if summary['active']:
WorkspaceView._show_active_workspace(summary)
else:
OutputFormatter.error("Failed to load workspace")
@staticmethod
def _show_active_workspace(summary: Dict[str, Any]) -> None:
"""Display active workspace details."""
OutputFormatter.header("Active Issue Workspace", "=")
issue_num = summary['issue_number']
issue_title = summary['issue_title']
print(f"🎯 Issue #{issue_num}: {issue_title}")
OutputFormatter.key_value("Status", summary['issue_state'])
OutputFormatter.key_value("Workspace", f"{summary['workspace_dir']}/issue_{issue_num}/")
OutputFormatter.empty_line()
# Test files
test_count = summary['test_count']
test_files = summary.get('test_files', [])
print(f"🧪 Generated Tests ({test_count}):")
if test_files:
for test_file in test_files:
print(f" - {test_file}")
else:
print(" - No tests generated yet")
OutputFormatter.empty_line()
# Workspace files
print("📋 Workspace Files:")
print(" - requirements.md (review and break down issue)")
print(" - test_plan.md (plan test scenarios)")
print(" - tests/ (generated test files)")
OutputFormatter.empty_line()
# Commands
commands = [
"make tdd-add-test (generate another test)",
"make tdd-finish (complete and move tests to main)"
]
OutputFormatter.format_command_list(commands)
@staticmethod
def show_start_success(summary: Dict[str, Any]) -> None:
"""Display successful workspace start."""
issue_num = summary['issue_number']
OutputFormatter.success(f"Workspace created for issue #{issue_num}")
OutputFormatter.key_value("Workspace", f"{summary['workspace_dir']}/issue_{issue_num}/")
OutputFormatter.key_value("Requirements", summary['requirements_file'])
OutputFormatter.key_value("Test plan", summary['test_plan_file'])
OutputFormatter.empty_line()
print("💡 Next steps:")
print(" 1. Review requirements.md and break down the issue")
print(" 2. Plan test scenarios in test_plan.md")
print(" 3. Use 'make tdd-add-test' to generate tests")
print(" 4. Use 'make tdd-finish' when complete")
@staticmethod
def show_finish_success(issue_number: int, test_count: int) -> None:
"""Display successful workspace finish."""
OutputFormatter.info(f"Finishing work on issue #{issue_number}")
OutputFormatter.empty_line()
if test_count > 0:
print(f"📦 Moving {test_count} test(s) to tests/ directory...")
OutputFormatter.success("Tests moved to main tests/ directory")
else:
OutputFormatter.warning("No tests found in workspace")
print("🧹 Cleaning up workspace...")
OutputFormatter.success(f"Issue #{issue_number} workspace cleaned up")
OutputFormatter.empty_line()
print("💡 Next steps:")
print(" - Run 'make test' to verify tests fail (red state)")
print(" - Implement code to make tests pass (green state)")
print(" - Start next issue with 'make tdd-start NUM=X'")
class IssueView:
"""View for issue information display."""
@staticmethod
def show_list(issues: List[Any], title: str = "Project Issues") -> None:
"""Display issue list."""
OutputFormatter.header(title, "=")
if not issues:
print("No issues found")
return
for issue in issues:
status_icon = "🟢" if issue.state == "open" else "🔴"
print(f"{status_icon} #{issue.number}: {issue.title}")
print(f" Status: {issue.state.upper()} | Created: {issue.created_at.strftime('%Y-%m-%d')}")
# Truncate body for list view
body_preview = issue.body[:80] + "..." if len(issue.body) > 80 else issue.body
if body_preview:
print(f" {body_preview}")
OutputFormatter.empty_line()
print("💡 Tip: Use 'make show-issue NUM=X' for full details")
@staticmethod
def show_open_issues(issues: List[Any]) -> None:
"""Display open issues list."""
OutputFormatter.header("Open Project Issues (Active Backlog)", "=")
if not issues:
print("No open issues found")
return
for issue in issues:
print(f"[OPEN] #{issue.number}: {issue.title}")
created = issue.created_at.strftime('%Y-%m-%d')
updated = issue.updated_at.strftime('%Y-%m-%d')
print(f" Created: {created} | Updated: {updated}")
# Truncate body for list view
body_preview = issue.body[:80] + "..." if len(issue.body) > 80 else issue.body
if body_preview:
print(f" {body_preview}")
OutputFormatter.empty_line()
print("💡 Tip: Use 'make show-issue NUM=X' for full details or 'make list-issues' for all issues")
@staticmethod
def show_creation_success(result: Dict[str, Any], issue_type: str = "issue") -> None:
"""Display successful issue creation."""
OutputFormatter.success(f"{issue_type.title()} created successfully!")
OutputFormatter.key_value("Number", f"#{result['number']}")
OutputFormatter.key_value("Title", result['title'])
OutputFormatter.key_value("Status", result['state'])
if 'html_url' in result:
OutputFormatter.key_value("URL", result['html_url'])
OutputFormatter.empty_line()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
@staticmethod
def show_issue_details(issue_data: Dict[str, Any]) -> None:
"""Display comprehensive issue details."""
OutputFormatter.header(f"Issue #{issue_data['number']} Details", "=")
print(f"**Title:** {issue_data['title']}")
print(f"**Status:** {issue_data['state'].upper()}")
print(f"**Number:** #{issue_data['number']}")
print(f"**Created:** {issue_data['created_at'].strftime('%Y-%m-%d %H:%M')}")
print(f"**Updated:** {issue_data['updated_at'].strftime('%Y-%m-%d %H:%M')}")
print(f"**URL:** {issue_data['html_url']}")
if issue_data['assignee']:
print(f"**Assignee:** {issue_data['assignee']}")
OutputFormatter.empty_line()
print("**Project Management:**")
# Milestone information
if issue_data['milestone']:
milestone = issue_data['milestone']
print(f" 📋 Milestone: #{milestone['id']} - {milestone['title']} ({milestone['state']})")
else:
print(" 📋 Milestone: None")
# Project/Board information
print(" 🎯 Project: Getting Started (assumed - requires board API)")
# Labels and state information
print(f" 📊 State: {issue_data['state_label']}")
print(f" 🚨 Priority: {issue_data['priority_label']}")
if issue_data['type_labels']:
type_display = ', '.join(issue_data['type_labels'])
print(f" 🏷️ Type: {type_display}")
if issue_data['other_labels']:
print(f" 🏷️ Other Labels: {', '.join(issue_data['other_labels'])}")
print(f" 📝 Kanban Column: {issue_data['kanban_column']}")
OutputFormatter.empty_line()
print("**Description:**")
print(issue_data['body'])
OutputFormatter.empty_line()
print("💡 Tip: Use 'make list-issues' to see all issues")
@staticmethod
def show_coverage_analysis(coverage_data: Dict[str, Any]) -> None:
"""Display test coverage analysis results."""
OutputFormatter.header(f"Test Coverage Analysis for Issue #{coverage_data['issue_number']}", "=")
print(f"📋 Issue: #{coverage_data['issue_number']} - {coverage_data['issue_title']}")
print(f"📊 Coverage: {coverage_data['coverage_percentage']:.1f}%")
OutputFormatter.empty_line()
# Show requirements analysis
print("🎯 Identified Requirements:")
if coverage_data['requirements']:
for req in coverage_data['requirements']:
priority_icon = {"critical": "🚨", "important": "⚠️", "nice-to-have": "💡"}
icon = priority_icon.get(req.priority, "📝")
print(f" {icon} [{req.priority.upper()}] {req.category}: {req.description}")
else:
print(" No specific requirements detected")
OutputFormatter.empty_line()
# Show existing tests
print("🧪 Existing Test Coverage:")
issue_number = coverage_data['issue_number']
issue_related_tests = [t for t in coverage_data['existing_tests'] if t.related_issue == issue_number]
if issue_related_tests:
for test in issue_related_tests:
test_count = len(test.test_methods)
print(f"{test.file_path.name} ({test_count} test methods)")
if test.test_methods:
for method in test.test_methods[:3]: # Show first 3
print(f" - {method}")
if len(test.test_methods) > 3:
print(f" - ... and {len(test.test_methods) - 3} more")
else:
print(" 📝 No tests specifically for this issue found")
# Show general tests that might be relevant
relevant_tests = [t for t in coverage_data['existing_tests']
if any(keyword in ' '.join(t.coverage_keywords)
for req in coverage_data['requirements']
for keyword in req.keywords)]
if relevant_tests:
print(" 📋 Potentially relevant tests:")
for test in relevant_tests[:3]:
print(f" 📄 {test.file_path.name}")
OutputFormatter.empty_line()
# Show coverage gaps
if coverage_data['coverage_gaps']:
print("❌ Coverage Gaps Found:")
for gap in coverage_data['coverage_gaps']:
priority_icon = {"critical": "🚨", "important": "⚠️", "nice-to-have": "💡"}
icon = priority_icon.get(gap.requirement.priority, "📝")
print(f" {icon} Missing: {gap.requirement.description}")
print(f" 💡 Suggested test: {gap.suggested_test_name}")
print(f" 📄 Suggested file: {gap.suggested_test_file}")
OutputFormatter.empty_line()
else:
print("✅ No significant coverage gaps detected!")
OutputFormatter.empty_line()
# Show recommendations
print("📝 Recommendations:")
for recommendation in coverage_data['recommendations']:
print(f" {recommendation}")
class ProjectView:
"""View for project management information display."""
@staticmethod
def show_setup_success() -> None:
"""Display successful project setup."""
OutputFormatter.success("Project management setup complete!")
print("📋 Available states: todo, active, review, done, blocked")
print("📊 Available priorities: low, medium, high, critical")
@staticmethod
def show_milestone_list(milestones: List[Any]) -> None:
"""Display milestone list."""
OutputFormatter.header("Project Milestones", "=")
if not milestones:
print("No milestones found")
return
for milestone in milestones:
status_icon = "🟢" if milestone.state == "open" else "🔴"
print(f"{status_icon} Milestone #{milestone.id}: {milestone.title}")
print(f" State: {milestone.state.upper()}")
print(f" Issues: {milestone.open_issues} open, {milestone.closed_issues} closed")
if milestone.description:
print(f" Description: {milestone.description}")
if milestone.due_on:
print(f" Due: {milestone.due_on}")
OutputFormatter.empty_line()
@staticmethod
def show_overview(overview: Dict[str, Any]) -> None:
"""Display project overview."""
OutputFormatter.header("Project Management Overview", "=")
OutputFormatter.key_value("Milestones", f"{overview['milestones']} total")
OutputFormatter.key_value("Active Projects", overview['active_projects'], 1)
OutputFormatter.key_value("Completed Projects", overview['completed_projects'], 1)
OutputFormatter.key_value("Total Labels", overview['total_labels'])
ready_status = "✅ Yes" if overview['project_management_ready'] else "❌ No - run setup-project-mgmt"
OutputFormatter.key_value("Project Management Ready", ready_status)

33
gitea/__init__.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Gitea API facade - Clean interface for Gitea repository operations.
This package provides a clean, well-structured interface to Gitea API operations,
following the facade pattern to decouple application logic from specific API
implementation details.
Structure:
- client: Main GiteaClient facade
- models: Domain models (Issue, Milestone, Label, etc.)
- config: Gitea-specific configuration
- exceptions: Gitea-specific exceptions
Usage:
from gitea import GiteaClient
client = GiteaClient()
issues = client.issues.list()
issue = client.issues.get(42)
client.issues.create("Bug fix", "Description")
"""
from .client import GiteaClient
from .models import Issue, Milestone, Label, ProjectState, Priority
from .config import GiteaConfig
from .exceptions import GiteaError, GiteaAuthError, GiteaNotFoundError
__all__ = [
'GiteaClient',
'Issue', 'Milestone', 'Label', 'ProjectState', 'Priority',
'GiteaConfig',
'GiteaError', 'GiteaAuthError', 'GiteaNotFoundError'
]

203
gitea/api_client.py Normal file
View File

@@ -0,0 +1,203 @@
"""
High-level API client that converts between API responses and domain models.
"""
from datetime import datetime
from typing import List, Optional, Dict, Any
from .http_client import GiteaHttpClient
from .models import Issue, Milestone, Label, User, IssueCreateData, IssueUpdateData, MilestoneCreateData, LabelCreateData
from .config import GiteaConfig
from .exceptions import GiteaNotFoundError, GiteaError
class GiteaApiClient:
"""High-level API client with domain model conversion."""
def __init__(self, config: GiteaConfig):
self.config = config
self.http = GiteaHttpClient(config)
# Issue operations
def get_issue(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
try:
url = f"{self.config.issues_api_url}/{issue_number}"
data = self.http.get(url)
return self._parse_issue(data)
except GiteaError as e:
if "not found" in str(e).lower():
raise GiteaNotFoundError(f"Issue #{issue_number} not found")
raise
def list_issues(self, state: str = "all", page: int = 1, per_page: int = 50) -> List[Issue]:
"""List issues with optional filtering."""
params = {"page": str(page), "limit": str(per_page)}
if state != "all":
params["state"] = state
data = self.http.get(self.config.issues_api_url, params)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of issues")
return [self._parse_issue(issue_data) for issue_data in data]
def create_issue(self, issue_data: IssueCreateData) -> Issue:
"""Create a new issue."""
payload = {
"title": issue_data.title,
"body": issue_data.body,
}
if issue_data.assignees:
payload["assignees"] = issue_data.assignees
if issue_data.milestone:
payload["milestone"] = issue_data.milestone
if issue_data.labels:
payload["labels"] = issue_data.labels
data = self.http.post(self.config.issues_api_url, payload)
return self._parse_issue(data)
def update_issue(self, issue_number: int, update_data: IssueUpdateData) -> Issue:
"""Update an existing issue."""
payload = {}
if update_data.title is not None:
payload["title"] = update_data.title
if update_data.body is not None:
payload["body"] = update_data.body
if update_data.state is not None:
payload["state"] = update_data.state
if update_data.assignees is not None:
payload["assignees"] = update_data.assignees
if update_data.milestone is not None:
payload["milestone"] = update_data.milestone
if update_data.labels is not None:
payload["labels"] = update_data.labels
url = f"{self.config.issues_api_url}/{issue_number}"
data = self.http.patch(url, payload)
return self._parse_issue(data)
# Milestone operations
def list_milestones(self, state: str = "all") -> List[Milestone]:
"""List repository milestones."""
params = {}
if state != "all":
params["state"] = state
data = self.http.get(self.config.milestones_api_url, params)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of milestones")
return [self._parse_milestone(milestone_data) for milestone_data in data]
def create_milestone(self, milestone_data: MilestoneCreateData) -> Milestone:
"""Create a new milestone."""
payload = {
"title": milestone_data.title,
"description": milestone_data.description,
}
if milestone_data.due_on:
payload["due_on"] = milestone_data.due_on
data = self.http.post(self.config.milestones_api_url, payload)
return self._parse_milestone(data)
# Label operations
def list_labels(self) -> List[Label]:
"""List repository labels."""
data = self.http.get(self.config.labels_api_url)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of labels")
return [self._parse_label(label_data) for label_data in data]
def create_label(self, label_data: LabelCreateData) -> Label:
"""Create a new label."""
payload = {
"name": label_data.name,
"color": label_data.color,
"description": label_data.description,
}
data = self.http.post(self.config.labels_api_url, payload)
return self._parse_label(data)
# Parsing methods
def _parse_issue(self, data: Dict[str, Any]) -> Issue:
"""Parse issue data from API response."""
try:
# Parse labels
labels = []
if data.get('labels'):
labels = [self._parse_label(label_data) for label_data in data['labels']]
# Parse assignee
assignee = None
if data.get('assignee'):
assignee = self._parse_user(data['assignee'])
# Parse milestone
milestone = None
if data.get('milestone'):
milestone = self._parse_milestone(data['milestone'])
return Issue(
number=data['number'],
title=data['title'],
body=data.get('body', ''),
state=data['state'],
created_at=self._parse_datetime(data['created_at']),
updated_at=self._parse_datetime(data['updated_at']),
html_url=data['html_url'],
assignee=assignee,
labels=labels,
milestone=milestone
)
except (KeyError, ValueError) as e:
raise GiteaError(f"Failed to parse issue data: {e}")
def _parse_milestone(self, data: Dict[str, Any]) -> Milestone:
"""Parse milestone data from API response."""
return Milestone(
id=data['id'],
title=data['title'],
description=data.get('description', ''),
state=data['state'],
open_issues=data.get('open_issues', 0),
closed_issues=data.get('closed_issues', 0),
due_on=data.get('due_on'),
created_at=self._parse_datetime(data.get('created_at')) if data.get('created_at') else None,
updated_at=self._parse_datetime(data.get('updated_at')) if data.get('updated_at') else None
)
def _parse_label(self, data: Dict[str, Any]) -> Label:
"""Parse label data from API response."""
return Label(
id=data['id'],
name=data['name'],
color=data['color'],
description=data.get('description', '')
)
def _parse_user(self, data: Dict[str, Any]) -> User:
"""Parse user data from API response."""
return User(
id=data['id'],
login=data['login'],
full_name=data.get('full_name', ''),
email=data.get('email', ''),
avatar_url=data.get('avatar_url', '')
)
def _parse_datetime(self, date_str: str) -> datetime:
"""Parse datetime from API response."""
# Remove Z and microseconds for consistent parsing
date_str = date_str.replace('Z', '').split('.')[0]
return datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S')

195
gitea/client.py Normal file
View File

@@ -0,0 +1,195 @@
"""
Main Gitea client facade.
This provides a clean, organized interface for all Gitea operations,
following the facade pattern to hide complexity and provide a stable API.
"""
from typing import List, Optional
from .config import GiteaConfig
from .api_client import GiteaApiClient
from .models import Issue, Milestone, Label, IssueCreateData, IssueUpdateData, MilestoneCreateData, LabelCreateData, ProjectState, Priority
class IssuesClient:
"""Client for issue operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def get(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
return self._api.get_issue(issue_number)
def list(self, state: str = "all", page: int = 1, per_page: int = 50) -> List[Issue]:
"""List issues with optional filtering."""
return self._api.list_issues(state, page, per_page)
def list_open(self) -> List[Issue]:
"""List only open issues."""
return self._api.list_issues("open")
def list_closed(self) -> List[Issue]:
"""List only closed issues."""
return self._api.list_issues("closed")
def create(self, title: str, body: str = "", **kwargs) -> Issue:
"""Create a new issue."""
issue_data = IssueCreateData(
title=title,
body=body,
assignees=kwargs.get('assignees', []),
milestone=kwargs.get('milestone'),
labels=kwargs.get('labels', [])
)
return self._api.create_issue(issue_data)
def update(self, issue_number: int, **kwargs) -> Issue:
"""Update an existing issue."""
update_data = IssueUpdateData(
title=kwargs.get('title'),
body=kwargs.get('body'),
state=kwargs.get('state'),
assignees=kwargs.get('assignees'),
milestone=kwargs.get('milestone'),
labels=kwargs.get('labels')
)
return self._api.update_issue(issue_number, update_data)
def close(self, issue_number: int) -> Issue:
"""Close an issue."""
return self.update(issue_number, state="closed")
def reopen(self, issue_number: int) -> Issue:
"""Reopen an issue."""
return self.update(issue_number, state="open")
def add_labels(self, issue_number: int, labels: List[str]) -> Issue:
"""Add labels to an issue."""
issue = self.get(issue_number)
existing_labels = [label.name for label in issue.labels]
new_labels = list(set(existing_labels + labels))
return self.update(issue_number, labels=new_labels)
def remove_labels(self, issue_number: int, labels: List[str]) -> Issue:
"""Remove labels from an issue."""
issue = self.get(issue_number)
existing_labels = [label.name for label in issue.labels]
new_labels = [label for label in existing_labels if label not in labels]
return self.update(issue_number, labels=new_labels)
def set_priority(self, issue_number: int, priority: Priority) -> Issue:
"""Set issue priority."""
issue = self.get(issue_number)
labels = [label.name for label in issue.labels if not label.name.startswith('priority:')]
labels.append(priority.value)
return self.update(issue_number, labels=labels)
def set_status(self, issue_number: int, status: ProjectState) -> Issue:
"""Set issue status."""
issue = self.get(issue_number)
labels = [label.name for label in issue.labels if not label.name.startswith('status:')]
labels.append(status.value)
return self.update(issue_number, labels=labels)
class MilestonesClient:
"""Client for milestone operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def list(self, state: str = "all") -> List[Milestone]:
"""List milestones."""
return self._api.list_milestones(state)
def list_open(self) -> List[Milestone]:
"""List open milestones."""
return self._api.list_milestones("open")
def list_closed(self) -> List[Milestone]:
"""List closed milestones."""
return self._api.list_milestones("closed")
def create(self, title: str, description: str = "", due_on: str = None) -> Milestone:
"""Create a new milestone."""
milestone_data = MilestoneCreateData(
title=title,
description=description,
due_on=due_on
)
return self._api.create_milestone(milestone_data)
class LabelsClient:
"""Client for label operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def list(self) -> List[Label]:
"""List all labels."""
return self._api.list_labels()
def create(self, name: str, color: str, description: str = "") -> Label:
"""Create a new label."""
label_data = LabelCreateData(
name=name,
color=color,
description=description
)
return self._api.create_label(label_data)
def ensure_project_labels(self) -> None:
"""Ensure all standard project management labels exist."""
existing_labels = [label.name for label in self.list()]
# Define standard project labels
standard_labels = [
("status:todo", "d73a4a", "Ready to work on"),
("status:active", "0075ca", "Currently being worked on"),
("status:review", "fbca04", "Ready for review"),
("status:done", "0e8a16", "Completed work"),
("status:blocked", "b60205", "Blocked by dependencies"),
("priority:low", "c5def5", "Low priority"),
("priority:medium", "a2eeef", "Medium priority"),
("priority:high", "fef2c0", "High priority"),
("priority:critical", "d93f0b", "Critical priority"),
]
for name, color, description in standard_labels:
if name not in existing_labels:
self.create(name, color, description)
class GiteaClient:
"""Main Gitea client facade."""
def __init__(self, config: Optional[GiteaConfig] = None):
"""Initialize Gitea client.
Args:
config: GiteaConfig instance. If None, loads from environment.
"""
if config is None:
config = GiteaConfig.from_environment()
config.validate()
self.config = config
self._api = GiteaApiClient(config)
# Initialize sub-clients
self.issues = IssuesClient(self._api)
self.milestones = MilestonesClient(self._api)
self.labels = LabelsClient(self._api)
@classmethod
def from_tddai_config(cls, tddai_config) -> 'GiteaClient':
"""Create client from legacy TddaiConfig for backwards compatibility."""
gitea_config = GiteaConfig.from_tddai_config(tddai_config)
return cls(gitea_config)
def setup_project_management(self) -> None:
"""Setup standard project management labels and structure."""
self.labels.ensure_project_labels()

113
gitea/config.py Normal file
View File

@@ -0,0 +1,113 @@
"""
Gitea-specific configuration management.
"""
import os
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
from .exceptions import GiteaConfigError
def load_dotenv_file(env_file: Path) -> None:
"""Load environment variables from a .env file."""
if not env_file.exists():
return
with open(env_file, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
os.environ.setdefault(key.strip(), value.strip())
@dataclass
class GiteaConfig:
"""Configuration for Gitea API access."""
# Repository settings (required)
gitea_url: str = ""
repo_owner: str = ""
repo_name: str = ""
# Authentication (optional for read operations)
auth_token: Optional[str] = None
@property
def base_api_url(self) -> str:
"""Get the base API URL for this repository."""
return f"{self.gitea_url}/api/v1"
@property
def repo_api_url(self) -> str:
"""Get the repository API URL."""
return f"{self.base_api_url}/repos/{self.repo_owner}/{self.repo_name}"
@property
def issues_api_url(self) -> str:
"""Get the issues API URL."""
return f"{self.repo_api_url}/issues"
@property
def milestones_api_url(self) -> str:
"""Get the milestones API URL."""
return f"{self.repo_api_url}/milestones"
@property
def labels_api_url(self) -> str:
"""Get the labels API URL."""
return f"{self.repo_api_url}/labels"
@classmethod
def from_environment(cls, env_prefix: str = "GITEA") -> "GiteaConfig":
"""Create config from environment variables.
Args:
env_prefix: Environment variable prefix (default: GITEA)
Looks for {prefix}_URL, {prefix}_REPO_OWNER, etc.
"""
# Auto-load .env.gitea file if it exists
env_file = Path(".env.gitea")
load_dotenv_file(env_file)
config = cls()
# Load from environment
config.gitea_url = os.getenv(f"{env_prefix}_URL", "")
config.repo_owner = os.getenv(f"{env_prefix}_REPO_OWNER", "")
config.repo_name = os.getenv(f"{env_prefix}_REPO_NAME", "")
config.auth_token = os.getenv(f"{env_prefix}_API_TOKEN")
return config
@classmethod
def from_tddai_config(cls, tddai_config) -> "GiteaConfig":
"""Create GiteaConfig from legacy TddaiConfig for backwards compatibility."""
return cls(
gitea_url=tddai_config.gitea_url,
repo_owner=tddai_config.repo_owner,
repo_name=tddai_config.repo_name,
auth_token=os.getenv('GITEA_API_TOKEN')
)
def validate(self) -> None:
"""Validate configuration settings."""
if not self.gitea_url:
raise GiteaConfigError("gitea_url cannot be empty")
if not self.repo_owner:
raise GiteaConfigError("repo_owner cannot be empty")
if not self.repo_name:
raise GiteaConfigError("repo_name cannot be empty")
# Validate URL format
if not (self.gitea_url.startswith('http://') or self.gitea_url.startswith('https://')):
raise GiteaConfigError("gitea_url must start with http:// or https://")
def requires_auth(self, operation: str = "read") -> bool:
"""Check if operation requires authentication."""
write_operations = {"create", "update", "delete", "write"}
return operation in write_operations and not self.auth_token

31
gitea/exceptions.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Gitea-specific exceptions.
"""
class GiteaError(Exception):
"""Base exception for Gitea API operations."""
pass
class GiteaAuthError(GiteaError):
"""Raised when authentication fails or token is missing."""
pass
class GiteaNotFoundError(GiteaError):
"""Raised when requested resource is not found."""
pass
class GiteaApiError(GiteaError):
"""Raised when API returns an error response."""
def __init__(self, message: str, status_code: int = None):
super().__init__(message)
self.status_code = status_code
class GiteaConfigError(GiteaError):
"""Raised when Gitea configuration is invalid or missing."""
pass

98
gitea/http_client.py Normal file
View File

@@ -0,0 +1,98 @@
"""
Low-level HTTP client for Gitea API operations.
This module handles the actual HTTP requests to Gitea API using subprocess + curl
for maximum compatibility and minimal dependencies.
"""
import json
import subprocess
from subprocess import PIPE
from typing import Dict, Any, Optional, List
from .exceptions import GiteaError, GiteaApiError, GiteaAuthError
from .config import GiteaConfig
class GiteaHttpClient:
"""Low-level HTTP client for Gitea API."""
def __init__(self, config: GiteaConfig):
self.config = config
def get(self, url: str, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
"""Make GET request to Gitea API."""
if params:
param_string = '&'.join(f"{k}={v}" for k, v in params.items())
url = f"{url}?{param_string}"
return self._make_request('GET', url)
def post(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make POST request to Gitea API."""
self._require_auth()
return self._make_request('POST', url, data)
def patch(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make PATCH request to Gitea API."""
self._require_auth()
return self._make_request('PATCH', url, data)
def delete(self, url: str) -> Dict[str, Any]:
"""Make DELETE request to Gitea API."""
self._require_auth()
return self._make_request('DELETE', url)
def _make_request(self, method: str, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make HTTP request using curl."""
cmd = ['curl', '-s', '-X', method]
# Add authentication if available
if self.config.auth_token:
cmd.extend(['-H', f'Authorization: token {self.config.auth_token}'])
# Add content type for requests with data
if data is not None:
cmd.extend(['-H', 'Content-Type: application/json'])
cmd.extend(['-d', json.dumps(data)])
cmd.append(url)
try:
result = subprocess.run(
cmd,
stdout=PIPE,
stderr=PIPE,
universal_newlines=True,
check=True
)
if result.returncode != 0:
raise GiteaApiError(f"HTTP request failed: {result.stderr}")
# Handle empty responses
if not result.stdout.strip():
return {}
response_data = json.loads(result.stdout)
# Check for API error responses
if isinstance(response_data, dict):
if 'message' in response_data:
# This could be an error or just a response with a message field
# We need to distinguish based on context or HTTP status
if any(error_word in response_data['message'].lower()
for error_word in ['error', 'not found', 'forbidden', 'unauthorized']):
raise GiteaApiError(response_data['message'])
return response_data
except subprocess.CalledProcessError as e:
raise GiteaApiError(f"HTTP request failed: {e.stderr}")
except json.JSONDecodeError as e:
raise GiteaError(f"Failed to parse API response: {e}")
def _require_auth(self):
"""Ensure authentication token is available."""
if not self.config.auth_token:
raise GiteaAuthError("Authentication token required for this operation")

151
gitea/models.py Normal file
View File

@@ -0,0 +1,151 @@
"""
Gitea domain models.
These models represent the core entities in Gitea and provide a clean interface
independent of the underlying API representation.
"""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
class ProjectState(Enum):
"""Standard project states using labels."""
TODO = "status:todo"
ACTIVE = "status:active"
REVIEW = "status:review"
DONE = "status:done"
BLOCKED = "status:blocked"
class Priority(Enum):
"""Priority levels using labels."""
LOW = "priority:low"
MEDIUM = "priority:medium"
HIGH = "priority:high"
CRITICAL = "priority:critical"
@dataclass
class Label:
"""Represents a Gitea issue label."""
id: int
name: str
color: str
description: str = ""
@dataclass
class User:
"""Represents a Gitea user."""
id: int
login: str
full_name: str = ""
email: str = ""
avatar_url: str = ""
@dataclass
class Milestone:
"""Represents a Gitea milestone (used as projects)."""
id: int
title: str
description: str
state: str # 'open' or 'closed'
open_issues: int
closed_issues: int
due_on: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
@dataclass
class Issue:
"""Represents a Gitea issue."""
number: int
title: str
body: str
state: str # 'open' or 'closed'
created_at: datetime
updated_at: datetime
html_url: str
assignee: Optional[User] = None
labels: List[Label] = None
milestone: Optional[Milestone] = None
def __post_init__(self):
if self.labels is None:
self.labels = []
@property
def priority(self) -> Optional[str]:
"""Get issue priority from labels."""
for label in self.labels:
if label.name.startswith('priority:'):
return label.name.replace('priority:', '')
return None
@property
def status(self) -> Optional[str]:
"""Get issue status from labels."""
for label in self.labels:
if label.name.startswith('status:'):
return label.name.replace('status:', '')
return None
def has_label(self, label_name: str) -> bool:
"""Check if issue has a specific label."""
return any(label.name == label_name for label in self.labels)
def has_priority(self, priority: Priority) -> bool:
"""Check if issue has a specific priority."""
return self.has_label(priority.value)
def has_status(self, status: ProjectState) -> bool:
"""Check if issue has a specific status."""
return self.has_label(status.value)
@dataclass
class IssueCreateData:
"""Data for creating a new issue."""
title: str
body: str = ""
assignees: List[str] = None
milestone: Optional[int] = None
labels: List[str] = None
def __post_init__(self):
if self.assignees is None:
self.assignees = []
if self.labels is None:
self.labels = []
@dataclass
class IssueUpdateData:
"""Data for updating an existing issue."""
title: Optional[str] = None
body: Optional[str] = None
state: Optional[str] = None
assignees: Optional[List[str]] = None
milestone: Optional[int] = None
labels: Optional[List[str]] = None
@dataclass
class MilestoneCreateData:
"""Data for creating a new milestone."""
title: str
description: str = ""
due_on: Optional[str] = None
@dataclass
class LabelCreateData:
"""Data for creating a new label."""
name: str
color: str
description: str = ""

View File

@@ -162,8 +162,15 @@ class CacheDirectoryService:
try:
cache_file.unlink()
removed_count += 1
except Exception as e:
except (OSError, PermissionError) as e:
errors.append(f"Could not remove {cache_file}: {e}")
except Exception as e:
# Log unexpected errors but continue cleanup
import logging
logging.getLogger(__name__).warning(
f"Unexpected error removing cache file {cache_file}: {e}"
)
errors.append(f"Unexpected error removing {cache_file}: {e}")
if errors:
return {
@@ -212,10 +219,22 @@ class CacheDirectoryService:
'file_removed': True,
'cache_file': str(cache_file)
}
except Exception as e:
except (OSError, PermissionError) as e:
return {
'success': False,
'message': f'Error removing cache for {source_path.name}: {e}',
'message': f'File system error removing cache for {source_path.name}: {e}',
'file_removed': False,
'error': str(e)
}
except Exception as e:
import logging
logging.getLogger(__name__).error(
f"Unexpected error removing cache for {source_path.name}: {e}",
exc_info=True
)
return {
'success': False,
'message': f'Unexpected error removing cache for {source_path.name}: {e}',
'file_removed': False,
'error': str(e)
}

127
markitect/exceptions.py Normal file
View File

@@ -0,0 +1,127 @@
"""
Markitect domain-specific exceptions.
This module provides a hierarchy of exceptions for the Markitect markdown processing system.
All exceptions preserve context and support proper exception chaining.
"""
from typing import Optional, Dict, Any
class MarkitectError(Exception):
"""Base exception for all Markitect operations.
Provides enhanced error context and proper exception chaining support.
Args:
message: Human-readable error description
cause: Original exception that caused this error (for chaining)
context: Additional context information as key-value pairs
"""
def __init__(self, message: str, cause: Optional[Exception] = None, context: Optional[Dict[str, Any]] = None):
super().__init__(message)
self.cause = cause
self.context = context or {}
# Automatically chain if cause is provided
if cause:
self.__cause__ = cause
def __str__(self) -> str:
"""Enhanced string representation with context."""
base_message = super().__str__()
if self.context:
context_info = ", ".join(f"{k}={v}" for k, v in self.context.items())
base_message = f"{base_message} [Context: {context_info}]"
return base_message
class DocumentError(MarkitectError):
"""Errors related to document processing and management.
Raised when:
- Document parsing fails
- Document structure is invalid
- Document metadata is corrupt
"""
pass
class ASTError(MarkitectError):
"""Errors related to Abstract Syntax Tree operations.
Raised when:
- AST parsing fails
- AST structure is invalid
- AST transformation fails
"""
pass
class CacheError(MarkitectError):
"""Errors related to cache operations.
Raised when:
- Cache read/write operations fail
- Cache corruption is detected
- Cache invalidation fails
"""
pass
class DatabaseError(MarkitectError):
"""Errors related to database operations.
Raised when:
- Database connection fails
- Query execution fails
- Data integrity violations occur
"""
pass
class SchemaError(MarkitectError):
"""Errors related to schema validation and processing.
Raised when:
- Schema validation fails
- Schema parsing errors occur
- Schema generation fails
"""
pass
class ValidationError(MarkitectError):
"""Errors related to document validation against schemas.
Raised when:
- Document doesn't match schema
- Validation rules are violated
- Required fields are missing
"""
pass
class GraphQLError(MarkitectError):
"""Errors related to GraphQL operations.
Raised when:
- GraphQL query parsing fails
- GraphQL execution errors occur
- GraphQL schema issues are encountered
"""
pass
class ConfigurationError(MarkitectError):
"""Errors related to configuration and setup.
Raised when:
- Configuration files are missing or invalid
- Environment setup is incomplete
- Required settings are not configured
"""
pass

28
services/__init__.py Normal file
View File

@@ -0,0 +1,28 @@
"""
Business logic services layer.
This package contains pure business logic services that are independent of
CLI presentation concerns. Services focus on:
- Core business operations
- Data transformation
- Validation and error handling
- Integration with lower-level modules
Services should NOT:
- Handle CLI arguments directly
- Print output or format data for display
- Call sys.exit() or handle CLI-specific errors
"""
from .workspace_service import WorkspaceService
from .issue_service import IssueService
from .project_service import ProjectService
from .export_service import ExportService
__all__ = [
'WorkspaceService',
'IssueService',
'ProjectService',
'ExportService'
]

150
services/export_service.py Normal file
View File

@@ -0,0 +1,150 @@
"""
Export service - business logic for data export and formatting operations.
"""
import json
from typing import List, Dict, Any, Optional
from datetime import datetime
from gitea.models import Issue
from .issue_service import IssueService
class ExportService:
"""Service for export and data formatting operations."""
def __init__(self):
self.issue_service = IssueService()
def get_issues_data(self, state: str = "all",
sort_by: str = "number",
filter_state: Optional[str] = None,
filter_priority: Optional[str] = None,
include_state: bool = False) -> List[Dict[str, Any]]:
"""Get structured issue data for export.
Args:
state: Issue state filter (all, open, closed)
sort_by: Sort field (number, title, priority, state, created, updated)
filter_state: Additional state filter (open, closed)
filter_priority: Priority filter (low, medium, high, critical, none)
include_state: Whether to include detailed state information
Returns:
List of issue data dictionaries
"""
issues = self.issue_service.list_issues(state)
# Convert to structured data
issue_data = []
for issue in issues:
# Get priority and state from labels
priority = issue.priority or "none"
status = issue.status or "none"
issue_info = {
'number': issue.number,
'title': issue.title.replace('\t', ' ').replace('\n', ' '), # Clean for TSV
'priority': priority,
'state': issue.state, # open/closed from basic data
'status': status, # detailed status from labels
'created': issue.created_at.strftime('%Y-%m-%d'),
'updated': issue.updated_at.strftime('%Y-%m-%d')
}
issue_data.append(issue_info)
# Apply filters
if filter_state:
if filter_state == "open":
issue_data = [i for i in issue_data if i['state'] == 'open']
elif filter_state == "closed":
issue_data = [i for i in issue_data if i['state'] == 'closed']
if filter_priority:
issue_data = [i for i in issue_data if i['priority'] == filter_priority]
# Sort issues
sort_key_map = {
'number': lambda x: x['number'],
'title': lambda x: x['title'].lower(),
'priority': lambda x: {'critical': 4, 'high': 3, 'medium': 2, 'low': 1, 'none': 0}[x['priority']],
'state': lambda x: x['state'],
'created': lambda x: x['created'],
'updated': lambda x: x['updated']
}
if sort_by in sort_key_map:
issue_data.sort(key=sort_key_map[sort_by], reverse=(sort_by in ['number', 'priority', 'created', 'updated']))
return issue_data
def format_issues_tsv(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
"""Format issues as TSV."""
lines = []
for issue in issue_data:
if include_state:
lines.append(f'{issue["number"]}\t{issue["title"]}\t{issue["priority"]}\t{issue["state"]}\t{issue["created"]}\t{issue["updated"]}')
else:
lines.append(f'{issue["number"]}\t{issue["title"]}\t{issue["priority"]}\t{issue["created"]}\t{issue["updated"]}')
return '\n'.join(lines)
def format_issues_csv(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
"""Format issues as CSV."""
lines = []
# Header
if include_state:
lines.append("number,title,priority,state,created,updated")
for issue in issue_data:
title = issue['title'].replace('"', '""') # Escape quotes
lines.append(f'{issue["number"]},"{title}",{issue["priority"]},{issue["state"]},{issue["created"]},{issue["updated"]}')
else:
lines.append("number,title,priority,created,updated")
for issue in issue_data:
title = issue['title'].replace('"', '""')
lines.append(f'{issue["number"]},"{title}",{issue["priority"]},{issue["created"]},{issue["updated"]}')
return '\n'.join(lines)
def format_issues_json(self, issue_data: List[Dict[str, Any]]) -> str:
"""Format issues as JSON."""
return json.dumps(issue_data, indent=2)
def format_issues_fields(self, issue_data: List[Dict[str, Any]], include_state: bool = False) -> str:
"""Format issues as space-separated fields for awk processing."""
lines = []
# Header
if include_state:
lines.append("NUMBER TITLE PRIORITY STATE CREATED UPDATED")
for issue in issue_data:
title = issue['title'].replace(' ', '_')
lines.append(f'{issue["number"]} {title} {issue["priority"]} {issue["state"]} {issue["created"]} {issue["updated"]}')
else:
lines.append("NUMBER TITLE PRIORITY CREATED UPDATED")
for issue in issue_data:
title = issue['title'].replace(' ', '_')
lines.append(f'{issue["number"]} {title} {issue["priority"]} {issue["created"]} {issue["updated"]}')
return '\n'.join(lines)
def export_issues(self, format_type: str = "tsv", **kwargs) -> str:
"""Export issues in specified format.
Args:
format_type: Output format (tsv, csv, json, fields)
**kwargs: Export parameters (sort_by, filter_state, etc.)
Returns:
Formatted string output
"""
issue_data = self.get_issues_data(**kwargs)
if format_type == "json":
return self.format_issues_json(issue_data)
elif format_type == "csv":
return self.format_issues_csv(issue_data, kwargs.get('include_state', False))
elif format_type == "fields":
return self.format_issues_fields(issue_data, kwargs.get('include_state', False))
else: # Default TSV
return self.format_issues_tsv(issue_data, kwargs.get('include_state', False))

154
services/issue_service.py Normal file
View File

@@ -0,0 +1,154 @@
"""
Issue service - business logic for issue operations.
"""
from typing import List, Dict, Any, Optional
from datetime import datetime
from tddai import IssueFetcher, TddaiError
from tddai.issue_creator import IssueCreator
from gitea.models import Issue, Priority
class IssueService:
"""Service for issue operations."""
def __init__(self):
self.issue_fetcher = IssueFetcher()
self.issue_creator = IssueCreator()
def get_issue(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
return self.issue_fetcher.fetch_issue(issue_number)
def list_issues(self, state: str = "all") -> List[Issue]:
"""List issues with optional state filter."""
return self.issue_fetcher.fetch_issues(state)
def list_open_issues(self) -> List[Issue]:
"""List only open issues."""
return self.issue_fetcher.fetch_open_issues()
def create_issue(self, title: str, body: str, **kwargs) -> Dict[str, Any]:
"""Create a new issue."""
return self.issue_creator.create_issue(title, body, **kwargs)
def create_enhancement_issue(self, title: str, use_case: str,
technical_requirements: str = "",
acceptance_criteria: List[str] = None,
dependencies: List[str] = None,
priority: str = "Medium") -> Dict[str, Any]:
"""Create a structured enhancement issue."""
return self.issue_creator.create_enhancement_issue(
title, use_case, technical_requirements,
acceptance_criteria, dependencies, priority
)
def create_from_template(self, template_file: str, **kwargs) -> Dict[str, Any]:
"""Create issue from template file."""
return self.issue_creator.create_from_template(template_file, **kwargs)
def get_issue_details(self, issue_number: int) -> Dict[str, Any]:
"""Get comprehensive issue details for display purposes."""
issue = self.get_issue(issue_number)
# Get additional project management information
from tddai.project_manager import ProjectManager
project_mgr = ProjectManager()
# Get detailed issue data via API for milestone and project information
from tddai.config import get_config
config = get_config()
issue_url = f"{config.issues_api_url}/{issue_number}"
detailed_issue = project_mgr._make_api_call('GET', issue_url)
# Process labels
labels = detailed_issue.get('labels', [])
state_labels = [l['name'] for l in labels if l['name'].startswith('status:')]
priority_labels = [l['name'] for l in labels if l['name'].startswith('priority:')]
type_labels = [l['name'] for l in labels if l['name'].startswith('type:')]
other_labels = [l['name'] for l in labels if not any(l['name'].startswith(p) for p in ['status:', 'priority:', 'type:'])]
# Determine project column/state
if detailed_issue.get('state') == 'closed':
if any(l['name'] == 'status:done' for l in labels):
column = "Done"
else:
column = "Closed"
else:
state_labels = [l['name'] for l in labels if l['name'].startswith('status:')]
if state_labels:
state = state_labels[0].replace('status:', '')
column_map = {
'todo': 'Todo',
'active': 'Active',
'review': 'Review',
'blocked': 'Blocked'
}
column = column_map.get(state, 'Todo')
else:
column = "Todo"
return {
'number': issue.number,
'title': issue.title,
'body': issue.body,
'state': issue.state,
'created_at': issue.created_at,
'updated_at': issue.updated_at,
'html_url': issue.html_url,
'assignee': issue.assignee.login if issue.assignee else None,
'milestone': detailed_issue.get('milestone'),
'state_label': state_labels[0].replace('status:', '').title() if state_labels else "No state label",
'priority_label': priority_labels[0].replace('priority:', '').title() if priority_labels else "No priority set",
'type_labels': [l.replace('type:', '').title() for l in type_labels],
'other_labels': other_labels,
'kanban_column': column
}
def get_issue_summary(self, issue_number: int) -> Dict[str, Any]:
"""Get issue summary for display purposes."""
issue = self.get_issue(issue_number)
return {
'number': issue.number,
'title': issue.title,
'body': issue.body,
'state': issue.state,
'priority': issue.priority,
'status': issue.status,
'created_at': issue.created_at,
'updated_at': issue.updated_at,
'html_url': issue.html_url,
'assignee': issue.assignee.login if issue.assignee else None,
'labels': [label.name for label in issue.labels],
'has_milestone': issue.milestone is not None,
'milestone_title': issue.milestone.title if issue.milestone else None
}
def analyze_coverage(self, issue_number: int) -> Dict[str, Any]:
"""Analyze test coverage for a specific issue.
Args:
issue_number: The issue number to analyze
Returns:
Coverage analysis data for the issue
Raises:
IssueError: When coverage analysis fails
"""
from tddai.coverage_analyzer import CoverageAnalyzer
analyzer = CoverageAnalyzer()
assessment = analyzer.analyze_issue_coverage(issue_number)
return {
'issue_number': assessment.issue_number,
'issue_title': assessment.issue_title,
'coverage_percentage': assessment.coverage_percentage,
'requirements': assessment.requirements,
'existing_tests': assessment.existing_tests,
'coverage_gaps': assessment.coverage_gaps,
'recommendations': assessment.recommendations
}

View File

@@ -0,0 +1,76 @@
"""
Project service - business logic for project management operations.
"""
from typing import List, Dict, Any
from tddai.project_manager import ProjectManager, ProjectState, Priority, Milestone, Label
from tddai import TddaiError
class ProjectService:
"""Service for project management operations."""
def __init__(self):
self.project_manager = ProjectManager()
def setup_project_management(self) -> None:
"""Setup project management labels and structure."""
self.project_manager.ensure_project_labels()
def create_milestone(self, title: str, description: str = "", due_date: str = None) -> Milestone:
"""Create a new milestone (project)."""
return self.project_manager.create_milestone(title, description, due_date)
def list_milestones(self, state: str = "open") -> List[Milestone]:
"""List milestones."""
return self.project_manager.list_milestones(state)
def list_labels(self) -> List[Label]:
"""List repository labels."""
return self.project_manager.list_labels()
def set_issue_state(self, issue_number: int, state_name: str) -> Dict[str, Any]:
"""Set issue project state."""
# Convert string to ProjectState enum
state_map = {
'todo': ProjectState.TODO,
'active': ProjectState.ACTIVE,
'review': ProjectState.REVIEW,
'done': ProjectState.DONE,
'blocked': ProjectState.BLOCKED
}
if state_name not in state_map:
raise TddaiError(f"Invalid state '{state_name}'. Valid states: {list(state_map.keys())}")
project_state = state_map[state_name]
return self.project_manager.set_issue_state(issue_number, project_state)
def set_issue_priority(self, issue_number: int, priority_name: str) -> Dict[str, Any]:
"""Set issue priority."""
# Convert string to Priority enum
priority_map = {
'low': Priority.LOW,
'medium': Priority.MEDIUM,
'high': Priority.HIGH,
'critical': Priority.CRITICAL
}
if priority_name not in priority_map:
raise TddaiError(f"Invalid priority '{priority_name}'. Valid priorities: {list(priority_map.keys())}")
priority_level = priority_map[priority_name]
return self.project_manager.set_issue_priority(issue_number, priority_level)
def move_issue_to_done(self, issue_number: int) -> Dict[str, Any]:
"""Move issue to done state and close it."""
return self.project_manager.move_issue_to_done(issue_number)
def assign_issue_to_milestone(self, issue_number: int, milestone_id: int) -> Dict[str, Any]:
"""Assign issue to a milestone."""
return self.project_manager.assign_issue_to_milestone(issue_number, milestone_id)
def get_project_overview(self) -> Dict[str, Any]:
"""Get project management overview."""
return self.project_manager.get_project_overview()

View File

@@ -0,0 +1,116 @@
"""
Workspace service - business logic for TDD workspace operations.
"""
from typing import Optional, Dict, Any
from pathlib import Path
from tddai import WorkspaceManager, IssueFetcher, WorkspaceStatus, TddaiError
class WorkspaceInfo:
"""Value object for workspace information."""
def __init__(self, status: WorkspaceStatus, workspace=None):
self.status = status
self.workspace = workspace
@property
def is_clean(self) -> bool:
return self.status == WorkspaceStatus.CLEAN
@property
def is_dirty(self) -> bool:
return self.status == WorkspaceStatus.DIRTY
@property
def is_active(self) -> bool:
return self.status == WorkspaceStatus.ACTIVE
def get_test_files(self) -> list:
"""Get list of test files in workspace."""
if not self.workspace or not self.workspace.tests_dir.exists():
return []
return list(self.workspace.tests_dir.glob("*.py"))
class WorkspaceService:
"""Service for workspace operations."""
def __init__(self):
self.workspace_manager = WorkspaceManager()
self.issue_fetcher = IssueFetcher()
def get_workspace_info(self) -> WorkspaceInfo:
"""Get current workspace information."""
status = self.workspace_manager.get_status()
workspace = None
if status == WorkspaceStatus.ACTIVE:
workspace = self.workspace_manager.get_current_workspace()
return WorkspaceInfo(status, workspace)
def start_issue_workspace(self, issue_number: int) -> WorkspaceInfo:
"""Start working on an issue.
Returns:
WorkspaceInfo with the created workspace
Raises:
TddaiError: If workspace already active or issue cannot be fetched
"""
# Check if workspace already active
current_info = self.get_workspace_info()
if current_info.is_active:
raise TddaiError(f"Already working on issue #{current_info.workspace.issue_number}")
# Fetch issue data
issue_data = self.issue_fetcher.get_issue_data_dict(issue_number)
# Create workspace
workspace = self.workspace_manager.create_workspace(issue_data)
return WorkspaceInfo(WorkspaceStatus.ACTIVE, workspace)
def finish_current_workspace(self) -> Optional[int]:
"""Finish current workspace and return the issue number.
Returns:
Issue number that was finished, or None if no active workspace
Raises:
TddaiError: If workspace operations fail
"""
current_info = self.get_workspace_info()
if not current_info.is_active:
return None
issue_number = current_info.workspace.issue_number
self.workspace_manager.finish_workspace()
return issue_number
def get_workspace_summary(self) -> Dict[str, Any]:
"""Get workspace summary for display purposes."""
info = self.get_workspace_info()
summary = {
'status': info.status,
'active': info.is_active,
'clean': info.is_clean,
'dirty': info.is_dirty
}
if info.workspace:
test_files = info.get_test_files()
summary.update({
'issue_number': info.workspace.issue_number,
'issue_title': info.workspace.issue_title,
'issue_state': info.workspace.issue_state,
'workspace_dir': str(info.workspace.workspace_dir),
'test_count': len(test_files),
'test_files': [f.name for f in test_files],
'requirements_file': str(info.workspace.requirements_file),
'test_plan_file': str(info.workspace.test_plan_file)
})
return summary

View File

@@ -246,8 +246,21 @@ class CoverageAnalyzer:
coverage_keywords=coverage_keywords,
related_issue=related_issue
))
except (OSError, IOError, UnicodeDecodeError) as e:
# Skip files that can't be read due to file system or encoding issues
# Log the issue but continue processing other files
import logging
logging.getLogger(__name__).warning(
f"Could not read test file {test_file}: {e}"
)
continue
except Exception as e:
# Skip files that can't be read
# Unexpected errors should be logged but not silently ignored
import logging
logging.getLogger(__name__).error(
f"Unexpected error processing test file {test_file}: {e}",
exc_info=True
)
continue
return existing_tests

View File

@@ -1,24 +1,31 @@
"""
Issue creation for Gitea API.
Issue creation using the Gitea facade.
This module now acts as an adapter to the new gitea package,
maintaining backwards compatibility while using the cleaner API.
"""
import json
import os
import subprocess
from subprocess import PIPE
from typing import Dict, Any, Optional, List
from gitea import GiteaClient, GiteaConfig, Priority
from .config import get_config
from .exceptions import IssueError
class IssueCreator:
"""Creates new issues via Gitea API."""
"""Creates new issues using the Gitea facade."""
def __init__(self, config=None, auth_token=None):
self.config = config or get_config()
self.auth_token = auth_token or os.getenv('GITEA_API_TOKEN')
# Create Gitea client from tddai config
gitea_config = GiteaConfig.from_tddai_config(self.config)
if self.auth_token:
gitea_config.auth_token = self.auth_token
self.gitea_client = GiteaClient(gitea_config)
def create_issue(self, title: str, body: str, **kwargs) -> Dict[str, Any]:
"""Create a new issue via POST operation.
@@ -33,63 +40,34 @@ class IssueCreator:
Raises:
IssueError: If creation fails
"""
if not self.auth_token:
raise IssueError("Authentication token required for issue creation")
if not title.strip():
# Validate input
if not title or not title.strip():
raise IssueError("Issue title cannot be empty")
# Prepare issue data
issue_data = {
'title': title.strip(),
'body': body.strip() if body else ''
}
# Add optional fields
if 'assignees' in kwargs and kwargs['assignees']:
issue_data['assignees'] = kwargs['assignees']
if 'milestone' in kwargs and kwargs['milestone']:
issue_data['milestone'] = kwargs['milestone']
if 'labels' in kwargs and kwargs['labels']:
issue_data['labels'] = kwargs['labels']
url = self.config.issues_api_url
try:
# Prepare curl command with authentication
curl_cmd = [
'curl', '-s', '-X', 'POST',
'-H', 'Content-Type: application/json',
'-H', f'Authorization: token {self.auth_token}',
'-d', json.dumps(issue_data),
url
]
result = subprocess.run(
curl_cmd,
stdout=PIPE,
stderr=PIPE,
universal_newlines=True,
check=True
issue = self.gitea_client.issues.create(
title=title,
body=body,
assignees=kwargs.get('assignees', []),
milestone=kwargs.get('milestone'),
labels=kwargs.get('labels', [])
)
if result.returncode != 0:
raise IssueError(f"Failed to create issue: {result.stderr}")
# Convert back to dict format for backwards compatibility
return {
'number': issue.number,
'title': issue.title,
'body': issue.body,
'state': issue.state,
'html_url': issue.html_url,
'created_at': issue.created_at.isoformat(),
'updated_at': issue.updated_at.isoformat(),
'assignee': {'login': issue.assignee.login} if issue.assignee else None,
'labels': [{'name': label.name} for label in issue.labels]
}
response_data = json.loads(result.stdout)
# Check for API error responses
if 'message' in response_data and 'number' not in response_data:
raise IssueError(f"Failed to create issue: {response_data['message']}")
return response_data
except subprocess.CalledProcessError as e:
except Exception as e:
raise IssueError(f"Failed to create issue: {e}")
except json.JSONDecodeError as e:
raise IssueError(f"Failed to parse response data: {e}")
def create_enhancement_issue(self, title: str, use_case: str,
technical_requirements: str = "",

View File

@@ -1,127 +1,80 @@
"""
Issue fetching from Gitea API.
Issue fetching using the Gitea facade.
This module now acts as an adapter to the new gitea package,
maintaining backwards compatibility while using the cleaner API.
"""
import json
import subprocess
from subprocess import PIPE
from dataclasses import dataclass
from datetime import datetime
from typing import List, Optional, Dict, Any
from typing import List, Dict, Any
from gitea import GiteaClient, Issue as GiteaIssue, GiteaConfig
from gitea.exceptions import GiteaError, GiteaNotFoundError, GiteaAuthError, GiteaApiError
from .config import get_config
from .exceptions import IssueError
@dataclass
class Issue:
"""Represents a Gitea issue."""
number: int
title: str
body: str
state: str
created_at: datetime
updated_at: datetime
html_url: str
assignee: Optional[str] = None
labels: List[str] = None
def __post_init__(self):
if self.labels is None:
self.labels = []
# Re-export Issue for backwards compatibility
Issue = GiteaIssue
class IssueFetcher:
"""Fetches issues from Gitea API."""
"""Fetches issues using the Gitea facade."""
def __init__(self, config=None):
self.config = config or get_config()
# Create Gitea client from tddai config
gitea_config = GiteaConfig.from_tddai_config(self.config)
self.gitea_client = GiteaClient(gitea_config)
def fetch_issue(self, issue_number: int) -> Issue:
"""Fetch a specific issue by number."""
"""Fetch a specific issue by number.
Raises:
IssueError: When issue cannot be fetched (with specific context)
"""
try:
result = subprocess.run(
['curl', '-s', f"{self.config.issues_api_url}/{issue_number}"],
stdout=PIPE,
stderr=PIPE,
universal_newlines=True,
check=True
)
if result.returncode != 0:
raise IssueError(f"Failed to fetch issue #{issue_number}: {result.stderr}")
issue_data = json.loads(result.stdout)
if 'message' in issue_data:
raise IssueError(f"Issue #{issue_number} not found: {issue_data['message']}")
return self._parse_issue(issue_data)
except subprocess.CalledProcessError as e:
raise IssueError(f"Failed to fetch issue #{issue_number}: {e}")
except json.JSONDecodeError as e:
raise IssueError(f"Failed to parse issue data: {e}")
return self.gitea_client.issues.get(issue_number)
except GiteaNotFoundError as e:
raise IssueError(f"Issue #{issue_number} not found") from e
except GiteaAuthError as e:
raise IssueError(f"Authentication failed when fetching issue #{issue_number}") from e
except GiteaApiError as e:
raise IssueError(f"API error fetching issue #{issue_number}: {e}") from e
except GiteaError as e:
raise IssueError(f"Gitea error fetching issue #{issue_number}: {e}") from e
def fetch_issues(self, state: str = "all") -> List[Issue]:
"""Fetch all issues with optional state filter."""
"""Fetch all issues with optional state filter.
Args:
state: Issue state filter ("all", "open", "closed")
Raises:
IssueError: When issues cannot be fetched (with specific context)
"""
try:
url = self.config.issues_api_url
if state != "all":
url += f"?state={state}"
result = subprocess.run(
['curl', '-s', url],
stdout=PIPE,
stderr=PIPE,
universal_newlines=True,
check=True
)
if result.returncode != 0:
raise IssueError(f"Failed to fetch issues: {result.stderr}")
issues_data = json.loads(result.stdout)
if isinstance(issues_data, dict) and 'message' in issues_data:
raise IssueError(f"Failed to fetch issues: {issues_data['message']}")
if not isinstance(issues_data, list):
raise IssueError("Invalid response format: expected list of issues")
return [self._parse_issue(issue_data) for issue_data in issues_data]
except subprocess.CalledProcessError as e:
raise IssueError(f"Failed to fetch issues: {e}")
except json.JSONDecodeError as e:
raise IssueError(f"Failed to parse issues data: {e}")
return self.gitea_client.issues.list(state=state)
except GiteaAuthError as e:
raise IssueError("Authentication failed when fetching issues") from e
except GiteaApiError as e:
raise IssueError(f"API error fetching issues with state '{state}': {e}") from e
except GiteaError as e:
raise IssueError(f"Gitea error fetching issues: {e}") from e
def fetch_open_issues(self) -> List[Issue]:
"""Fetch only open issues."""
return self.fetch_issues(state="open")
"""Fetch only open issues.
def _parse_issue(self, issue_data: Dict[str, Any]) -> Issue:
"""Parse issue data from API response."""
Raises:
IssueError: When open issues cannot be fetched (with specific context)
"""
try:
labels = [label['name'] for label in issue_data.get('labels', [])]
assignee = None
if issue_data.get('assignee'):
assignee = issue_data['assignee'].get('login')
return Issue(
number=issue_data['number'],
title=issue_data['title'],
body=issue_data.get('body', ''),
state=issue_data['state'],
created_at=datetime.strptime(issue_data['created_at'].replace('Z', '').split('.')[0], '%Y-%m-%dT%H:%M:%S'),
updated_at=datetime.strptime(issue_data['updated_at'].replace('Z', '').split('.')[0], '%Y-%m-%dT%H:%M:%S'),
html_url=issue_data['html_url'],
assignee=assignee,
labels=labels
)
except (KeyError, ValueError) as e:
raise IssueError(f"Failed to parse issue data: {e}")
return self.gitea_client.issues.list_open()
except GiteaAuthError as e:
raise IssueError("Authentication failed when fetching open issues") from e
except GiteaApiError as e:
raise IssueError(f"API error fetching open issues: {e}") from e
except GiteaError as e:
raise IssueError(f"Gitea error fetching open issues: {e}") from e
def get_issue_data_dict(self, issue_number: int) -> Dict[str, Any]:
"""Get issue data as dictionary for workspace creation."""
@@ -134,6 +87,6 @@ class IssueFetcher:
'created_at': issue.created_at.isoformat(),
'updated_at': issue.updated_at.isoformat(),
'html_url': issue.html_url,
'assignee': {'login': issue.assignee} if issue.assignee else None,
'labels': [{'name': label} for label in issue.labels]
'assignee': {'login': issue.assignee.login} if issue.assignee else None,
'labels': [{'name': label.name} for label in issue.labels]
}

View File

@@ -1,150 +1,107 @@
"""
Project management functionality for Gitea using milestones and labels.
Project management functionality using the Gitea facade.
Since Gitea project boards may not be available in all instances, this module
provides project management using milestones (for projects) and labels (for states).
This module now acts as an adapter to the new gitea package,
maintaining backwards compatibility while using the cleaner API.
"""
import json
import os
import subprocess
from subprocess import PIPE
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum
from gitea import GiteaClient, GiteaConfig
from gitea.models import ProjectState, Priority, Milestone as GiteaMilestone, Label as GiteaLabel
from gitea.exceptions import GiteaError, GiteaNotFoundError, GiteaAuthError, GiteaApiError
from .config import get_config
from .exceptions import IssueError
class ProjectState(Enum):
"""Standard project states using labels."""
TODO = "status:todo"
ACTIVE = "status:active"
REVIEW = "status:review"
DONE = "status:done"
BLOCKED = "status:blocked"
class Priority(Enum):
"""Priority levels using labels."""
LOW = "priority:low"
MEDIUM = "priority:medium"
HIGH = "priority:high"
CRITICAL = "priority:critical"
@dataclass
class Milestone:
"""Represents a project milestone."""
id: int
title: str
description: str
state: str
open_issues: int
closed_issues: int
due_on: Optional[str] = None
@dataclass
class Label:
"""Represents an issue label."""
id: int
name: str
color: str
description: str
# Re-export for backwards compatibility
Milestone = GiteaMilestone
Label = GiteaLabel
class ProjectManager:
"""Manages project organization using milestones and labels."""
"""Manages project organization using the Gitea facade."""
def __init__(self, config=None, auth_token=None):
self.config = config or get_config()
self.auth_token = auth_token or os.getenv('GITEA_API_TOKEN')
# Create Gitea client from tddai config
gitea_config = GiteaConfig.from_tddai_config(self.config)
if self.auth_token:
gitea_config.auth_token = self.auth_token
self.gitea_client = GiteaClient(gitea_config)
def _make_api_call(self, method: str, url: str, data: Dict[str, Any] = None) -> Dict[str, Any]:
"""Make authenticated API call to Gitea."""
if not self.auth_token:
raise IssueError("Authentication token required for project operations")
"""Make authenticated API call to Gitea (kept for backwards compatibility).
cmd = [
'curl', '-s', '-X', method,
'-H', 'Content-Type: application/json',
'-H', f'Authorization: token {self.auth_token}',
]
if data:
cmd.extend(['-d', json.dumps(data)])
cmd.append(url)
Args:
method: HTTP method (GET, POST, etc.)
url: API endpoint URL
data: Optional request data
Raises:
IssueError: When API call fails (with specific context)
"""
# This method is kept for backwards compatibility but now delegates to the gitea client
# For new code, use the gitea_client directly
try:
result = subprocess.run(cmd, stdout=PIPE, stderr=PIPE, universal_newlines=True, check=True)
if result.returncode != 0:
raise IssueError(f"API call failed: {result.stderr}")
if result.stdout.strip():
response_data = json.loads(result.stdout)
# Check for API error responses
if isinstance(response_data, dict) and 'message' in response_data and 'id' not in response_data:
raise IssueError(f"API error: {response_data['message']}")
return response_data
if method == 'GET' and 'issues' in url and url.endswith('/issues'):
issues = self.gitea_client.issues.list()
return [self._issue_to_dict(issue) for issue in issues]
elif method == 'GET' and '/issues/' in url and not url.endswith('/labels'):
issue_number = int(url.split('/issues/')[-1])
issue = self.gitea_client.issues.get(issue_number)
return self._issue_to_dict(issue)
else:
return {}
raise IssueError(f"Legacy API call not supported: {method} {url}")
except GiteaNotFoundError as e:
raise IssueError(f"Resource not found for {method} {url}") from e
except GiteaAuthError as e:
raise IssueError(f"Authentication failed for {method} {url}") from e
except GiteaApiError as e:
raise IssueError(f"API error for {method} {url}: {e}") from e
except GiteaError as e:
raise IssueError(f"Gitea error for {method} {url}: {e}") from e
except (ValueError, IndexError) as e:
raise IssueError(f"Invalid URL format for {method} {url}") from e
except subprocess.CalledProcessError as e:
raise IssueError(f"API call failed: {e}")
except json.JSONDecodeError as e:
raise IssueError(f"Failed to parse API response: {e}")
def _issue_to_dict(self, issue) -> Dict[str, Any]:
"""Convert Issue object to dict for backwards compatibility."""
return {
'number': issue.number,
'title': issue.title,
'body': issue.body,
'state': issue.state,
'html_url': issue.html_url,
'created_at': issue.created_at.isoformat(),
'updated_at': issue.updated_at.isoformat(),
'assignee': {'login': issue.assignee.login} if issue.assignee else None,
'labels': [{'name': label.name, 'color': label.color} for label in issue.labels]
}
# Milestone Management (Projects)
def create_milestone(self, title: str, description: str = "", due_date: str = None) -> Milestone:
"""Create a new milestone (project)."""
url = f"{self.config.gitea_url}/api/v1/repos/{self.config.repo_owner}/{self.config.repo_name}/milestones"
data = {
'title': title,
'description': description,
}
if due_date:
data['due_on'] = due_date
response = self._make_api_call('POST', url, data)
return Milestone(
id=response['id'],
title=response['title'],
description=response.get('description', ''),
state=response['state'],
open_issues=response['open_issues'],
closed_issues=response['closed_issues'],
due_on=response.get('due_on')
)
try:
return self.gitea_client.milestones.create(title, description, due_date)
except Exception as e:
raise IssueError(f"Failed to create milestone: {e}")
def list_milestones(self, state: str = "open") -> List[Milestone]:
"""List all milestones (projects)."""
url = f"{self.config.gitea_url}/api/v1/repos/{self.config.repo_owner}/{self.config.repo_name}/milestones"
params = f"?state={state}" if state else ""
response = self._make_api_call('GET', url + params)
return [
Milestone(
id=m['id'],
title=m['title'],
description=m.get('description', ''),
state=m['state'],
open_issues=m['open_issues'],
closed_issues=m['closed_issues'],
due_on=m.get('due_on')
)
for m in response
]
try:
if state == "all":
return self.gitea_client.milestones.list()
elif state == "open":
return self.gitea_client.milestones.list_open()
elif state == "closed":
return self.gitea_client.milestones.list_closed()
else:
return self.gitea_client.milestones.list(state)
except Exception as e:
raise IssueError(f"Failed to list milestones: {e}")
def update_milestone(self, milestone_id: int, **kwargs) -> Milestone:
"""Update milestone details."""
@@ -209,36 +166,24 @@ class ProjectManager:
def ensure_project_labels(self) -> None:
"""Ensure all required project management labels exist."""
existing_labels = {label.name for label in self.list_labels()}
try:
self.gitea_client.labels.ensure_project_labels()
except Exception as e:
raise IssueError(f"Failed to ensure project labels: {e}")
# Standard state labels
required_labels = [
('status:todo', 'e6e6e6', 'Issues ready to be worked on'),
('status:active', '0052cc', 'Issues currently being worked on'),
('status:review', 'fbca04', 'Issues under review'),
('status:done', '0e8a16', 'Completed issues'),
('status:blocked', 'd93f0b', 'Issues blocked by dependencies'),
def list_labels(self) -> List[Label]:
"""List all repository labels."""
try:
return self.gitea_client.labels.list()
except Exception as e:
raise IssueError(f"Failed to list labels: {e}")
# Priority labels
('priority:low', 'c2e0c6', 'Low priority issue'),
('priority:medium', 'fef2c0', 'Medium priority issue'),
('priority:high', 'f9d0c4', 'High priority issue'),
('priority:critical', 'f4c2c2', 'Critical priority issue'),
# Type labels
('type:bug', 'fc2929', 'Bug report'),
('type:feature', '84b6eb', 'New feature request'),
('type:enhancement', '7057ff', 'Enhancement to existing feature'),
('type:documentation', '0075ca', 'Documentation update'),
]
for name, color, description in required_labels:
if name not in existing_labels:
try:
self.create_label(name, color, description)
print(f"✅ Created label: {name}")
except IssueError as e:
print(f"⚠️ Failed to create label {name}: {e}")
def create_label(self, name: str, color: str, description: str = "") -> Label:
"""Create a new label."""
try:
return self.gitea_client.labels.create(name, color, description)
except Exception as e:
raise IssueError(f"Failed to create label: {e}")
# Project Management Operations
@@ -251,61 +196,31 @@ class ProjectManager:
def set_issue_state(self, issue_number: int, state: ProjectState) -> Dict[str, Any]:
"""Set issue project state using labels."""
# Use the dedicated labels endpoint which works more reliably
labels_url = f"{self.config.issues_api_url}/{issue_number}/labels"
# First get current labels
issue_url = f"{self.config.issues_api_url}/{issue_number}"
issue_data = self._make_api_call('GET', issue_url)
current_labels = [label['name'] for label in issue_data.get('labels', [])]
state_labels = [label for label in current_labels if label.startswith('status:')]
# Remove old state labels
for old_state in state_labels:
if old_state in current_labels:
current_labels.remove(old_state)
# Add new state label
current_labels.append(state.value)
# Use PUT to replace all labels on the dedicated labels endpoint
data = {'labels': current_labels}
return self._make_api_call('PUT', labels_url, data)
try:
issue = self.gitea_client.issues.set_status(issue_number, state)
return self._issue_to_dict(issue)
except Exception as e:
raise IssueError(f"Failed to set issue state: {e}")
def set_issue_priority(self, issue_number: int, priority: Priority) -> Dict[str, Any]:
"""Set issue priority using labels."""
# Use the dedicated labels endpoint which works more reliably
labels_url = f"{self.config.issues_api_url}/{issue_number}/labels"
# First get current labels
issue_url = f"{self.config.issues_api_url}/{issue_number}"
issue_data = self._make_api_call('GET', issue_url)
current_labels = [label['name'] for label in issue_data.get('labels', [])]
priority_labels = [label for label in current_labels if label.startswith('priority:')]
# Remove old priority labels
for old_priority in priority_labels:
if old_priority in current_labels:
current_labels.remove(old_priority)
# Add new priority label
current_labels.append(priority.value)
# Use PUT to replace all labels on the dedicated labels endpoint
data = {'labels': current_labels}
return self._make_api_call('PUT', labels_url, data)
try:
issue = self.gitea_client.issues.set_priority(issue_number, priority)
return self._issue_to_dict(issue)
except Exception as e:
raise IssueError(f"Failed to set issue priority: {e}")
def move_issue_to_done(self, issue_number: int) -> Dict[str, Any]:
"""Move issue to done state and close it."""
# Set state to done
self.set_issue_state(issue_number, ProjectState.DONE)
try:
# Set state to done
self.set_issue_state(issue_number, ProjectState.DONE)
# Close the issue
url = f"{self.config.issues_api_url}/{issue_number}"
data = {'state': 'closed'}
return self._make_api_call('PATCH', url, data)
# Close the issue
issue = self.gitea_client.issues.close(issue_number)
return self._issue_to_dict(issue)
except Exception as e:
raise IssueError(f"Failed to move issue to done: {e}")
def get_project_overview(self) -> Dict[str, Any]:
"""Get overview of project status."""

View File

@@ -1,703 +1,149 @@
#!/usr/bin/env python3
"""
CLI interface for tddai library.
This module now uses the separated architecture with services and presenters.
Business logic is handled by services, presentation by CLI framework.
"""
import sys
import argparse
from pathlib import Path
# Add current directory to path so we can import tddai
# Add current directory to path so we can import modules
sys.path.insert(0, str(Path(__file__).parent))
from tddai import (
WorkspaceManager, IssueFetcher, TestGenerator, CoverageAnalyzer,
WorkspaceStatus, TddaiError
)
from tddai.issue_creator import IssueCreator
from tddai.project_manager import ProjectManager, ProjectState, Priority
from cli import CLIFramework
# Lazy initialization of CLI framework
_cli_framework = None
def _get_cli():
"""Get CLI framework instance (lazy initialization)."""
global _cli_framework
if _cli_framework is None:
_cli_framework = CLIFramework()
return _cli_framework
def workspace_status():
"""Show current workspace status."""
try:
manager = WorkspaceManager()
status = manager.get_status()
if status == WorkspaceStatus.CLEAN:
print("📋 No active issue workspace")
print(" Use 'make tdd-start NUM=X' to begin working on an issue")
return
if status == WorkspaceStatus.DIRTY:
print("⚠️ Workspace directory exists but no current issue file")
print(" Run 'make tdd-finish' to clean up or 'make tdd-start' to create new workspace")
return
workspace = manager.get_current_workspace()
if not workspace:
print("❌ Failed to load workspace")
return
print("📋 Active Issue Workspace")
print("========================")
print()
print(f"🎯 Issue #{workspace.issue_number}: {workspace.issue_title}")
print(f"📊 Status: {workspace.issue_state}")
print(f"📁 Workspace: {workspace.workspace_dir}/issue_{workspace.issue_number}/")
print()
if workspace.tests_dir.exists():
test_files = list(workspace.tests_dir.glob("*.py"))
print(f"🧪 Generated Tests ({len(test_files)}):")
if test_files:
for test_file in test_files:
print(f" - {test_file.name}")
else:
print(" - No tests generated yet")
print()
print("📋 Workspace Files:")
print(" - requirements.md (review and break down issue)")
print(" - test_plan.md (plan test scenarios)")
print(" - tests/ (generated test files)")
print()
print("💡 Commands:")
print(" - make tdd-add-test (generate another test)")
print(" - make tdd-finish (complete and move tests to main)")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().workspace_status()
def start_issue(issue_number: int):
"""Start working on an issue."""
try:
manager = WorkspaceManager()
fetcher = IssueFetcher()
# Check if workspace already active
status = manager.get_status()
if status == WorkspaceStatus.ACTIVE:
current = manager.get_current_workspace()
print(f"⚠️ Already working on issue #{current.issue_number}")
print(" Run 'make tdd-finish' first or 'make tdd-status' to see details")
sys.exit(1)
print(f"🔍 Starting work on issue #{issue_number}...")
print(f"📋 Fetching issue #{issue_number} details...")
# Fetch issue data
issue_data = fetcher.get_issue_data_dict(issue_number)
# Create workspace
workspace = manager.create_workspace(issue_data)
print(f"✅ Workspace created for issue #{issue_number}")
print(f"📁 Workspace: {workspace.workspace_dir}/issue_{issue_number}/")
print(f"📋 Requirements: {workspace.requirements_file}")
print(f"🧪 Test plan: {workspace.test_plan_file}")
print()
print("💡 Next steps:")
print(" 1. Review requirements.md and break down the issue")
print(" 2. Plan test scenarios in test_plan.md")
print(" 3. Use 'make tdd-add-test' to generate tests")
print(" 4. Use 'make tdd-finish' when complete")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().start_issue(issue_number)
def finish_issue():
"""Finish current issue workspace."""
try:
manager = WorkspaceManager()
workspace = manager.get_current_workspace()
if not workspace:
print("❌ No active issue workspace")
print(" Nothing to finish")
sys.exit(1)
print(f"🏁 Finishing work on issue #{workspace.issue_number}")
print()
# Check for tests
if workspace.tests_dir.exists():
test_files = list(workspace.tests_dir.glob("*.py"))
if test_files:
print(f"📦 Moving {len(test_files)} test(s) to tests/ directory...")
print("✅ Tests moved to main tests/ directory")
else:
print("⚠️ No tests found in workspace")
# Finish workspace (moves tests and cleans up)
manager.finish_workspace()
print("🧹 Cleaning up workspace...")
print(f"✅ Issue #{workspace.issue_number} workspace cleaned up")
print()
print("💡 Next steps:")
print(" - Run 'make test' to verify tests fail (red state)")
print(" - Implement code to make tests pass (green state)")
print(" - Start next issue with 'make tdd-start NUM=X'")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().finish_issue()
def add_test_guidance():
"""Show guidance for adding tests."""
try:
manager = WorkspaceManager()
workspace = manager.get_current_workspace()
if not workspace:
print("❌ No active issue workspace")
print(" Run 'make tdd-start NUM=X' first")
sys.exit(1)
print(f"🧪 Adding test to issue #{workspace.issue_number} workspace")
print()
print(f"📋 Issue: {workspace.issue_title}")
print(f"📁 Workspace: {workspace.workspace_dir}/issue_{workspace.issue_number}/")
print()
print("🤖 Please ask Claude Code to generate a test:")
print()
print(" Command: 'Generate a test for the current workspace issue'")
print()
print("📝 Test Requirements:")
print(f" - Save test in: {workspace.tests_dir}/")
print(f" - Name format: test_issue_{workspace.issue_number}_<scenario>.py")
print(f" - Include docstring referencing issue #{workspace.issue_number}")
print(" - Follow TDD principles (test should fail initially)")
print(" - Review requirements.md and test_plan.md for context")
print()
print("📋 Issue Details:")
print(f" Title: {workspace.issue_title}")
print(f" Description: {workspace.issue_body}")
print()
print("💡 After generation: Use 'make tdd-status' to see all tests")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().add_test_guidance()
def list_issues():
"""List all issues."""
try:
fetcher = IssueFetcher()
print("📋 Project Issues")
print("==================")
print()
issues = fetcher.fetch_issues()
if not issues:
print("No issues found")
return
for issue in issues:
status_icon = "🟢" if issue.state == "open" else "🔴"
print(f"{status_icon} #{issue.number}: {issue.title}")
print(f" Status: {issue.state.upper()} | Created: {issue.created_at.strftime('%Y-%m-%d')}")
# Truncate body for list view
body_preview = issue.body[:80] + "..." if len(issue.body) > 80 else issue.body
if body_preview:
print(f" {body_preview}")
print()
print("💡 Tip: Use 'make show-issue NUM=X' for full details")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().list_issues()
def list_open_issues():
"""List only open issues."""
try:
fetcher = IssueFetcher()
print("📋 Open Project Issues (Active Backlog)")
print("========================================")
print()
issues = fetcher.fetch_open_issues()
if not issues:
print("No open issues found")
return
for issue in issues:
print(f"[OPEN] #{issue.number}: {issue.title}")
print(f" Created: {issue.created_at.strftime('%Y-%m-%d')} | Updated: {issue.updated_at.strftime('%Y-%m-%d')}")
# Truncate body for list view
body_preview = issue.body[:80] + "..." if len(issue.body) > 80 else issue.body
if body_preview:
print(f" {body_preview}")
print()
print("💡 Tip: Use 'make show-issue NUM=X' for full details or 'make list-issues' for all issues")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
def analyze_coverage(issue_number: int):
"""Analyze test coverage for a specific issue."""
try:
analyzer = CoverageAnalyzer()
print(f"🔍 Analyzing test coverage for Issue #{issue_number}")
print("=" * 50)
print()
assessment = analyzer.analyze_issue_coverage(issue_number)
print(f"📋 Issue: #{assessment.issue_number} - {assessment.issue_title}")
print(f"📊 Coverage: {assessment.coverage_percentage:.1f}%")
print()
# Show requirements analysis
print("🎯 Identified Requirements:")
if assessment.requirements:
for req in assessment.requirements:
priority_icon = {"critical": "🚨", "important": "⚠️", "nice-to-have": "💡"}
icon = priority_icon.get(req.priority, "📝")
print(f" {icon} [{req.priority.upper()}] {req.category}: {req.description}")
else:
print(" No specific requirements detected")
print()
# Show existing tests
print("🧪 Existing Test Coverage:")
issue_related_tests = [t for t in assessment.existing_tests if t.related_issue == issue_number]
if issue_related_tests:
for test in issue_related_tests:
test_count = len(test.test_methods)
print(f"{test.file_path.name} ({test_count} test methods)")
if test.test_methods:
for method in test.test_methods[:3]: # Show first 3
print(f" - {method}")
if len(test.test_methods) > 3:
print(f" - ... and {len(test.test_methods) - 3} more")
else:
print(" 📝 No tests specifically for this issue found")
# Show general tests that might be relevant
relevant_tests = [t for t in assessment.existing_tests
if any(keyword in ' '.join(t.coverage_keywords)
for req in assessment.requirements
for keyword in req.keywords)]
if relevant_tests:
print(" 📋 Potentially relevant tests:")
for test in relevant_tests[:3]:
print(f" 📄 {test.file_path.name}")
print()
# Show coverage gaps
if assessment.coverage_gaps:
print("❌ Coverage Gaps Found:")
for gap in assessment.coverage_gaps:
priority_icon = {"critical": "🚨", "important": "⚠️", "nice-to-have": "💡"}
icon = priority_icon.get(gap.requirement.priority, "📝")
print(f" {icon} Missing: {gap.requirement.description}")
print(f" 💡 Suggested test: {gap.suggested_test_name}")
print(f" 📄 Suggested file: {gap.suggested_test_file}")
print()
else:
print("✅ No significant coverage gaps detected!")
print()
# Show recommendations
print("📝 Recommendations:")
for recommendation in assessment.recommendations:
print(f" {recommendation}")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
_get_cli().list_open_issues()
def show_issue(issue_number: int):
"""Show detailed issue information with comprehensive project management details."""
try:
fetcher = IssueFetcher()
project_mgr = ProjectManager()
print(f"🔍 Issue #{issue_number} Details")
print("=======================")
print()
# Get basic issue information
issue = fetcher.fetch_issue(issue_number)
print(f"**Title:** {issue.title}")
print(f"**Status:** {issue.state.upper()}")
print(f"**Number:** #{issue.number}")
print(f"**Created:** {issue.created_at.strftime('%Y-%m-%d %H:%M')}")
print(f"**Updated:** {issue.updated_at.strftime('%Y-%m-%d %H:%M')}")
print(f"**URL:** {issue.html_url}")
if issue.assignee:
print(f"**Assignee:** {issue.assignee}")
# Enhanced project management information
print()
print("**Project Management:**")
# Get detailed issue data via API for milestone and project information
from tddai.config import get_config
config = get_config()
issue_url = f"{config.issues_api_url}/{issue_number}"
detailed_issue = project_mgr._make_api_call('GET', issue_url)
# Milestone information
if detailed_issue.get('milestone'):
milestone = detailed_issue['milestone']
print(f" 📋 Milestone: #{milestone['id']} - {milestone['title']} ({milestone['state']})")
else:
print(f" 📋 Milestone: None")
# Project/Board information (if available through API)
# Note: Gitea project boards may use different API endpoints
print(f" 🎯 Project: Getting Started (assumed - requires board API)")
# Labels and state information
labels = detailed_issue.get('labels', [])
if labels:
state_labels = [l['name'] for l in labels if l['name'].startswith('status:')]
priority_labels = [l['name'] for l in labels if l['name'].startswith('priority:')]
type_labels = [l['name'] for l in labels if l['name'].startswith('type:')]
other_labels = [l['name'] for l in labels if not any(l['name'].startswith(p) for p in ['status:', 'priority:', 'type:'])]
if state_labels:
state_display = state_labels[0].replace('status:', '').title()
print(f" 📊 State: {state_display}")
else:
print(f" 📊 State: No state label")
if priority_labels:
priority_display = priority_labels[0].replace('priority:', '').title()
print(f" 🚨 Priority: {priority_display}")
else:
print(f" 🚨 Priority: No priority set")
if type_labels:
type_display = ', '.join([l.replace('type:', '').title() for l in type_labels])
print(f" 🏷️ Type: {type_display}")
if other_labels:
print(f" 🏷️ Other Labels: {', '.join(other_labels)}")
else:
print(f" 📊 State: No state label")
print(f" 🚨 Priority: No priority set")
print(f" 🏷️ Labels: None")
# Column information (based on state and issue status)
if detailed_issue.get('state') == 'closed':
if any(l['name'] == 'status:done' for l in labels):
column = "Done"
else:
column = "Closed"
else:
state_labels = [l['name'] for l in labels if l['name'].startswith('status:')]
if state_labels:
state = state_labels[0].replace('status:', '')
column_map = {
'todo': 'Todo',
'active': 'Active',
'review': 'Review',
'blocked': 'Blocked'
}
column = column_map.get(state, 'Todo')
else:
column = "Todo"
print(f" 📝 Kanban Column: {column}")
print()
print("**Description:**")
print(issue.body)
print()
print("💡 Tip: Use 'make list-issues' to see all issues")
except TddaiError as e:
print(f"❌ Error: {e}")
sys.exit(1)
"""Show detailed issue information."""
_get_cli().show_issue(issue_number)
def create_issue(title: str, body: str, issue_type: str = "enhancement"):
"""Create a new issue."""
try:
creator = IssueCreator()
print(f"🚀 Creating {issue_type} issue: {title}")
print()
if issue_type == "enhancement":
# For enhancements, assume body contains structured content
result = creator.create_issue(title, body, labels=[issue_type])
elif issue_type == "bug":
result = creator.create_issue(title, body, labels=[issue_type])
else:
result = creator.create_issue(title, body)
print("✅ Issue created successfully!")
print(f" Number: #{result['number']}")
print(f" Title: {result['title']}")
print(f" Status: {result['state']}")
if 'html_url' in result:
print(f" URL: {result['html_url']}")
print()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
except TddaiError as e:
print(f"❌ Error creating issue: {e}")
sys.exit(1)
_get_cli().create_issue(title, body, issue_type)
def create_enhancement_issue(title: str, use_case: str, technical_requirements: str = "",
acceptance_criteria: str = "", dependencies: str = "",
priority: str = "Medium"):
"""Create a structured enhancement issue."""
try:
creator = IssueCreator()
print(f"🚀 Creating enhancement issue: {title}")
print()
# Parse acceptance criteria if provided
criteria_list = []
if acceptance_criteria:
criteria_list = [line.strip() for line in acceptance_criteria.split('\n') if line.strip()]
# Parse acceptance criteria if provided
criteria_list = []
if acceptance_criteria:
criteria_list = [line.strip() for line in acceptance_criteria.split('\n') if line.strip()]
# Parse dependencies if provided
deps_list = []
if dependencies:
deps_list = [line.strip() for line in dependencies.split('\n') if line.strip()]
# Parse dependencies if provided
deps_list = []
if dependencies:
deps_list = [line.strip() for line in dependencies.split('\n') if line.strip()]
result = creator.create_enhancement_issue(
title=title,
use_case=use_case,
technical_requirements=technical_requirements,
acceptance_criteria=criteria_list,
dependencies=deps_list,
priority=priority
)
print("✅ Enhancement issue created successfully!")
print(f" Number: #{result['number']}")
print(f" Title: {result['title']}")
print(f" Priority: {priority}")
if 'html_url' in result:
print(f" URL: {result['html_url']}")
print()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
except TddaiError as e:
print(f"❌ Error creating enhancement issue: {e}")
sys.exit(1)
_get_cli().create_enhancement_issue(
title=title,
use_case=use_case,
technical_requirements=technical_requirements,
acceptance_criteria=criteria_list,
dependencies=deps_list,
priority=priority
)
def create_from_template(template_file: str, **kwargs):
"""Create issue from template file."""
try:
creator = IssueCreator()
print(f"🚀 Creating issue from template: {template_file}")
print()
_get_cli().create_from_template(template_file, **kwargs)
result = creator.create_from_template(template_file, **kwargs)
print("✅ Issue created from template successfully!")
print(f" Number: #{result['number']}")
print(f" Title: {result['title']}")
if 'html_url' in result:
print(f" URL: {result['html_url']}")
print()
print("💡 Next steps:")
print(f" - Use 'make tdd-start NUM={result['number']}' to begin work")
print(f" - Use 'make show-issue NUM={result['number']}' to view details")
except TddaiError as e:
print(f"❌ Error creating issue from template: {e}")
sys.exit(1)
def analyze_coverage(issue_number: int):
"""Analyze test coverage for a specific issue."""
_get_cli().analyze_coverage(issue_number)
def setup_project_management():
"""Setup project management labels and milestones."""
try:
project_mgr = ProjectManager()
print("🚀 Setting up project management system...")
# Ensure all required labels exist
project_mgr.ensure_project_labels()
print("✅ Project management setup complete!")
print("📋 Available states: todo, active, review, done, blocked")
print("📊 Available priorities: low, medium, high, critical")
except TddaiError as e:
print(f"❌ Error setting up project management: {e}")
sys.exit(1)
_get_cli().setup_project_management()
def move_issue_to_state(issue_number: int, state: str):
"""Move issue to a specific project state."""
try:
project_mgr = ProjectManager()
# Convert string to ProjectState enum
state_map = {
'todo': ProjectState.TODO,
'active': ProjectState.ACTIVE,
'review': ProjectState.REVIEW,
'done': ProjectState.DONE,
'blocked': ProjectState.BLOCKED
}
if state not in state_map:
print(f"❌ Invalid state '{state}'. Valid states: {list(state_map.keys())}")
sys.exit(1)
project_state = state_map[state]
print(f"📋 Moving issue #{issue_number} to {state} state...")
result = project_mgr.set_issue_state(issue_number, project_state)
# If moving to done, also close the issue
if state == 'done':
project_mgr.move_issue_to_done(issue_number)
print(f"✅ Issue #{issue_number} moved to {state} and closed")
else:
print(f"✅ Issue #{issue_number} moved to {state}")
except TddaiError as e:
print(f"❌ Error moving issue to {state}: {e}")
sys.exit(1)
_get_cli().move_issue_to_state(issue_number, state)
def set_issue_priority(issue_number: int, priority: str):
"""Set issue priority."""
try:
project_mgr = ProjectManager()
# Convert string to Priority enum
priority_map = {
'low': Priority.LOW,
'medium': Priority.MEDIUM,
'high': Priority.HIGH,
'critical': Priority.CRITICAL
}
if priority not in priority_map:
print(f"❌ Invalid priority '{priority}'. Valid priorities: {list(priority_map.keys())}")
sys.exit(1)
priority_level = priority_map[priority]
print(f"📊 Setting issue #{issue_number} priority to {priority}...")
result = project_mgr.set_issue_priority(issue_number, priority_level)
print(f"✅ Issue #{issue_number} priority set to {priority}")
except TddaiError as e:
print(f"❌ Error setting issue priority: {e}")
sys.exit(1)
_get_cli().set_issue_priority(issue_number, priority)
def create_milestone(title: str, description: str = ""):
"""Create a new milestone (project)."""
try:
project_mgr = ProjectManager()
print(f"🚀 Creating milestone: {title}")
milestone = project_mgr.create_milestone(title, description)
print(f"✅ Milestone created successfully!")
print(f" ID: {milestone.id}")
print(f" Title: {milestone.title}")
print(f" Description: {milestone.description}")
print(f" State: {milestone.state}")
except TddaiError as e:
print(f"❌ Error creating milestone: {e}")
sys.exit(1)
_get_cli().create_milestone(title, description)
def list_milestones():
"""List all milestones."""
try:
project_mgr = ProjectManager()
print("📋 Project Milestones")
print("====================")
print()
milestones = project_mgr.list_milestones("all")
if not milestones:
print("No milestones found")
return
for milestone in milestones:
status_icon = "🟢" if milestone.state == "open" else "🔴"
print(f"{status_icon} Milestone #{milestone.id}: {milestone.title}")
print(f" State: {milestone.state.upper()}")
print(f" Issues: {milestone.open_issues} open, {milestone.closed_issues} closed")
if milestone.description:
print(f" Description: {milestone.description}")
if milestone.due_on:
print(f" Due: {milestone.due_on}")
print()
except TddaiError as e:
print(f"❌ Error listing milestones: {e}")
sys.exit(1)
_get_cli().list_milestones()
def assign_issue_to_milestone(issue_number: int, milestone_id: int):
"""Assign issue to a milestone."""
try:
from tddai.issue_writer import IssueWriter
writer = IssueWriter()
print(f"📋 Assigning issue #{issue_number} to milestone #{milestone_id}...")
result = writer.assign_to_milestone(issue_number, milestone_id)
print(f"✅ Issue #{issue_number} assigned to milestone #{milestone_id}")
except TddaiError as e:
print(f"❌ Error assigning issue to milestone: {e}")
sys.exit(1)
_get_cli().assign_issue_to_milestone(issue_number, milestone_id)
def project_overview():
"""Show project management overview."""
try:
project_mgr = ProjectManager()
print("📊 Project Management Overview")
print("==============================")
print()
_get_cli().project_overview()
overview = project_mgr.get_project_overview()
print(f"📋 Milestones: {overview['milestones']} total")
print(f" Active Projects: {overview['active_projects']}")
print(f" Completed Projects: {overview['completed_projects']}")
print(f"🏷️ Total Labels: {overview['total_labels']}")
print(f"🎯 Project Management Ready: {'✅ Yes' if overview['project_management_ready'] else '❌ No - run setup-project-mgmt'}")
except TddaiError as e:
print(f"❌ Error getting project overview: {e}")
sys.exit(1)
def issue_index(format_type="tsv", sort_by="number", filter_state=None, filter_priority=None, include_state=False):
"""Output compact index of all issues for Unix processing."""
_get_cli().issue_index(
format_type=format_type,
sort_by=sort_by,
filter_state=filter_state,
filter_priority=filter_priority,
include_state=include_state
)
def main():
@@ -718,6 +164,18 @@ def main():
subparsers.add_parser('list-issues', help='List all issues')
subparsers.add_parser('list-open-issues', help='List open issues')
index_parser = subparsers.add_parser('issue-index', help='Output compact issue index for Unix processing')
index_parser.add_argument('--format', choices=['tsv', 'csv', 'json', 'fields'], default='tsv',
help='Output format (default: tsv)')
index_parser.add_argument('--sort', choices=['number', 'title', 'priority', 'state', 'created', 'updated'],
default='number', help='Sort by field (default: number)')
index_parser.add_argument('--filter-state', choices=['open', 'closed'],
help='Filter by issue state')
index_parser.add_argument('--filter-priority', choices=['low', 'medium', 'high', 'critical', 'none'],
help='Filter by priority level')
index_parser.add_argument('--include-state', action='store_true',
help='Include state column in output')
show_parser = subparsers.add_parser('show-issue', help='Show issue details')
show_parser.add_argument('issue_number', type=int, help='Issue number')
@@ -783,6 +241,14 @@ def main():
list_issues()
elif args.command == 'list-open-issues':
list_open_issues()
elif args.command == 'issue-index':
issue_index(
format_type=args.format,
sort_by=args.sort,
filter_state=args.filter_state,
filter_priority=args.filter_priority,
include_state=args.include_state
)
elif args.command == 'show-issue':
show_issue(args.issue_number)
elif args.command == 'analyze-coverage':

View File

@@ -108,16 +108,16 @@ class TestWorkspaceCreation:
with pytest.raises(WorkspaceError, match="Workspace already active"):
manager.create_workspace(second_issue_data)
@patch('tddai.issue_fetcher.subprocess.run')
@patch('gitea.http_client.subprocess.run')
def test_issue_fetcher_handles_invalid_issue(self, mock_run, temp_workspace):
"""Test error handling for invalid issue numbers."""
# Mock curl response for non-existent issue
mock_run.return_value.returncode = 0
mock_run.return_value.stdout = '{"message": "404 Not Found"}'
# Mock curl response for non-existent issue (404 error)
from subprocess import CalledProcessError
mock_run.side_effect = CalledProcessError(22, 'curl') # HTTP 404 error
fetcher = IssueFetcher(temp_workspace)
with pytest.raises(IssueError, match="not found"):
with pytest.raises(IssueError, match="API error fetching issue.*HTTP request failed"):
fetcher.fetch_issue(999)
def test_workspace_cleanup(self, temp_workspace, mock_issue_data):

View File

@@ -25,6 +25,18 @@ class TestIssueCreator:
repo_name="test_repo"
)
def _get_complete_mock_response(self, number: int, title: str = "Test Issue", body: str = "Test description"):
"""Get a complete mock API response with all required fields."""
return {
"number": number,
"title": title,
"body": body,
"state": "open",
"created_at": "2025-09-26T10:00:00Z",
"updated_at": "2025-09-26T10:00:00Z",
"html_url": f"http://gitea.example.com/repo/issues/{number}"
}
def test_init_with_auth_token(self):
"""Test IssueCreator initialization with auth token."""
config = self._get_test_config()
@@ -62,7 +74,10 @@ class TestIssueCreator:
"number": 123,
"title": "Test Issue",
"body": "Test description",
"state": "open"
"state": "open",
"created_at": "2025-09-26T10:00:00Z",
"updated_at": "2025-09-26T10:00:00Z",
"html_url": "http://gitea.example.com/repo/issues/123"
}
mock_run.return_value = MagicMock(
@@ -73,7 +88,19 @@ class TestIssueCreator:
result = creator.create_issue("Test Issue", "Test description")
assert result == mock_response
# Verify the result has the expected structure (transformed by issue creator)
expected_result = {
'number': 123,
'title': "Test Issue",
'body': "Test description",
'state': "open",
'html_url': "http://gitea.example.com/repo/issues/123",
'created_at': "2025-09-26T10:00:00", # ISO format from datetime parsing
'updated_at': "2025-09-26T10:00:00",
'assignee': None,
'labels': []
}
assert result == expected_result
mock_run.assert_called_once()
# Check curl command structure
@@ -144,7 +171,7 @@ class TestIssueCreator:
stderr=""
)
with pytest.raises(IssueError, match="Failed to parse response data"):
with pytest.raises(IssueError, match="Failed to create issue.*parse.*response"):
creator.create_issue("Test Issue", "Test description")
@patch('subprocess.run')
@@ -153,7 +180,7 @@ class TestIssueCreator:
config = self._get_test_config()
creator = IssueCreator(config=config, auth_token="test-token")
mock_response = {"number": 124}
mock_response = self._get_complete_mock_response(124)
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps(mock_response),
@@ -183,7 +210,7 @@ class TestIssueCreator:
config = self._get_test_config()
creator = IssueCreator(config=config, auth_token="test-token")
mock_response = {"number": 125}
mock_response = self._get_complete_mock_response(125)
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps(mock_response),
@@ -218,7 +245,7 @@ class TestIssueCreator:
config = self._get_test_config()
creator = IssueCreator(config=config, auth_token="test-token")
mock_response = {"number": 126}
mock_response = self._get_complete_mock_response(126)
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps(mock_response),
@@ -255,7 +282,7 @@ class TestIssueCreator:
config = self._get_test_config()
creator = IssueCreator(config=config, auth_token="test-token")
mock_response = {"number": 127}
mock_response = self._get_complete_mock_response(127)
mock_run.return_value = MagicMock(
returncode=0,
stdout=json.dumps(mock_response),