Getting ready for the lint gods

This commit is contained in:
AndyMik90
2025-12-15 19:49:58 +01:00
parent d8ad17dbfa
commit 142cd67c88
17 changed files with 5033 additions and 166 deletions
+15 -5
View File
@@ -48,6 +48,7 @@ export function App() {
const selectedProjectId = useProjectStore((state) => state.selectedProjectId); const selectedProjectId = useProjectStore((state) => state.selectedProjectId);
const tasks = useTaskStore((state) => state.tasks); const tasks = useTaskStore((state) => state.tasks);
const settings = useSettingsStore((state) => state.settings); const settings = useSettingsStore((state) => state.settings);
const settingsLoading = useSettingsStore((state) => state.isLoading);
// UI State // UI State
const [selectedTask, setSelectedTask] = useState<Task | null>(null); const [selectedTask, setSelectedTask] = useState<Task | null>(null);
@@ -72,14 +73,23 @@ export function App() {
loadSettings(); loadSettings();
}, []); }, []);
// First-run detection - show onboarding wizard if not completed // Track if settings have been loaded at least once
const [settingsHaveLoaded, setSettingsHaveLoaded] = useState(false);
// Mark settings as loaded when loading completes
useEffect(() => { useEffect(() => {
// Only show wizard if onboardingCompleted is explicitly false (not undefined) if (!settingsLoading && !settingsHaveLoaded) {
// This ensures we don't show the wizard before settings are loaded setSettingsHaveLoaded(true);
if (settings.onboardingCompleted === false) { }
}, [settingsLoading, settingsHaveLoaded]);
// First-run detection - show onboarding wizard if not completed
// Only check AFTER settings have been loaded from disk to avoid race condition
useEffect(() => {
if (settingsHaveLoaded && settings.onboardingCompleted === false) {
setIsOnboardingWizardOpen(true); setIsOnboardingWizardOpen(true);
} }
}, [settings.onboardingCompleted]); }, [settingsHaveLoaded, settings.onboardingCompleted]);
// Listen for open-app-settings events (e.g., from project settings) // Listen for open-app-settings events (e.g., from project settings)
useEffect(() => { useEffect(() => {
@@ -81,28 +81,42 @@ export function OnboardingWizard({
} }
}, [currentStepIndex]); }, [currentStepIndex]);
const skipWizard = useCallback(async () => { // Reset wizard state (for re-running) - defined before skipWizard/finishWizard that use it
// Mark onboarding as completed and close - save to disk AND update local state
await window.electronAPI.saveSettings({ onboardingCompleted: true });
updateSettings({ onboardingCompleted: true });
onOpenChange(false);
resetWizard();
}, [updateSettings, onOpenChange]);
const finishWizard = useCallback(async () => {
// Mark onboarding as completed - save to disk AND update local state
await window.electronAPI.saveSettings({ onboardingCompleted: true });
updateSettings({ onboardingCompleted: true });
onOpenChange(false);
resetWizard();
}, [updateSettings, onOpenChange]);
// Reset wizard state (for re-running)
const resetWizard = useCallback(() => { const resetWizard = useCallback(() => {
setCurrentStepIndex(0); setCurrentStepIndex(0);
setCompletedSteps(new Set()); setCompletedSteps(new Set());
}, []); }, []);
const skipWizard = useCallback(async () => {
// Mark onboarding as completed and close - save to disk AND update local state
try {
const result = await window.electronAPI.saveSettings({ onboardingCompleted: true });
if (!result?.success) {
console.error('Failed to save onboarding completion:', result?.error);
}
} catch (err) {
console.error('Error saving onboarding completion:', err);
}
updateSettings({ onboardingCompleted: true });
onOpenChange(false);
resetWizard();
}, [updateSettings, onOpenChange, resetWizard]);
const finishWizard = useCallback(async () => {
// Mark onboarding as completed - save to disk AND update local state
try {
const result = await window.electronAPI.saveSettings({ onboardingCompleted: true });
if (!result?.success) {
console.error('Failed to save onboarding completion:', result?.error);
}
} catch (err) {
console.error('Error saving onboarding completion:', err);
}
updateSettings({ onboardingCompleted: true });
onOpenChange(false);
resetWizard();
}, [updateSettings, onOpenChange, resetWizard]);
// Handle opening task creator from within wizard // Handle opening task creator from within wizard
const handleOpenTaskCreator = useCallback(() => { const handleOpenTaskCreator = useCallback(() => {
if (onOpenTaskCreator) { if (onOpenTaskCreator) {
@@ -16,7 +16,7 @@ interface SettingsState {
export const useSettingsStore = create<SettingsState>((set) => ({ export const useSettingsStore = create<SettingsState>((set) => ({
settings: DEFAULT_APP_SETTINGS as AppSettings, settings: DEFAULT_APP_SETTINGS as AppSettings,
isLoading: false, isLoading: true, // Start as true since we load settings on app init
error: null, error: null,
setSettings: (settings) => set({ settings }), setSettings: (settings) => set({ settings }),
+18 -6
View File
@@ -183,18 +183,30 @@ CLAUDE_CODE_OAUTH_TOKEN=your-oauth-token-here
# - Add discoveries to the graph (add_episode) # - Add discoveries to the graph (add_episode)
# - Retrieve recent sessions (get_episodes) # - Retrieve recent sessions (get_episodes)
# #
# Prerequisites: # Quick Start (uses docker-compose.yml in project root):
# 1. Run the Graphiti MCP server via Docker: # 1. Set OPENAI_API_KEY in your .env file (or export it)
# docker run -d -p 8000:8000 \ # 2. Run: docker-compose up -d
# -e OPENAI_API_KEY=$OPENAI_API_KEY \ # 3. Set GRAPHITI_MCP_URL below
# -e FALKORDB_URI=redis://host.docker.internal:6380 \ #
# falkordb/graphiti-knowledge-graph-mcp # Manual Docker (alternative):
# docker run -d -p 8000:8000 \
# -e DATABASE_TYPE=falkordb \
# -e OPENAI_API_KEY=$OPENAI_API_KEY \
# -e FALKORDB_URI=redis://host.docker.internal:6380 \
# falkordb/graphiti-knowledge-graph-mcp
# #
# See: https://docs.falkordb.com/agentic-memory/graphiti-mcp-server.html # See: https://docs.falkordb.com/agentic-memory/graphiti-mcp-server.html
# Graphiti MCP Server URL (setting this enables MCP integration) # Graphiti MCP Server URL (setting this enables MCP integration)
# GRAPHITI_MCP_URL=http://localhost:8000/mcp/ # GRAPHITI_MCP_URL=http://localhost:8000/mcp/
# Graphiti MCP Server Port (for docker-compose, default: 8000)
# GRAPHITI_MCP_PORT=8000
# Optional: Override model settings for MCP server
# GRAPHITI_MODEL_NAME=gpt-4o-mini
# GRAPHITI_EMBEDDING_MODEL=text-embedding-3-small
# ============================================================================= # =============================================================================
# GRAPHITI: FalkorDB Connection Settings # GRAPHITI: FalkorDB Connection Settings
# ============================================================================= # =============================================================================
+34 -3
View File
@@ -3,12 +3,18 @@
# #
# Services: # Services:
# - falkordb: Graph database for Graphiti memory integration # - falkordb: Graph database for Graphiti memory integration
# - graphiti-mcp: MCP server for agent access to knowledge graph
# #
# Usage: # Usage:
# docker-compose up -d falkordb # Start FalkorDB # docker-compose up -d # Start all services
# docker-compose up -d falkordb # Start FalkorDB only
# docker-compose ps # Check status # docker-compose ps # Check status
# docker-compose logs falkordb # View logs # docker-compose logs -f # View logs
# docker-compose down # Stop services # docker-compose down # Stop services
#
# Note: Set OPENAI_API_KEY in your environment or .env file for the MCP server
name: auto-claude
services: services:
falkordb: falkordb:
@@ -19,7 +25,6 @@ services:
volumes: volumes:
- falkordb_data:/data - falkordb_data:/data
environment: environment:
# FalkorDB configuration
- FALKORDB_ARGS=--appendonly yes - FALKORDB_ARGS=--appendonly yes
healthcheck: healthcheck:
test: ["CMD", "redis-cli", "ping"] test: ["CMD", "redis-cli", "ping"]
@@ -28,6 +33,32 @@ services:
retries: 5 retries: 5
restart: unless-stopped restart: unless-stopped
graphiti-mcp:
image: falkordb/graphiti-knowledge-graph-mcp:latest
container_name: auto-claude-graphiti-mcp
ports:
- "${GRAPHITI_MCP_PORT:-8000}:8000"
environment:
# Required: Database type
- DATABASE_TYPE=falkordb
# FalkorDB connection (uses docker network)
- FALKORDB_URI=redis://falkordb:6379
# OpenAI API key for embeddings/LLM (required)
- OPENAI_API_KEY=${OPENAI_API_KEY}
# Optional: Custom model settings
- MODEL_NAME=${GRAPHITI_MODEL_NAME:-gpt-4o-mini}
- EMBEDDING_MODEL=${GRAPHITI_EMBEDDING_MODEL:-text-embedding-3-small}
depends_on:
falkordb:
condition: service_healthy
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 10s
restart: unless-stopped
volumes: volumes:
falkordb_data: falkordb_data:
driver: local driver: local
+449
View File
@@ -391,3 +391,452 @@ def stage_files(temp_git_repo: Path):
filepath.write_text(content) filepath.write_text(content)
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True) subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
return _stage_files return _stage_files
# =============================================================================
# PHASE TESTING FIXTURES - Mock functions for spec/phases.py testing
# =============================================================================
@pytest.fixture
def mock_run_agent_fn():
"""
Mock agent function for testing PhaseExecutor.
Returns a factory that creates mock agent functions with configurable responses.
Usage:
async def test_something(mock_run_agent_fn):
agent_fn = mock_run_agent_fn(success=True, output="Done")
result = await agent_fn("prompt.md")
assert result == (True, "Done")
"""
def _create_mock(
success: bool = True,
output: str = "Agent completed successfully",
side_effect: list = None,
):
"""Create a mock agent function.
Args:
success: Whether the agent should succeed
output: The output message to return
side_effect: Optional list of (success, output) tuples for sequential calls
"""
call_count = 0
async def _mock_agent(
prompt_file: str,
additional_context: str = None,
) -> tuple[bool, str]:
nonlocal call_count
if side_effect is not None:
if call_count < len(side_effect):
result = side_effect[call_count]
call_count += 1
return result
# Fallback to last result if more calls than expected
return side_effect[-1]
return (success, output)
_mock_agent.call_count = 0
return _mock_agent
return _create_mock
@pytest.fixture
def mock_task_logger():
"""
Mock TaskLogger for testing PhaseExecutor.
Returns a mock object that tracks all log calls without side effects.
Usage:
def test_something(mock_task_logger):
executor = PhaseExecutor(..., task_logger=mock_task_logger, ...)
# After test
assert mock_task_logger.log.call_count > 0
"""
from unittest.mock import MagicMock
logger = MagicMock()
logger.log = MagicMock()
logger.start_phase = MagicMock()
logger.end_phase = MagicMock()
logger.tool_start = MagicMock()
logger.tool_end = MagicMock()
logger.save = MagicMock()
return logger
@pytest.fixture
def mock_ui_module():
"""
Mock UI module for testing PhaseExecutor.
Provides mock implementations of UI functions used by PhaseExecutor.
Usage:
def test_something(mock_ui_module):
executor = PhaseExecutor(..., ui_module=mock_ui_module, ...)
# UI calls are captured
assert mock_ui_module.print_status.called
"""
from unittest.mock import MagicMock
ui = MagicMock()
ui.print_status = MagicMock()
ui.muted = MagicMock(return_value="")
ui.bold = MagicMock(return_value="")
ui.success = MagicMock(return_value="")
ui.error = MagicMock(return_value="")
ui.warning = MagicMock(return_value="")
ui.info = MagicMock(return_value="")
ui.highlight = MagicMock(return_value="")
return ui
@pytest.fixture
def mock_spec_validator():
"""
Mock spec validator for testing PhaseExecutor.
Returns a mock validator with configurable validation results.
Usage:
def test_something(mock_spec_validator):
validator = mock_spec_validator(spec_valid=True, plan_valid=True)
result = validator.validate_spec_document()
assert result.valid
"""
from unittest.mock import MagicMock
from dataclasses import dataclass
@dataclass
class MockValidationResult:
valid: bool
checkpoint: str = "test"
errors: list = None
fixes: list = None
def __post_init__(self):
if self.errors is None:
self.errors = []
if self.fixes is None:
self.fixes = []
def _create_mock(
spec_valid: bool = True,
plan_valid: bool = True,
context_valid: bool = True,
all_valid: bool = None,
):
validator = MagicMock()
# validate_spec_document
spec_result = MockValidationResult(
valid=spec_valid,
checkpoint="spec_document",
errors=[] if spec_valid else ["Spec validation failed"],
)
validator.validate_spec_document = MagicMock(return_value=spec_result)
# validate_implementation_plan
plan_result = MockValidationResult(
valid=plan_valid,
checkpoint="implementation_plan",
errors=[] if plan_valid else ["Plan validation failed"],
)
validator.validate_implementation_plan = MagicMock(return_value=plan_result)
# validate_context
context_result = MockValidationResult(
valid=context_valid,
checkpoint="context",
errors=[] if context_valid else ["Context validation failed"],
)
validator.validate_context = MagicMock(return_value=context_result)
# validate_all returns list of all results
if all_valid is None:
all_valid = spec_valid and plan_valid and context_valid
all_results = [spec_result, plan_result, context_result]
if not all_valid:
# Add at least one failing result
if spec_valid and plan_valid and context_valid:
all_results[0] = MockValidationResult(
valid=False,
checkpoint="spec_document",
errors=["Override: all_valid=False"],
)
validator.validate_all = MagicMock(return_value=all_results)
return validator
return _create_mock
# =============================================================================
# SAMPLE DATA FIXTURES - Sample JSON data for phase testing
# =============================================================================
@pytest.fixture
def sample_requirements_json() -> dict:
"""
Sample requirements.json data for testing.
Returns a dict that can be written to requirements.json in test specs.
"""
return {
"task_description": "Add user authentication using OAuth2 with Google provider",
"workflow_type": "feature",
"services_involved": ["backend", "frontend"],
"user_requirements": [
"Users should be able to sign in with Google",
"Session should persist across page refreshes",
"Logout should clear all session data",
],
"acceptance_criteria": [
"POST /api/auth/google endpoint accepts OAuth token",
"Frontend shows Google sign-in button",
"User profile displays after successful login",
],
"constraints": [
"Must use existing user table schema",
"No third-party auth libraries except google-auth",
],
"out_of_scope": [
"Other OAuth providers",
"Password-based authentication",
],
}
@pytest.fixture
def sample_complexity_assessment() -> dict:
"""
Sample complexity_assessment.json data for testing.
Returns a dict representing an AI-assessed complexity for a standard task.
"""
return {
"complexity": "standard",
"confidence": 0.85,
"reasoning": "2 services involved, OAuth integration requires research",
"signals": {
"simple_keywords": 0,
"complex_keywords": 2,
"multi_service_keywords": 2,
"external_integrations": 1,
"infrastructure_changes": False,
"estimated_files": 6,
"estimated_services": 2,
"explicit_services": 2,
},
"estimated_files": 6,
"estimated_services": 2,
"external_integrations": ["oauth", "google"],
"infrastructure_changes": False,
"phases_to_run": [
"discovery",
"historical_context",
"requirements",
"research",
"context",
"spec_writing",
"planning",
"validation",
],
"needs_research": True,
"needs_self_critique": False,
"dev_mode": False,
"created_at": "2024-01-15T10:30:00",
}
@pytest.fixture
def sample_context_json() -> dict:
"""
Sample context.json data for testing.
Returns a dict representing discovered file context for a task.
"""
return {
"task_description": "Add user authentication using OAuth2",
"services_involved": ["backend", "frontend"],
"files_to_modify": [
{
"path": "backend/app/routes/auth.py",
"reason": "Add OAuth endpoints",
"service": "backend",
},
{
"path": "frontend/src/components/Login.tsx",
"reason": "Add Google sign-in button",
"service": "frontend",
},
],
"files_to_create": [
{
"path": "backend/app/services/oauth.py",
"reason": "OAuth service implementation",
"service": "backend",
},
],
"files_to_reference": [
{
"path": "backend/app/models/user.py",
"reason": "Existing user model schema",
"service": "backend",
},
{
"path": "backend/app/config.py",
"reason": "Configuration patterns",
"service": "backend",
},
],
"created_at": "2024-01-15T10:35:00",
}
@pytest.fixture
def sample_project_index() -> dict:
"""
Sample project_index.json data for testing.
Returns a dict representing discovered project structure.
"""
return {
"project_type": "monorepo",
"services": {
"backend": {
"path": "backend",
"language": "python",
"framework": "fastapi",
"package_manager": "pip",
},
"frontend": {
"path": "frontend",
"language": "typescript",
"framework": "next",
"package_manager": "npm",
},
},
"file_count": 150,
"top_level_dirs": ["backend", "frontend", "docs", ".github"],
"config_files": ["pyproject.toml", "package.json", "docker-compose.yml"],
"has_tests": True,
"has_ci": True,
"created_at": "2024-01-15T10:25:00",
}
@pytest.fixture
def sample_graph_hints() -> dict:
"""
Sample graph_hints.json data for testing historical context phase.
Returns a dict representing Graphiti knowledge graph hints.
"""
return {
"enabled": True,
"query": "Add user authentication using OAuth2",
"hints": [
{
"type": "session_insight",
"content": "Previous OAuth implementation used refresh tokens stored in HTTP-only cookies",
"relevance": 0.92,
},
{
"type": "gotcha",
"content": "Google OAuth requires verified domain for production",
"relevance": 0.88,
},
{
"type": "pattern",
"content": "Auth routes follow /api/auth/{provider} convention",
"relevance": 0.85,
},
],
"hint_count": 3,
"created_at": "2024-01-15T10:28:00",
}
@pytest.fixture
def sample_research_json() -> dict:
"""
Sample research.json data for testing research phase.
Returns a dict representing external research findings.
"""
return {
"integrations_researched": [
{
"name": "google-auth",
"package": "google-auth>=2.0.0",
"documentation_url": "https://google-auth.readthedocs.io/",
"findings": [
"Use google.oauth2.id_token for token verification",
"Requires GOOGLE_CLIENT_ID environment variable",
],
"gotchas": [
"Token verification requires network call to Google",
],
},
],
"api_patterns": {
"oauth_flow": "Authorization code flow with PKCE recommended",
"token_storage": "Store refresh token server-side, access token in memory",
},
"security_considerations": [
"Validate token audience matches client ID",
"Use state parameter to prevent CSRF",
],
"created_at": "2024-01-15T10:40:00",
}
@pytest.fixture
def populated_spec_dir(
spec_dir: Path,
sample_requirements_json: dict,
sample_complexity_assessment: dict,
sample_context_json: dict,
sample_project_index: dict,
) -> Path:
"""
Create a fully populated spec directory with all required files.
Useful for testing phases that depend on earlier phase outputs.
"""
# Write all JSON files
(spec_dir / "requirements.json").write_text(json.dumps(sample_requirements_json, indent=2))
(spec_dir / "complexity_assessment.json").write_text(json.dumps(sample_complexity_assessment, indent=2))
(spec_dir / "context.json").write_text(json.dumps(sample_context_json, indent=2))
(spec_dir / "project_index.json").write_text(json.dumps(sample_project_index, indent=2))
# Write sample spec.md
spec_content = """# User Authentication with OAuth2
## Overview
Add Google OAuth2 authentication to the application.
## Requirements
1. Users can sign in with Google
2. Sessions persist across page refreshes
3. Logout clears all session data
## Implementation Notes
- Use google-auth library for token verification
- Store refresh tokens server-side
## Acceptance Criteria
- [ ] POST /api/auth/google endpoint works
- [ ] Frontend shows Google sign-in button
- [ ] User profile displays after login
"""
(spec_dir / "spec.md").write_text(spec_content)
return spec_dir
+3
View File
@@ -4,8 +4,11 @@ python_files = test_*.py
python_classes = Test* python_classes = Test*
python_functions = test_* python_functions = test_*
addopts = -v --tb=short addopts = -v --tb=short
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
markers = markers =
slow: marks tests as slow (deselect with '-m "not slow"') slow: marks tests as slow (deselect with '-m "not slow"')
integration: marks tests as integration tests integration: marks tests as integration tests
asyncio: marks tests as async tests
filterwarnings = filterwarnings =
ignore::DeprecationWarning ignore::DeprecationWarning
+3 -3
View File
@@ -215,7 +215,7 @@ def test_complete_workflow():
# 6. Format summary # 6. Format summary
summary = format_critique_summary(result) summary = format_critique_summary(result)
assert "PASSED ✓" in summary assert "PASSED ✓" in summary
assert "Chunk is ready to be marked complete" in summary assert "Subtask is ready to be marked complete" in summary
# 7. Store in chunk # 7. Store in chunk
chunk_obj = Chunk( chunk_obj = Chunk(
@@ -247,7 +247,7 @@ def test_summary_formatting():
summary = format_critique_summary(result) summary = format_critique_summary(result)
assert "PASSED ✓" in summary assert "PASSED ✓" in summary
assert "Fixed error handling" in summary assert "Fixed error handling" in summary
assert "Chunk is ready to be marked complete" in summary assert "Subtask is ready to be marked complete" in summary
# Test failed critique summary # Test failed critique summary
result_fail = CritiqueResult( result_fail = CritiqueResult(
@@ -260,7 +260,7 @@ def test_summary_formatting():
summary_fail = format_critique_summary(result_fail) summary_fail = format_critique_summary(result_fail)
assert "FAILED ✗" in summary_fail assert "FAILED ✗" in summary_fail
assert "Missing validation" in summary_fail assert "Missing validation" in summary_fail
assert "Chunk needs more work" in summary_fail assert "Subtask needs more work" in summary_fail
print("✓ Summary formatting works correctly") print("✓ Summary formatting works correctly")
+952
View File
@@ -0,0 +1,952 @@
#!/usr/bin/env python3
"""
Tests for QA Criteria Module
============================
Tests the qa/criteria.py module functionality including:
- Implementation plan I/O
- QA signoff status management
- QA readiness checks (should_run_qa, should_run_fixes)
- Status display functions
Note: This test module mocks all dependencies to avoid importing
the Claude SDK which is not available in the test environment.
"""
import json
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import MagicMock
import pytest
# =============================================================================
# MOCK SETUP - Must happen before ANY imports from auto-claude
# =============================================================================
# Mock claude_agent_sdk FIRST (before any other imports)
mock_sdk = MagicMock()
mock_sdk.ClaudeSDKClient = MagicMock()
mock_sdk.ClaudeAgentOptions = MagicMock()
mock_sdk.ClaudeCodeOptions = MagicMock()
sys.modules['claude_agent_sdk'] = mock_sdk
# Mock UI module (used by progress)
mock_ui = MagicMock()
mock_ui.Icons = MagicMock()
mock_ui.icon = MagicMock(return_value="")
mock_ui.color = MagicMock()
mock_ui.Color = MagicMock()
mock_ui.success = MagicMock(return_value="")
mock_ui.error = MagicMock(return_value="")
mock_ui.warning = MagicMock(return_value="")
mock_ui.info = MagicMock(return_value="")
mock_ui.muted = MagicMock(return_value="")
mock_ui.highlight = MagicMock(return_value="")
mock_ui.bold = MagicMock(return_value="")
mock_ui.box = MagicMock(return_value="")
mock_ui.divider = MagicMock(return_value="")
mock_ui.progress_bar = MagicMock(return_value="")
mock_ui.print_header = MagicMock()
mock_ui.print_section = MagicMock()
mock_ui.print_status = MagicMock()
mock_ui.print_phase_status = MagicMock()
mock_ui.print_key_value = MagicMock()
sys.modules['ui'] = mock_ui
# Mock progress module
mock_progress = MagicMock()
mock_progress.count_subtasks = MagicMock(return_value=(3, 3))
mock_progress.is_build_complete = MagicMock(return_value=True)
sys.modules['progress'] = mock_progress
# Mock task_logger
mock_task_logger = MagicMock()
mock_task_logger.LogPhase = MagicMock()
mock_task_logger.LogEntryType = MagicMock()
mock_task_logger.get_task_logger = MagicMock(return_value=None)
sys.modules['task_logger'] = mock_task_logger
# Mock linear_updater
mock_linear = MagicMock()
mock_linear.is_linear_enabled = MagicMock(return_value=False)
mock_linear.LinearTaskState = MagicMock()
mock_linear.linear_qa_started = MagicMock()
mock_linear.linear_qa_approved = MagicMock()
mock_linear.linear_qa_rejected = MagicMock()
mock_linear.linear_qa_max_iterations = MagicMock()
sys.modules['linear_updater'] = mock_linear
# Mock client module
mock_client = MagicMock()
mock_client.create_client = MagicMock()
sys.modules['client'] = mock_client
# Now we can safely add the auto-claude path and import
sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude"))
# Import criteria functions directly to avoid going through qa/__init__.py
# which imports reviewer and fixer that need the SDK
from qa.criteria import (
load_implementation_plan,
save_implementation_plan,
get_qa_signoff_status,
is_qa_approved,
is_qa_rejected,
is_fixes_applied,
get_qa_iteration_count,
should_run_qa,
should_run_fixes,
print_qa_status,
)
# Mock the qa.report import inside print_qa_status
mock_report = MagicMock()
mock_report.get_iteration_history = MagicMock(return_value=[])
mock_report.get_recurring_issue_summary = MagicMock(return_value={})
# =============================================================================
# FIXTURES
# =============================================================================
@pytest.fixture
def temp_dir():
"""Create a temporary directory for tests."""
with tempfile.TemporaryDirectory() as tmpdir:
yield Path(tmpdir)
@pytest.fixture
def spec_dir(temp_dir):
"""Create a spec directory with basic structure."""
spec = temp_dir / "spec"
spec.mkdir()
return spec
@pytest.fixture
def qa_signoff_approved():
"""Return an approved QA signoff structure."""
return {
"status": "approved",
"qa_session": 1,
"timestamp": "2024-01-01T12:00:00",
"tests_passed": {
"unit": True,
"integration": True,
"e2e": True,
},
}
@pytest.fixture
def qa_signoff_rejected():
"""Return a rejected QA signoff structure."""
return {
"status": "rejected",
"qa_session": 1,
"timestamp": "2024-01-01T12:00:00",
"issues_found": [
{"title": "Test failure", "type": "unit_test"},
{"title": "Missing validation", "type": "acceptance"},
],
}
@pytest.fixture
def sample_implementation_plan():
"""Return a sample implementation plan structure."""
return {
"feature": "User Avatar Upload",
"workflow_type": "feature",
"services_involved": ["backend", "worker", "frontend"],
"phases": [
{
"phase": 1,
"name": "Backend Foundation",
"subtasks": [
{"id": "subtask-1-1", "description": "Add avatar fields", "status": "completed"},
],
},
],
}
class TestImplementationPlanIO:
"""Tests for implementation plan loading/saving."""
def test_load_implementation_plan(self, spec_dir: Path, sample_implementation_plan: dict):
"""Loads implementation plan from JSON."""
plan_file = spec_dir / "implementation_plan.json"
plan_file.write_text(json.dumps(sample_implementation_plan))
plan = load_implementation_plan(spec_dir)
assert plan is not None
assert plan["feature"] == "User Avatar Upload"
def test_load_missing_plan_returns_none(self, spec_dir: Path):
"""Returns None when plan file doesn't exist."""
plan = load_implementation_plan(spec_dir)
assert plan is None
def test_load_invalid_json_returns_none(self, spec_dir: Path):
"""Returns None for invalid JSON."""
plan_file = spec_dir / "implementation_plan.json"
plan_file.write_text("{ invalid json }")
plan = load_implementation_plan(spec_dir)
assert plan is None
def test_load_empty_file_returns_none(self, spec_dir: Path):
"""Returns None for empty file."""
plan_file = spec_dir / "implementation_plan.json"
plan_file.write_text("")
plan = load_implementation_plan(spec_dir)
assert plan is None
def test_save_implementation_plan(self, spec_dir: Path):
"""Saves implementation plan to JSON."""
plan = {"feature": "Test", "phases": []}
result = save_implementation_plan(spec_dir, plan)
assert result is True
assert (spec_dir / "implementation_plan.json").exists()
loaded = json.loads((spec_dir / "implementation_plan.json").read_text())
assert loaded["feature"] == "Test"
def test_save_implementation_plan_creates_file(self, spec_dir: Path):
"""Creates the file if it doesn't exist."""
plan = {"feature": "New Feature", "phases": []}
result = save_implementation_plan(spec_dir, plan)
assert result is True
assert (spec_dir / "implementation_plan.json").exists()
def test_save_implementation_plan_overwrites(self, spec_dir: Path):
"""Overwrites existing plan file."""
plan_file = spec_dir / "implementation_plan.json"
plan_file.write_text('{"feature": "Old"}')
new_plan = {"feature": "New", "phases": []}
save_implementation_plan(spec_dir, new_plan)
loaded = json.loads(plan_file.read_text())
assert loaded["feature"] == "New"
def test_save_implementation_plan_with_indentation(self, spec_dir: Path):
"""Saves with proper JSON indentation."""
plan = {"feature": "Test", "phases": [{"name": "Phase 1"}]}
save_implementation_plan(spec_dir, plan)
content = (spec_dir / "implementation_plan.json").read_text()
# Check for indentation (2 spaces as per json.dump with indent=2)
assert " " in content
class TestGetQASignoffStatus:
"""Tests for get_qa_signoff_status function."""
def test_get_qa_signoff_status(self, spec_dir: Path):
"""Gets QA signoff status from plan."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "approved",
"qa_session": 1,
"timestamp": "2024-01-01T12:00:00",
},
}
save_implementation_plan(spec_dir, plan)
status = get_qa_signoff_status(spec_dir)
assert status is not None
assert status["status"] == "approved"
def test_get_qa_signoff_status_none(self, spec_dir: Path):
"""Returns None when no signoff status."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
status = get_qa_signoff_status(spec_dir)
assert status is None
def test_get_qa_signoff_status_missing_plan(self, spec_dir: Path):
"""Returns None when plan doesn't exist."""
status = get_qa_signoff_status(spec_dir)
assert status is None
def test_get_qa_signoff_status_empty_signoff(self, spec_dir: Path):
"""Returns empty dict when qa_signoff is empty."""
plan = {"feature": "Test", "qa_signoff": {}}
save_implementation_plan(spec_dir, plan)
status = get_qa_signoff_status(spec_dir)
assert status == {}
class TestIsQAApproved:
"""Tests for is_qa_approved function."""
def test_is_qa_approved_true(self, spec_dir: Path, qa_signoff_approved: dict):
"""is_qa_approved returns True when approved."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is True
def test_is_qa_approved_false_when_rejected(self, spec_dir: Path, qa_signoff_rejected: dict):
"""is_qa_approved returns False when rejected."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is False
def test_is_qa_approved_no_signoff(self, spec_dir: Path):
"""is_qa_approved returns False when no signoff."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is False
def test_is_qa_approved_no_plan(self, spec_dir: Path):
"""is_qa_approved returns False when no plan exists."""
assert is_qa_approved(spec_dir) is False
def test_is_qa_approved_other_status(self, spec_dir: Path):
"""is_qa_approved returns False for other status values."""
plan = {
"feature": "Test",
"qa_signoff": {"status": "in_progress"},
}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is False
class TestIsQARejected:
"""Tests for is_qa_rejected function."""
def test_is_qa_rejected_true(self, spec_dir: Path, qa_signoff_rejected: dict):
"""is_qa_rejected returns True when rejected."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is True
def test_is_qa_rejected_false_when_approved(self, spec_dir: Path, qa_signoff_approved: dict):
"""is_qa_rejected returns False when approved."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is False
def test_is_qa_rejected_no_signoff(self, spec_dir: Path):
"""is_qa_rejected returns False when no signoff."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is False
def test_is_qa_rejected_no_plan(self, spec_dir: Path):
"""is_qa_rejected returns False when no plan exists."""
assert is_qa_rejected(spec_dir) is False
def test_is_qa_rejected_fixes_applied(self, spec_dir: Path):
"""is_qa_rejected returns False when status is fixes_applied."""
plan = {
"feature": "Test",
"qa_signoff": {"status": "fixes_applied"},
}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is False
class TestIsFixesApplied:
"""Tests for is_fixes_applied function."""
def test_is_fixes_applied_true(self, spec_dir: Path):
"""is_fixes_applied returns True when status is fixes_applied and ready."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "fixes_applied",
"ready_for_qa_revalidation": True,
},
}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is True
def test_is_fixes_applied_not_ready(self, spec_dir: Path):
"""is_fixes_applied returns False when not ready for revalidation."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "fixes_applied",
"ready_for_qa_revalidation": False,
},
}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is False
def test_is_fixes_applied_missing_ready_flag(self, spec_dir: Path):
"""is_fixes_applied returns False when ready flag is missing."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "fixes_applied",
},
}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is False
def test_is_fixes_applied_wrong_status(self, spec_dir: Path):
"""is_fixes_applied returns False when status is not fixes_applied."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"ready_for_qa_revalidation": True,
},
}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is False
def test_is_fixes_applied_no_signoff(self, spec_dir: Path):
"""is_fixes_applied returns False when no signoff."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is False
class TestGetQAIterationCount:
"""Tests for get_qa_iteration_count function."""
def test_get_qa_iteration_count(self, spec_dir: Path):
"""Gets QA iteration count from signoff."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 3,
},
}
save_implementation_plan(spec_dir, plan)
count = get_qa_iteration_count(spec_dir)
assert count == 3
def test_get_qa_iteration_count_zero(self, spec_dir: Path):
"""Returns 0 when no QA sessions."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
count = get_qa_iteration_count(spec_dir)
assert count == 0
def test_get_qa_iteration_count_no_plan(self, spec_dir: Path):
"""Returns 0 when no plan exists."""
count = get_qa_iteration_count(spec_dir)
assert count == 0
def test_get_qa_iteration_count_missing_session(self, spec_dir: Path):
"""Returns 0 when qa_session is missing from signoff."""
plan = {
"feature": "Test",
"qa_signoff": {"status": "rejected"},
}
save_implementation_plan(spec_dir, plan)
count = get_qa_iteration_count(spec_dir)
assert count == 0
def test_get_qa_iteration_count_high_value(self, spec_dir: Path):
"""Handles high iteration count."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 25,
},
}
save_implementation_plan(spec_dir, plan)
count = get_qa_iteration_count(spec_dir)
assert count == 25
class TestShouldRunQA:
"""Tests for should_run_qa function."""
def test_should_run_qa_build_not_complete(self, spec_dir: Path):
"""Returns False when build not complete."""
# Set up mock to return build not complete
mock_progress.is_build_complete.return_value = False
plan = {"feature": "Test", "phases": []}
save_implementation_plan(spec_dir, plan)
result = should_run_qa(spec_dir)
assert result is False
# Reset mock
mock_progress.is_build_complete.return_value = True
def test_should_run_qa_already_approved(self, spec_dir: Path, qa_signoff_approved: dict):
"""Returns False when already approved."""
mock_progress.is_build_complete.return_value = True
plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
save_implementation_plan(spec_dir, plan)
result = should_run_qa(spec_dir)
assert result is False
def test_should_run_qa_build_complete_not_approved(self, spec_dir: Path):
"""Returns True when build complete but not approved."""
mock_progress.is_build_complete.return_value = True
plan = {"feature": "Test", "phases": []}
save_implementation_plan(spec_dir, plan)
result = should_run_qa(spec_dir)
assert result is True
def test_should_run_qa_rejected_status(self, spec_dir: Path, qa_signoff_rejected: dict):
"""Returns True when rejected (needs re-review after fixes)."""
mock_progress.is_build_complete.return_value = True
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
result = should_run_qa(spec_dir)
assert result is True
def test_should_run_qa_no_plan(self, spec_dir: Path):
"""Returns False when no plan exists (build not complete)."""
mock_progress.is_build_complete.return_value = False
result = should_run_qa(spec_dir)
assert result is False
# Reset mock
mock_progress.is_build_complete.return_value = True
class TestShouldRunFixes:
"""Tests for should_run_fixes function."""
def test_should_run_fixes_when_rejected(self, spec_dir: Path, qa_signoff_rejected: dict):
"""Returns True when QA rejected and under max iterations."""
# Ensure qa_session is below MAX_QA_ITERATIONS
qa_signoff_rejected["qa_session"] = 1
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is True
def test_should_run_fixes_max_iterations(self, spec_dir: Path):
"""Returns False when max iterations reached."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 50, # MAX_QA_ITERATIONS
},
}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is False
def test_should_run_fixes_over_max_iterations(self, spec_dir: Path):
"""Returns False when over max iterations."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 100,
},
}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is False
def test_should_run_fixes_not_rejected(self, spec_dir: Path, qa_signoff_approved: dict):
"""Returns False when not rejected."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is False
def test_should_run_fixes_no_signoff(self, spec_dir: Path):
"""Returns False when no signoff exists."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is False
def test_should_run_fixes_fixes_applied_status(self, spec_dir: Path):
"""Returns False when status is fixes_applied (not rejected)."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "fixes_applied",
"qa_session": 1,
},
}
save_implementation_plan(spec_dir, plan)
result = should_run_fixes(spec_dir)
assert result is False
class TestPrintQAStatus:
"""Tests for print_qa_status function."""
def test_print_qa_status_not_started(self, spec_dir: Path, capsys):
"""Prints 'Not started' when no signoff exists."""
plan = {"feature": "Test"}
save_implementation_plan(spec_dir, plan)
# Mock the report module functions
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Not started" in captured.out
def test_print_qa_status_approved(self, spec_dir: Path, qa_signoff_approved: dict, capsys):
"""Prints approved status with test results."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_approved}
save_implementation_plan(spec_dir, plan)
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "APPROVED" in captured.out
assert "Tests:" in captured.out
def test_print_qa_status_rejected(self, spec_dir: Path, qa_signoff_rejected: dict, capsys):
"""Prints rejected status with issues found."""
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "REJECTED" in captured.out
assert "Issues Found:" in captured.out
def test_print_qa_status_with_history(self, spec_dir: Path, qa_signoff_rejected: dict, capsys):
"""Prints iteration history summary when available."""
from unittest.mock import patch
plan = {"feature": "Test", "qa_signoff": qa_signoff_rejected}
save_implementation_plan(spec_dir, plan)
# Mock iteration history using patch for the actual import location
import qa.report as report_module
with patch.object(report_module, 'get_iteration_history', return_value=[
{"iteration": 1, "status": "rejected", "issues": []},
{"iteration": 2, "status": "rejected", "issues": []},
]), patch.object(report_module, 'get_recurring_issue_summary', return_value={
"iterations_approved": 0,
"iterations_rejected": 2,
"most_common": [],
}):
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Iteration History:" in captured.out
assert "Total iterations:" in captured.out
def test_print_qa_status_missing_plan(self, spec_dir: Path, capsys):
"""Prints 'Not started' when plan doesn't exist."""
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Not started" in captured.out
def test_print_qa_status_shows_qa_sessions(self, spec_dir: Path, capsys):
"""Prints QA session count."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 5,
"timestamp": "2024-01-01T12:00:00",
},
}
save_implementation_plan(spec_dir, plan)
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "QA Sessions: 5" in captured.out
def test_print_qa_status_shows_timestamp(self, spec_dir: Path, capsys):
"""Prints last updated timestamp."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "approved",
"qa_session": 1,
"timestamp": "2024-01-15T10:30:00",
},
}
save_implementation_plan(spec_dir, plan)
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Last Updated:" in captured.out
def test_print_qa_status_truncates_issues(self, spec_dir: Path, capsys):
"""Shows only first 3 issues and indicates more."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 1,
"issues_found": [
{"title": "Issue 1", "type": "unit_test"},
{"title": "Issue 2", "type": "unit_test"},
{"title": "Issue 3", "type": "unit_test"},
{"title": "Issue 4", "type": "unit_test"},
{"title": "Issue 5", "type": "unit_test"},
],
},
}
save_implementation_plan(spec_dir, plan)
mock_report.get_iteration_history.return_value = []
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Issue 1" in captured.out
assert "Issue 2" in captured.out
assert "Issue 3" in captured.out
assert "and 2 more" in captured.out
def test_print_qa_status_with_most_common_issues(self, spec_dir: Path, capsys):
"""Prints most common issues from history."""
from unittest.mock import patch
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 3,
},
}
save_implementation_plan(spec_dir, plan)
# Mock iteration history using patch for the actual import location
import qa.report as report_module
with patch.object(report_module, 'get_iteration_history', return_value=[
{"iteration": 1, "status": "rejected"},
{"iteration": 2, "status": "rejected"},
{"iteration": 3, "status": "rejected"},
]), patch.object(report_module, 'get_recurring_issue_summary', return_value={
"iterations_approved": 0,
"iterations_rejected": 3,
"most_common": [
{"title": "Common Issue", "occurrences": 3},
],
}):
print_qa_status(spec_dir)
captured = capsys.readouterr()
assert "Most common issues:" in captured.out
assert "Common Issue" in captured.out
class TestQAStateMachine:
"""Tests for QA state transitions."""
def test_pending_to_rejected(self, spec_dir: Path):
"""Can transition from no signoff to rejected."""
# Start with no signoff
plan = {"feature": "Test", "phases": []}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is False
assert is_qa_rejected(spec_dir) is False
# Transition to rejected
plan["qa_signoff"] = {"status": "rejected", "qa_session": 1}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is True
def test_rejected_to_fixes_applied(self, spec_dir: Path):
"""Can transition from rejected to fixes_applied."""
plan = {
"feature": "Test",
"qa_signoff": {"status": "rejected", "qa_session": 1},
}
save_implementation_plan(spec_dir, plan)
assert is_qa_rejected(spec_dir) is True
# Transition to fixes_applied
plan["qa_signoff"] = {
"status": "fixes_applied",
"ready_for_qa_revalidation": True,
"qa_session": 1,
}
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is True
assert is_qa_rejected(spec_dir) is False
def test_fixes_applied_to_approved(self, spec_dir: Path):
"""Can transition from fixes_applied to approved."""
plan = {
"feature": "Test",
"qa_signoff": {
"status": "fixes_applied",
"ready_for_qa_revalidation": True,
},
}
save_implementation_plan(spec_dir, plan)
# Transition to approved
plan["qa_signoff"] = {"status": "approved", "qa_session": 2}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is True
assert is_fixes_applied(spec_dir) is False
def test_iteration_count_increments(self, spec_dir: Path):
"""QA session counter increments through iterations."""
plan = {"feature": "Test", "qa_signoff": {"status": "rejected", "qa_session": 1}}
save_implementation_plan(spec_dir, plan)
assert get_qa_iteration_count(spec_dir) == 1
plan["qa_signoff"]["qa_session"] = 2
save_implementation_plan(spec_dir, plan)
assert get_qa_iteration_count(spec_dir) == 2
plan["qa_signoff"]["qa_session"] = 3
save_implementation_plan(spec_dir, plan)
assert get_qa_iteration_count(spec_dir) == 3
class TestQAIntegration:
"""Integration tests for QA criteria logic."""
def test_full_qa_workflow_approved_first_try(self, spec_dir: Path):
"""Full workflow where QA approves on first try."""
mock_progress.is_build_complete.return_value = True
# Build complete
plan = {"feature": "Test Feature", "phases": []}
save_implementation_plan(spec_dir, plan)
# Should run QA
assert should_run_qa(spec_dir) is True
# QA approves
plan["qa_signoff"] = {
"status": "approved",
"qa_session": 1,
"tests_passed": {"unit": True, "integration": True, "e2e": True},
}
save_implementation_plan(spec_dir, plan)
# Should not run QA again or fixes
assert should_run_qa(spec_dir) is False
assert should_run_fixes(spec_dir) is False
assert is_qa_approved(spec_dir) is True
def test_full_qa_workflow_with_fixes(self, spec_dir: Path):
"""Full workflow with reject-fix-approve cycle."""
mock_progress.is_build_complete.return_value = True
# Build complete
plan = {"feature": "Test Feature", "phases": []}
save_implementation_plan(spec_dir, plan)
# Should run QA
assert should_run_qa(spec_dir) is True
# QA rejects
plan["qa_signoff"] = {
"status": "rejected",
"qa_session": 1,
"issues_found": [{"title": "Missing test", "type": "unit_test"}],
}
save_implementation_plan(spec_dir, plan)
assert should_run_fixes(spec_dir) is True
assert is_qa_rejected(spec_dir) is True
# Fixes applied
plan["qa_signoff"]["status"] = "fixes_applied"
plan["qa_signoff"]["ready_for_qa_revalidation"] = True
save_implementation_plan(spec_dir, plan)
assert is_fixes_applied(spec_dir) is True
# QA approves on second attempt
plan["qa_signoff"] = {
"status": "approved",
"qa_session": 2,
"tests_passed": {"unit": True, "integration": True, "e2e": True},
}
save_implementation_plan(spec_dir, plan)
assert is_qa_approved(spec_dir) is True
assert get_qa_iteration_count(spec_dir) == 2
def test_qa_workflow_max_iterations(self, spec_dir: Path):
"""Test behavior when max iterations are reached."""
mock_progress.is_build_complete.return_value = True
plan = {
"feature": "Test",
"qa_signoff": {
"status": "rejected",
"qa_session": 50,
},
}
save_implementation_plan(spec_dir, plan)
# Should not run more fixes after max iterations
assert should_run_fixes(spec_dir) is False
# But QA can still be run (to re-check)
assert should_run_qa(spec_dir) is True
File diff suppressed because it is too large Load Diff
+40 -40
View File
@@ -73,8 +73,8 @@ def test_initialization():
# Verify initial structure # Verify initial structure
with open(spec_dir / "memory" / "attempt_history.json") as f: with open(spec_dir / "memory" / "attempt_history.json") as f:
history = json.load(f) history = json.load(f)
assert "chunks" in history, "chunks key missing" assert "subtasks" in history, "subtasks key missing"
assert "stuck_chunks" in history, "stuck_chunks key missing" assert "stuck_subtasks" in history, "stuck_subtasks key missing"
assert "metadata" in history, "metadata key missing" assert "metadata" in history, "metadata key missing"
print(" ✓ Initialization successful") print(" ✓ Initialization successful")
@@ -95,7 +95,7 @@ def test_record_attempt():
# Record failed attempt # Record failed attempt
manager.record_attempt( manager.record_attempt(
chunk_id="chunk-1", subtask_id="subtask-1",
session=1, session=1,
success=False, success=False,
approach="First approach using async/await", approach="First approach using async/await",
@@ -103,25 +103,25 @@ def test_record_attempt():
) )
# Verify recorded # Verify recorded
assert manager.get_attempt_count("chunk-1") == 1, "Attempt not recorded" assert manager.get_attempt_count("subtask-1") == 1, "Attempt not recorded"
history = manager.get_chunk_history("chunk-1") history = manager.get_subtask_history("subtask-1")
assert len(history["attempts"]) == 1, "Wrong number of attempts" assert len(history["attempts"]) == 1, "Wrong number of attempts"
assert history["attempts"][0]["success"] is False, "Success flag wrong" assert history["attempts"][0]["success"] is False, "Success flag wrong"
assert history["status"] == "failed", "Status not updated" assert history["status"] == "failed", "Status not updated"
# Record successful attempt # Record successful attempt
manager.record_attempt( manager.record_attempt(
chunk_id="chunk-1", subtask_id="subtask-1",
session=2, session=2,
success=True, success=True,
approach="Second approach using callbacks", approach="Second approach using callbacks",
error=None error=None
) )
assert manager.get_attempt_count("chunk-1") == 2, "Second attempt not recorded" assert manager.get_attempt_count("subtask-1") == 2, "Second attempt not recorded"
history = manager.get_chunk_history("chunk-1") history = manager.get_subtask_history("subtask-1")
assert len(history["attempts"]) == 2, "Wrong number of attempts" assert len(history["attempts"]) == 2, "Wrong number of attempts"
assert history["attempts"][1]["success"] is True, "Success flag wrong" assert history["attempts"][1]["success"] is True, "Success flag wrong"
assert history["status"] == "completed", "Status not updated to completed" assert history["status"] == "completed", "Status not updated to completed"
@@ -143,18 +143,18 @@ def test_circular_fix_detection():
manager = RecoveryManager(spec_dir, project_dir) manager = RecoveryManager(spec_dir, project_dir)
# Record similar attempts # Record similar attempts
manager.record_attempt("chunk-1", 1, False, "Using async await pattern", "Error 1") manager.record_attempt("subtask-1", 1, False, "Using async await pattern", "Error 1")
manager.record_attempt("chunk-1", 2, False, "Using async await with different import", "Error 2") manager.record_attempt("subtask-1", 2, False, "Using async await with different import", "Error 2")
manager.record_attempt("chunk-1", 3, False, "Trying async await again", "Error 3") manager.record_attempt("subtask-1", 3, False, "Trying async await again", "Error 3")
# Check if circular fix is detected # Check if circular fix is detected
is_circular = manager.is_circular_fix("chunk-1", "Using async await pattern once more") is_circular = manager.is_circular_fix("subtask-1", "Using async await pattern once more")
assert is_circular, "Circular fix not detected" assert is_circular, "Circular fix not detected"
print(" ✓ Circular fix detected correctly") print(" ✓ Circular fix detected correctly")
# Test with different approach # Test with different approach
is_circular = manager.is_circular_fix("chunk-1", "Using completely different callback-based approach") is_circular = manager.is_circular_fix("subtask-1", "Using completely different callback-based approach")
# This might be detected as circular if word overlap is high # This might be detected as circular if word overlap is high
# But "callback-based" is sufficiently different from "async await" # But "callback-based" is sufficiently different from "async await"
@@ -175,17 +175,17 @@ def test_failure_classification():
manager = RecoveryManager(spec_dir, project_dir) manager = RecoveryManager(spec_dir, project_dir)
# Test broken build detection # Test broken build detection
failure = manager.classify_failure("SyntaxError: unexpected token", "chunk-1") failure = manager.classify_failure("SyntaxError: unexpected token", "subtask-1")
assert failure == FailureType.BROKEN_BUILD, "Broken build not detected" assert failure == FailureType.BROKEN_BUILD, "Broken build not detected"
print(" ✓ Broken build classified correctly") print(" ✓ Broken build classified correctly")
# Test verification failed detection # Test verification failed detection
failure = manager.classify_failure("Verification failed: expected 200 got 500", "chunk-2") failure = manager.classify_failure("Verification failed: expected 200 got 500", "subtask-2")
assert failure == FailureType.VERIFICATION_FAILED, "Verification failure not detected" assert failure == FailureType.VERIFICATION_FAILED, "Verification failure not detected"
print(" ✓ Verification failure classified correctly") print(" ✓ Verification failure classified correctly")
# Test context exhaustion # Test context exhaustion
failure = manager.classify_failure("Context length exceeded", "chunk-3") failure = manager.classify_failure("Context length exceeded", "subtask-3")
assert failure == FailureType.CONTEXT_EXHAUSTED, "Context exhaustion not detected" assert failure == FailureType.CONTEXT_EXHAUSTED, "Context exhaustion not detected"
print(" ✓ Context exhaustion classified correctly") print(" ✓ Context exhaustion classified correctly")
@@ -205,27 +205,27 @@ def test_recovery_action_determination():
manager = RecoveryManager(spec_dir, project_dir) manager = RecoveryManager(spec_dir, project_dir)
# Test verification failed with < 3 attempts # Test verification failed with < 3 attempts
manager.record_attempt("chunk-1", 1, False, "First try", "Error") manager.record_attempt("subtask-1", 1, False, "First try", "Error")
action = manager.determine_recovery_action(FailureType.VERIFICATION_FAILED, "chunk-1") action = manager.determine_recovery_action(FailureType.VERIFICATION_FAILED, "subtask-1")
assert action.action == "retry", "Should retry for first verification failure" assert action.action == "retry", "Should retry for first verification failure"
print(" ✓ Retry action for first failure") print(" ✓ Retry action for first failure")
# Test verification failed with >= 3 attempts # Test verification failed with >= 3 attempts
manager.record_attempt("chunk-1", 2, False, "Second try", "Error") manager.record_attempt("subtask-1", 2, False, "Second try", "Error")
manager.record_attempt("chunk-1", 3, False, "Third try", "Error") manager.record_attempt("subtask-1", 3, False, "Third try", "Error")
action = manager.determine_recovery_action(FailureType.VERIFICATION_FAILED, "chunk-1") action = manager.determine_recovery_action(FailureType.VERIFICATION_FAILED, "subtask-1")
assert action.action == "skip", "Should skip after 3 attempts" assert action.action == "skip", "Should skip after 3 attempts"
print(" ✓ Skip action after 3 attempts") print(" ✓ Skip action after 3 attempts")
# Test circular fix # Test circular fix
action = manager.determine_recovery_action(FailureType.CIRCULAR_FIX, "chunk-1") action = manager.determine_recovery_action(FailureType.CIRCULAR_FIX, "subtask-1")
assert action.action == "skip", "Should skip for circular fix" assert action.action == "skip", "Should skip for circular fix"
print(" ✓ Skip action for circular fix") print(" ✓ Skip action for circular fix")
# Test context exhausted # Test context exhausted
action = manager.determine_recovery_action(FailureType.CONTEXT_EXHAUSTED, "chunk-2") action = manager.determine_recovery_action(FailureType.CONTEXT_EXHAUSTED, "subtask-2")
assert action.action == "continue", "Should continue for context exhaustion" assert action.action == "continue", "Should continue for context exhaustion"
print(" ✓ Continue action for context exhaustion") print(" ✓ Continue action for context exhaustion")
@@ -255,7 +255,7 @@ def test_good_commit_tracking():
commit_hash = result.stdout.strip() commit_hash = result.stdout.strip()
# Record good commit # Record good commit
manager.record_good_commit(commit_hash, "chunk-1") manager.record_good_commit(commit_hash, "subtask-1")
# Verify recorded # Verify recorded
last_good = manager.get_last_good_commit() last_good = manager.get_last_good_commit()
@@ -276,7 +276,7 @@ def test_good_commit_tracking():
) )
commit_hash2 = result.stdout.strip() commit_hash2 = result.stdout.strip()
manager.record_good_commit(commit_hash2, "chunk-2") manager.record_good_commit(commit_hash2, "subtask-2")
# Last good should be updated # Last good should be updated
last_good = manager.get_last_good_commit() last_good = manager.get_last_good_commit()
@@ -288,7 +288,7 @@ def test_good_commit_tracking():
cleanup_test_environment(temp_dir) cleanup_test_environment(temp_dir)
def test_mark_chunk_stuck(): def test_mark_subtask_stuck():
"""Test marking chunks as stuck.""" """Test marking chunks as stuck."""
print("TEST: Mark Chunk Stuck") print("TEST: Mark Chunk Stuck")
@@ -298,21 +298,21 @@ def test_mark_chunk_stuck():
manager = RecoveryManager(spec_dir, project_dir) manager = RecoveryManager(spec_dir, project_dir)
# Record some attempts # Record some attempts
manager.record_attempt("chunk-1", 1, False, "Try 1", "Error 1") manager.record_attempt("subtask-1", 1, False, "Try 1", "Error 1")
manager.record_attempt("chunk-1", 2, False, "Try 2", "Error 2") manager.record_attempt("subtask-1", 2, False, "Try 2", "Error 2")
manager.record_attempt("chunk-1", 3, False, "Try 3", "Error 3") manager.record_attempt("subtask-1", 3, False, "Try 3", "Error 3")
# Mark as stuck # Mark as stuck
manager.mark_chunk_stuck("chunk-1", "Circular fix after 3 attempts") manager.mark_subtask_stuck("subtask-1", "Circular fix after 3 attempts")
# Verify stuck # Verify stuck
stuck_chunks = manager.get_stuck_chunks() stuck_subtasks = manager.get_stuck_subtasks()
assert len(stuck_chunks) == 1, "Stuck chunk not recorded" assert len(stuck_subtasks) == 1, "Stuck subtask not recorded"
assert stuck_chunks[0]["chunk_id"] == "chunk-1", "Wrong chunk marked as stuck" assert stuck_subtasks[0]["subtask_id"] == "subtask-1", "Wrong subtask marked as stuck"
assert "Circular fix" in stuck_chunks[0]["reason"], "Reason not recorded" assert "Circular fix" in stuck_subtasks[0]["reason"], "Reason not recorded"
# Check chunk status # Check subtask status
history = manager.get_chunk_history("chunk-1") history = manager.get_subtask_history("subtask-1")
assert history["status"] == "stuck", "Chunk status not updated to stuck" assert history["status"] == "stuck", "Chunk status not updated to stuck"
print(" ✓ Chunk marked as stuck correctly") print(" ✓ Chunk marked as stuck correctly")
@@ -332,11 +332,11 @@ def test_recovery_hints():
manager = RecoveryManager(spec_dir, project_dir) manager = RecoveryManager(spec_dir, project_dir)
# Record some attempts # Record some attempts
manager.record_attempt("chunk-1", 1, False, "Async/await approach", "Import error") manager.record_attempt("subtask-1", 1, False, "Async/await approach", "Import error")
manager.record_attempt("chunk-1", 2, False, "Threading approach", "Thread safety error") manager.record_attempt("subtask-1", 2, False, "Threading approach", "Thread safety error")
# Get hints # Get hints
hints = manager.get_recovery_hints("chunk-1") hints = manager.get_recovery_hints("subtask-1")
assert len(hints) > 0, "No hints generated" assert len(hints) > 0, "No hints generated"
assert "Previous attempts: 2" in hints[0], "Attempt count not in hints" assert "Previous attempts: 2" in hints[0], "Attempt count not in hints"
@@ -368,7 +368,7 @@ def run_all_tests():
test_failure_classification, test_failure_classification,
test_recovery_action_determination, test_recovery_action_determination,
test_good_commit_tracking, test_good_commit_tracking,
test_mark_chunk_stuck, test_mark_subtask_stuck,
test_recovery_hints, test_recovery_hints,
] ]
+21 -22
View File
@@ -23,13 +23,12 @@ from review import (
ReviewState, ReviewState,
ReviewChoice, ReviewChoice,
REVIEW_STATE_FILE, REVIEW_STATE_FILE,
_compute_file_hash,
_compute_spec_hash,
_extract_section,
_truncate_text,
get_review_status_summary, get_review_status_summary,
get_review_menu_options, get_review_menu_options,
extract_section,
truncate_text,
) )
from review.state import _compute_file_hash, _compute_spec_hash
# ============================================================================= # =============================================================================
@@ -590,8 +589,8 @@ class TestReviewStateFeedback:
class TestHelperFunctions: class TestHelperFunctions:
"""Tests for helper functions.""" """Tests for helper functions."""
def test_extract_section_found(self): def testextract_section_found(self):
"""_extract_section() extracts content correctly.""" """extract_section() extracts content correctly."""
content = """# Title content = """# Title
## Overview ## Overview
@@ -602,25 +601,25 @@ This is the overview section.
This is the details section. This is the details section.
""" """
overview = _extract_section(content, "## Overview") overview = extract_section(content, "## Overview")
assert "This is the overview section." in overview assert "This is the overview section." in overview
assert "This is the details section." not in overview assert "This is the details section." not in overview
def test_extract_section_not_found(self): def testextract_section_not_found(self):
"""_extract_section() returns empty string when not found.""" """extract_section() returns empty string when not found."""
content = """# Title content = """# Title
## Existing Section ## Existing Section
Content here. Content here.
""" """
result = _extract_section(content, "## Missing Section") result = extract_section(content, "## Missing Section")
assert result == "" assert result == ""
def test_extract_section_last_section(self): def testextract_section_last_section(self):
"""_extract_section() handles last section correctly.""" """extract_section() handles last section correctly."""
content = """# Title content = """# Title
## First ## First
@@ -631,23 +630,23 @@ First content.
Last content. Last content.
""" """
last = _extract_section(content, "## Last") last = extract_section(content, "## Last")
assert "Last content." in last assert "Last content." in last
def test_truncate_text_short(self): def testtruncate_text_short(self):
"""_truncate_text() returns short text unchanged.""" """truncate_text() returns short text unchanged."""
short_text = "Short text" short_text = "Short text"
result = _truncate_text(short_text, max_lines=10, max_chars=100) result = truncate_text(short_text, max_lines=10, max_chars=100)
assert result == "Short text" assert result == "Short text"
def test_truncate_text_too_many_lines(self): def testtruncate_text_too_many_lines(self):
"""_truncate_text() truncates by line count.""" """truncate_text() truncates by line count."""
long_text = "\n".join(f"Line {i}" for i in range(20)) long_text = "\n".join(f"Line {i}" for i in range(20))
result = _truncate_text(long_text, max_lines=5, max_chars=1000) result = truncate_text(long_text, max_lines=5, max_chars=1000)
# Should contain 5 lines from original + "..." on new line # Should contain 5 lines from original + "..." on new line
lines = result.split("\n") lines = result.split("\n")
@@ -656,11 +655,11 @@ Last content.
assert "Line 0" in result assert "Line 0" in result
assert "Line 4" in result assert "Line 4" in result
def test_truncate_text_too_many_chars(self): def testtruncate_text_too_many_chars(self):
"""_truncate_text() truncates by character count.""" """truncate_text() truncates by character count."""
long_text = "A" * 500 long_text = "A" * 500
result = _truncate_text(long_text, max_lines=100, max_chars=100) result = truncate_text(long_text, max_lines=100, max_chars=100)
assert len(result) <= 100 assert len(result) <= 100
assert result.endswith("...") assert result.endswith("...")
+776
View File
@@ -0,0 +1,776 @@
#!/usr/bin/env python3
"""
Tests for Complexity Assessment Module
======================================
Tests the auto-claude/spec/complexity.py module functionality including:
- Complexity enum values
- ComplexityAssessment dataclass
- ComplexityAnalyzer class methods
- Heuristic-based complexity detection
- Phase selection based on complexity
"""
import json
import pytest
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch, AsyncMock
# Mock claude_agent_sdk and related modules before importing spec modules
# The SDK isn't available in the test environment
mock_code_sdk = MagicMock()
mock_code_sdk.ClaudeSDKClient = MagicMock()
mock_code_sdk.ClaudeCodeOptions = MagicMock()
mock_code_types = MagicMock()
mock_code_types.HookMatcher = MagicMock()
mock_agent_sdk = MagicMock()
mock_agent_sdk.ClaudeAgentOptions = MagicMock()
mock_agent_sdk.ClaudeSDKClient = MagicMock()
mock_agent_types = MagicMock()
mock_agent_types.HookMatcher = MagicMock()
sys.modules['claude_code_sdk'] = mock_code_sdk
sys.modules['claude_code_sdk.types'] = mock_code_types
sys.modules['claude_agent_sdk'] = mock_agent_sdk
sys.modules['claude_agent_sdk.types'] = mock_agent_types
# Add auto-claude directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude"))
from spec.complexity import (
Complexity,
ComplexityAssessment,
ComplexityAnalyzer,
save_assessment,
run_ai_complexity_assessment,
)
class TestComplexityEnum:
"""Tests for Complexity enum values."""
def test_complexity_simple_value(self):
"""SIMPLE enum has correct value."""
assert Complexity.SIMPLE.value == "simple"
def test_complexity_standard_value(self):
"""STANDARD enum has correct value."""
assert Complexity.STANDARD.value == "standard"
def test_complexity_complex_value(self):
"""COMPLEX enum has correct value."""
assert Complexity.COMPLEX.value == "complex"
def test_complexity_from_string(self):
"""Can create Complexity from string value."""
assert Complexity("simple") == Complexity.SIMPLE
assert Complexity("standard") == Complexity.STANDARD
assert Complexity("complex") == Complexity.COMPLEX
def test_complexity_invalid_value_raises(self):
"""Invalid string raises ValueError."""
with pytest.raises(ValueError):
Complexity("invalid")
class TestComplexityAssessmentDataclass:
"""Tests for ComplexityAssessment dataclass."""
def test_default_values(self):
"""Dataclass has sensible defaults."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.8,
)
assert assessment.signals == {}
assert assessment.reasoning == ""
assert assessment.estimated_files == 1
assert assessment.estimated_services == 1
assert assessment.external_integrations == []
assert assessment.infrastructure_changes is False
assert assessment.recommended_phases == []
assert assessment.needs_research is False
assert assessment.needs_self_critique is False
def test_custom_values(self):
"""Can set custom values."""
assessment = ComplexityAssessment(
complexity=Complexity.COMPLEX,
confidence=0.95,
signals={"complex_keywords": 5},
reasoning="High complexity due to integrations",
estimated_files=15,
estimated_services=3,
external_integrations=["redis", "postgres"],
infrastructure_changes=True,
needs_research=True,
needs_self_critique=True,
)
assert assessment.complexity == Complexity.COMPLEX
assert assessment.confidence == 0.95
assert assessment.signals == {"complex_keywords": 5}
assert assessment.estimated_files == 15
assert assessment.infrastructure_changes is True
class TestPhasesToRun:
"""Tests for ComplexityAssessment.phases_to_run()."""
def test_simple_phases(self):
"""SIMPLE complexity returns minimal phases."""
assessment = ComplexityAssessment(
complexity=Complexity.SIMPLE,
confidence=0.9,
)
phases = assessment.phases_to_run()
assert phases == ["discovery", "historical_context", "quick_spec", "validation"]
def test_standard_phases_without_research(self):
"""STANDARD complexity without research flag."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.8,
needs_research=False,
)
phases = assessment.phases_to_run()
assert phases == [
"discovery", "historical_context", "requirements",
"context", "spec_writing", "planning", "validation"
]
def test_standard_phases_with_research(self):
"""STANDARD complexity with research flag includes research phase."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.8,
needs_research=True,
)
phases = assessment.phases_to_run()
assert "research" in phases
assert phases == [
"discovery", "historical_context", "requirements", "research",
"context", "spec_writing", "planning", "validation"
]
def test_complex_phases(self):
"""COMPLEX complexity returns full phase list."""
assessment = ComplexityAssessment(
complexity=Complexity.COMPLEX,
confidence=0.85,
)
phases = assessment.phases_to_run()
assert phases == [
"discovery", "historical_context", "requirements", "research",
"context", "spec_writing", "self_critique", "planning", "validation"
]
def test_recommended_phases_override(self):
"""AI-recommended phases override default phase sets."""
custom_phases = ["discovery", "custom_phase", "validation"]
assessment = ComplexityAssessment(
complexity=Complexity.COMPLEX,
confidence=0.9,
recommended_phases=custom_phases,
)
phases = assessment.phases_to_run()
assert phases == custom_phases
class TestComplexityAnalyzerInit:
"""Tests for ComplexityAnalyzer initialization."""
def test_default_init(self):
"""Initializes with empty project_index."""
analyzer = ComplexityAnalyzer()
assert analyzer.project_index == {}
def test_init_with_project_index(self):
"""Initializes with provided project_index."""
project_index = {"project_type": "monorepo", "services": {"backend": {}}}
analyzer = ComplexityAnalyzer(project_index=project_index)
assert analyzer.project_index == project_index
class TestDetectIntegrations:
"""Tests for ComplexityAnalyzer._detect_integrations()."""
def test_detects_graphiti(self):
"""Detects Graphiti integration."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("integrate with graphiti for memory")
assert "graphiti" in result
def test_detects_database_integrations(self):
"""Detects database integrations."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("migrate postgres database with redis cache")
assert "postgres" in result
assert "redis" in result
def test_detects_cloud_providers(self):
"""Detects cloud provider integrations."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("deploy to aws s3 and lambda")
assert "aws" in result or "s3" in result or "lambda" in result
def test_detects_auth_integrations(self):
"""Detects authentication integrations."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("add oauth authentication with jwt tokens")
assert "oauth" in result or "jwt" in result
def test_detects_queue_integrations(self):
"""Detects message queue integrations."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("process messages with kafka and rabbitmq")
assert "kafka" in result
assert "rabbitmq" in result
def test_returns_empty_for_no_integrations(self):
"""Returns empty list when no integrations detected."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("fix typo in button label")
assert result == []
def test_returns_unique_integrations(self):
"""Returns deduplicated list of integrations."""
analyzer = ComplexityAnalyzer()
result = analyzer._detect_integrations("redis cache with redis queue")
# Should only have redis once
assert result.count("redis") == 1 or "redis" in result
class TestDetectInfrastructureChanges:
"""Tests for ComplexityAnalyzer._detect_infrastructure_changes()."""
def test_detects_docker(self):
"""Detects Docker infrastructure."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("add docker container") is True
def test_detects_kubernetes(self):
"""Detects Kubernetes infrastructure."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("deploy to kubernetes cluster") is True
assert analyzer._detect_infrastructure_changes("configure k8s deployment") is True
def test_detects_deployment(self):
"""Detects deployment changes."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("deploy to production") is True
def test_detects_ci_cd(self):
"""Detects CI/CD changes."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("update ci/cd pipeline") is True
def test_detects_environment_config(self):
"""Detects environment configuration."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("add environment variable") is True
assert analyzer._detect_infrastructure_changes("update config file") is True
def test_detects_schema_changes(self):
"""Detects database schema changes."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("modify database schema") is True
def test_returns_false_for_no_infra(self):
"""Returns False when no infrastructure changes detected."""
analyzer = ComplexityAnalyzer()
assert analyzer._detect_infrastructure_changes("fix typo in button") is False
class TestEstimateFiles:
"""Tests for ComplexityAnalyzer._estimate_files()."""
def test_single_file_keywords(self):
"""Detects single file scope."""
analyzer = ComplexityAnalyzer()
assert analyzer._estimate_files("fix this file only", None) == 1
assert analyzer._estimate_files("update one component", None) == 1
def test_explicit_file_extensions(self):
"""Counts explicit file mentions."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_files("modify app.tsx and utils.py", None)
assert result >= 2
def test_simple_keywords_low_estimate(self):
"""Simple keywords result in low file estimate."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_files("fix typo", None)
assert result <= 3
def test_feature_keywords_medium_estimate(self):
"""Feature keywords result in medium file estimate."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_files("add new feature for users", None)
assert result >= 3
def test_complex_keywords_high_estimate(self):
"""Complex keywords result in high file estimate."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_files("integrate with kafka microservice", None)
assert result >= 10
def test_default_estimate(self):
"""Returns default estimate for generic tasks."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_files("do something", None)
assert result == 5
class TestEstimateServices:
"""Tests for ComplexityAnalyzer._estimate_services()."""
def test_multi_service_keywords(self):
"""Detects multiple services from keywords."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_services("backend api and frontend client", None)
assert result >= 2
def test_monorepo_service_detection(self):
"""Detects mentioned services from monorepo project_index."""
project_index = {
"project_type": "monorepo",
"services": {"backend": {}, "frontend": {}, "worker": {}},
}
analyzer = ComplexityAnalyzer(project_index=project_index)
result = analyzer._estimate_services("update backend and frontend", None)
assert result >= 2
def test_minimum_one_service(self):
"""Returns at least 1 service."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_services("fix typo", None)
assert result >= 1
def test_maximum_five_services(self):
"""Caps at 5 services."""
analyzer = ComplexityAnalyzer()
result = analyzer._estimate_services(
"backend frontend worker service api client server database queue cache proxy",
None
)
assert result <= 5
class TestCalculateComplexity:
"""Tests for ComplexityAnalyzer._calculate_complexity()."""
def test_simple_complexity(self):
"""Calculates SIMPLE complexity correctly."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 2,
"complex_keywords": 0,
"multi_service_keywords": 0,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=[],
infra_changes=False,
estimated_files=1,
estimated_services=1,
)
assert complexity == Complexity.SIMPLE
assert confidence >= 0.8
def test_complex_many_integrations(self):
"""Many integrations results in COMPLEX."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 0,
"complex_keywords": 2,
"multi_service_keywords": 1,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=["redis", "postgres"],
infra_changes=False,
estimated_files=5,
estimated_services=2,
)
assert complexity == Complexity.COMPLEX
def test_complex_infrastructure_changes(self):
"""Infrastructure changes results in COMPLEX."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 0,
"complex_keywords": 1,
"multi_service_keywords": 0,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=[],
infra_changes=True,
estimated_files=3,
estimated_services=1,
)
assert complexity == Complexity.COMPLEX
assert "infrastructure" in reasoning.lower()
def test_complex_many_services(self):
"""Many services results in COMPLEX."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 0,
"complex_keywords": 1,
"multi_service_keywords": 3,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=[],
infra_changes=False,
estimated_files=5,
estimated_services=3,
)
assert complexity == Complexity.COMPLEX
def test_complex_many_files(self):
"""Many files results in COMPLEX."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 0,
"complex_keywords": 2,
"multi_service_keywords": 0,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=[],
infra_changes=False,
estimated_files=15,
estimated_services=1,
)
assert complexity == Complexity.COMPLEX
def test_standard_default(self):
"""Falls back to STANDARD for moderate complexity."""
analyzer = ComplexityAnalyzer()
signals = {
"simple_keywords": 1,
"complex_keywords": 1,
"multi_service_keywords": 1,
}
complexity, confidence, reasoning = analyzer._calculate_complexity(
signals=signals,
integrations=["redis"],
infra_changes=False,
estimated_files=5,
estimated_services=2,
)
assert complexity == Complexity.STANDARD
class TestAnalyze:
"""Tests for ComplexityAnalyzer.analyze() method."""
def test_simple_task_analysis(self):
"""Analyzes a simple task correctly."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("fix typo in button label")
assert isinstance(result, ComplexityAssessment)
assert result.complexity == Complexity.SIMPLE
assert result.confidence > 0
assert "simple_keywords" in result.signals
assert result.estimated_files <= 3
def test_complex_task_analysis(self):
"""Analyzes a complex task correctly."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze(
"integrate kafka and redis with kubernetes deployment for microservice architecture"
)
assert result.complexity == Complexity.COMPLEX
assert len(result.external_integrations) > 0
assert result.infrastructure_changes is True
def test_standard_task_analysis(self):
"""Analyzes a standard task correctly."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("add new user profile feature with database storage")
assert result.complexity in [Complexity.STANDARD, Complexity.COMPLEX]
assert result.estimated_files > 1
def test_analysis_with_requirements(self):
"""Uses requirements data when provided."""
analyzer = ComplexityAnalyzer()
requirements = {
"services_involved": ["backend", "frontend", "worker"],
}
result = analyzer.analyze("add feature", requirements=requirements)
assert result.signals.get("explicit_services") == 3
assert result.estimated_services >= 3
def test_analysis_returns_assessment_object(self):
"""Returns ComplexityAssessment with all fields."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("test task")
assert hasattr(result, "complexity")
assert hasattr(result, "confidence")
assert hasattr(result, "signals")
assert hasattr(result, "reasoning")
assert hasattr(result, "estimated_files")
assert hasattr(result, "estimated_services")
assert hasattr(result, "external_integrations")
assert hasattr(result, "infrastructure_changes")
class TestSaveAssessment:
"""Tests for save_assessment() function."""
def test_saves_assessment_json(self, spec_dir: Path):
"""Saves assessment to complexity_assessment.json."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.85,
reasoning="Test reasoning",
estimated_files=5,
estimated_services=2,
)
result_path = save_assessment(spec_dir, assessment)
assert result_path.exists()
assert result_path.name == "complexity_assessment.json"
data = json.loads(result_path.read_text())
assert data["complexity"] == "standard"
assert data["confidence"] == 0.85
assert data["reasoning"] == "Test reasoning"
def test_saves_phases_to_run(self, spec_dir: Path):
"""Saves phases_to_run in output."""
assessment = ComplexityAssessment(
complexity=Complexity.SIMPLE,
confidence=0.9,
)
result_path = save_assessment(spec_dir, assessment)
data = json.loads(result_path.read_text())
assert "phases_to_run" in data
assert "discovery" in data["phases_to_run"]
def test_saves_dev_mode_flag(self, spec_dir: Path):
"""Saves dev_mode flag in output."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.8,
)
save_assessment(spec_dir, assessment, dev_mode=True)
data = json.loads((spec_dir / "complexity_assessment.json").read_text())
assert data["dev_mode"] is True
def test_saves_timestamp(self, spec_dir: Path):
"""Saves created_at timestamp."""
assessment = ComplexityAssessment(
complexity=Complexity.STANDARD,
confidence=0.8,
)
save_assessment(spec_dir, assessment)
data = json.loads((spec_dir / "complexity_assessment.json").read_text())
assert "created_at" in data
assert "T" in data["created_at"] # ISO format
class TestRunAIComplexityAssessment:
"""Tests for run_ai_complexity_assessment() async function."""
@pytest.mark.asyncio
async def test_returns_none_on_agent_failure(self, spec_dir: Path):
"""Returns None when agent fails."""
async def mock_agent(prompt_file, additional_context=None):
return (False, "Agent failed")
result = await run_ai_complexity_assessment(
spec_dir=spec_dir,
task_description="test task",
run_agent_fn=mock_agent,
)
assert result is None
@pytest.mark.asyncio
async def test_returns_none_on_missing_file(self, spec_dir: Path):
"""Returns None when assessment file not created."""
async def mock_agent(prompt_file, additional_context=None):
return (True, "Success but no file")
result = await run_ai_complexity_assessment(
spec_dir=spec_dir,
task_description="test task",
run_agent_fn=mock_agent,
)
assert result is None
@pytest.mark.asyncio
async def test_parses_ai_assessment(self, spec_dir: Path):
"""Parses AI assessment file correctly."""
# Pre-create the assessment file that the agent would create
assessment_data = {
"complexity": "standard",
"confidence": 0.9,
"reasoning": "AI determined standard",
"analysis": {
"scope": {
"estimated_files": 8,
"estimated_services": 2,
},
"integrations": {
"external_services": ["redis"],
},
"infrastructure": {
"docker_changes": True,
},
},
"recommended_phases": ["discovery", "requirements", "validation"],
"flags": {
"needs_research": True,
"needs_self_critique": False,
},
}
(spec_dir / "complexity_assessment.json").write_text(json.dumps(assessment_data))
async def mock_agent(prompt_file, additional_context=None):
return (True, "Assessment created")
result = await run_ai_complexity_assessment(
spec_dir=spec_dir,
task_description="test task",
run_agent_fn=mock_agent,
)
assert result is not None
assert result.complexity == Complexity.STANDARD
assert result.confidence == 0.9
assert result.recommended_phases == ["discovery", "requirements", "validation"]
assert result.needs_research is True
assert result.needs_self_critique is False
@pytest.mark.asyncio
async def test_includes_requirements_in_context(self, spec_dir: Path):
"""Includes requirements.json content in agent context."""
# Create requirements file
requirements = {
"task_description": "Test task from requirements",
"workflow_type": "feature",
"services_involved": ["backend", "frontend"],
"user_requirements": ["req1"],
"acceptance_criteria": ["crit1"],
"constraints": ["const1"],
}
(spec_dir / "requirements.json").write_text(json.dumps(requirements))
context_received = []
async def mock_agent(prompt_file, additional_context=None):
context_received.append(additional_context)
return (False, "Fail to inspect context")
await run_ai_complexity_assessment(
spec_dir=spec_dir,
task_description="test task",
run_agent_fn=mock_agent,
)
assert len(context_received) == 1
assert "Test task from requirements" in context_received[0]
assert "backend" in context_received[0]
@pytest.mark.asyncio
async def test_handles_exception_gracefully(self, spec_dir: Path):
"""Returns None on exception."""
async def mock_agent(prompt_file, additional_context=None):
raise Exception("Unexpected error")
result = await run_ai_complexity_assessment(
spec_dir=spec_dir,
task_description="test task",
run_agent_fn=mock_agent,
)
assert result is None
class TestKeywordLists:
"""Tests for keyword classification lists."""
def test_simple_keywords_are_lowercase(self):
"""All SIMPLE_KEYWORDS are lowercase."""
for kw in ComplexityAnalyzer.SIMPLE_KEYWORDS:
assert kw == kw.lower()
def test_complex_keywords_are_lowercase(self):
"""All COMPLEX_KEYWORDS are lowercase."""
for kw in ComplexityAnalyzer.COMPLEX_KEYWORDS:
assert kw == kw.lower()
def test_multi_service_keywords_are_lowercase(self):
"""All MULTI_SERVICE_KEYWORDS are lowercase."""
for kw in ComplexityAnalyzer.MULTI_SERVICE_KEYWORDS:
assert kw == kw.lower()
def test_keyword_lists_non_empty(self):
"""All keyword lists have entries."""
assert len(ComplexityAnalyzer.SIMPLE_KEYWORDS) > 0
assert len(ComplexityAnalyzer.COMPLEX_KEYWORDS) > 0
assert len(ComplexityAnalyzer.MULTI_SERVICE_KEYWORDS) > 0
def test_simple_complex_no_overlap(self):
"""SIMPLE and COMPLEX keywords don't overlap."""
simple_set = set(ComplexityAnalyzer.SIMPLE_KEYWORDS)
complex_set = set(ComplexityAnalyzer.COMPLEX_KEYWORDS)
overlap = simple_set.intersection(complex_set)
assert len(overlap) == 0, f"Overlapping keywords: {overlap}"
class TestEdgeCases:
"""Tests for edge cases and boundary conditions."""
def test_empty_task_description(self):
"""Handles empty task description."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("")
# Should return valid assessment
assert isinstance(result, ComplexityAssessment)
def test_very_long_task_description(self):
"""Handles very long task description."""
analyzer = ComplexityAnalyzer()
long_task = "implement feature " * 1000
result = analyzer.analyze(long_task)
assert isinstance(result, ComplexityAssessment)
def test_special_characters_in_task(self):
"""Handles special characters in task."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("fix bug in <Component /> with @decorator & regex /pattern/")
assert isinstance(result, ComplexityAssessment)
def test_unicode_in_task(self):
"""Handles unicode characters in task."""
analyzer = ComplexityAnalyzer()
result = analyzer.analyze("add emoji support for 🚀 and 日本語")
assert isinstance(result, ComplexityAssessment)
def test_case_insensitive_keyword_detection(self):
"""Keyword detection is case-insensitive."""
analyzer = ComplexityAnalyzer()
result1 = analyzer.analyze("FIX TYPO IN BUTTON")
result2 = analyzer.analyze("fix typo in button")
assert result1.signals["simple_keywords"] == result2.signals["simple_keywords"]
+950
View File
@@ -0,0 +1,950 @@
#!/usr/bin/env python3
"""
Tests for Spec Pipeline Phase Execution
========================================
Tests the PhaseExecutor class in auto-claude/spec/phases.py covering:
- PhaseResult dataclass
- All phase methods (discovery, requirements, context, etc.)
- Retry logic and error handling
- File existence checks and caching
"""
import json
import pytest
import sys
from pathlib import Path
from unittest.mock import MagicMock, AsyncMock, patch
# Mock ALL external dependencies before ANY imports from the spec module
# The import chain is: spec.phases -> spec.__init__ -> spec.pipeline -> client -> claude_agent_sdk
mock_sdk = MagicMock()
mock_sdk.ClaudeSDKClient = MagicMock()
mock_sdk.ClaudeCodeOptions = MagicMock()
mock_sdk.HookMatcher = MagicMock()
sys.modules['claude_code_sdk'] = mock_sdk
sys.modules['claude_code_sdk.types'] = mock_sdk
# Mock claude_agent_sdk
mock_agent_sdk = MagicMock()
mock_agent_sdk.ClaudeSDKClient = MagicMock()
mock_agent_sdk.ClaudeAgentOptions = MagicMock()
sys.modules['claude_agent_sdk'] = mock_agent_sdk
# Mock graphiti_providers module
mock_graphiti = MagicMock()
mock_graphiti.is_graphiti_enabled = MagicMock(return_value=False)
mock_graphiti.get_graph_hints = AsyncMock(return_value=[])
sys.modules['graphiti_providers'] = mock_graphiti
# Mock validate_spec module
mock_validate_spec = MagicMock()
mock_validate_spec.auto_fix_plan = MagicMock(return_value=False)
sys.modules['validate_spec'] = mock_validate_spec
# Mock client module to avoid circular imports
mock_client = MagicMock()
mock_client.create_client = MagicMock()
sys.modules['client'] = mock_client
# Now import the phases module directly (bypasses __init__.py issues)
from spec.phases import PhaseExecutor, PhaseResult, MAX_RETRIES
class TestPhaseResult:
"""Tests for PhaseResult dataclass."""
def test_phase_result_creation(self):
"""PhaseResult can be created with all fields."""
result = PhaseResult(
phase="discovery",
success=True,
output_files=["project_index.json"],
errors=[],
retries=0,
)
assert result.phase == "discovery"
assert result.success is True
assert result.output_files == ["project_index.json"]
assert result.errors == []
assert result.retries == 0
def test_phase_result_with_errors(self):
"""PhaseResult can store error messages."""
result = PhaseResult(
phase="context",
success=False,
output_files=[],
errors=["Attempt 1: Script failed", "Attempt 2: Timeout"],
retries=2,
)
assert result.success is False
assert len(result.errors) == 2
assert result.retries == 2
def test_phase_result_multiple_output_files(self):
"""PhaseResult can track multiple output files."""
result = PhaseResult(
phase="spec_writing",
success=True,
output_files=["spec.md", "implementation_plan.json"],
errors=[],
retries=0,
)
assert len(result.output_files) == 2
class TestPhaseExecutorInit:
"""Tests for PhaseExecutor initialization."""
def test_executor_initialization(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""PhaseExecutor initializes with all required parameters."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
assert executor.project_dir == temp_dir
assert executor.spec_dir == spec_dir
assert executor.task_description == "Test task"
def test_executor_stores_dependencies(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""PhaseExecutor stores all dependency objects."""
validator = mock_spec_validator()
agent_fn = mock_run_agent_fn()
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=validator,
run_agent_fn=agent_fn,
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
assert executor.spec_validator == validator
assert executor.run_agent_fn == agent_fn
assert executor.task_logger == mock_task_logger
assert executor.ui == mock_ui_module
class TestPhaseDiscovery:
"""Tests for phase_discovery method."""
@pytest.mark.asyncio
async def test_discovery_success(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Discovery phase succeeds when script creates project_index.json."""
# Create the project_index.json file
index_file = spec_dir / "project_index.json"
index_file.write_text(json.dumps({"files": [1, 2, 3], "project_type": "python"}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
with patch('spec.discovery.run_discovery_script', return_value=(True, "Created")):
with patch('spec.discovery.get_project_index_stats', return_value={"file_count": 3}):
result = await executor.phase_discovery()
assert result.success is True
assert result.phase == "discovery"
assert any("project_index.json" in f for f in result.output_files)
@pytest.mark.asyncio
async def test_discovery_retries_on_failure(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Discovery phase retries on failure."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
# Always fail
with patch('spec.discovery.run_discovery_script', return_value=(False, "Script failed")):
result = await executor.phase_discovery()
assert result.success is False
assert result.retries == MAX_RETRIES - 1
assert len(result.errors) == MAX_RETRIES
class TestPhaseHistoricalContext:
"""Tests for phase_historical_context method."""
@pytest.mark.asyncio
async def test_historical_context_file_exists(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Historical context phase returns early if hints file exists."""
hints_file = spec_dir / "graph_hints.json"
hints_file.write_text(json.dumps({"hints": [], "enabled": True}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_historical_context()
assert result.success is True
assert result.phase == "historical_context"
assert result.retries == 0
@pytest.mark.asyncio
async def test_historical_context_graphiti_disabled(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Historical context phase handles disabled Graphiti."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
with patch('graphiti_providers.is_graphiti_enabled', return_value=False):
result = await executor.phase_historical_context()
assert result.success is True
assert (spec_dir / "graph_hints.json").exists()
class TestPhaseRequirements:
"""Tests for phase_requirements method."""
@pytest.mark.asyncio
async def test_requirements_file_exists(
self,
spec_dir: Path,
temp_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Requirements phase returns early if file exists."""
requirements_file = spec_dir / "requirements.json"
requirements_file.write_text(json.dumps({"task_description": "Test"}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_requirements(interactive=False)
assert result.success is True
assert result.phase == "requirements"
assert result.retries == 0
@pytest.mark.asyncio
async def test_requirements_non_interactive_with_task(
self,
spec_dir: Path,
temp_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Requirements phase creates file from task description in non-interactive mode."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Add user authentication",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_requirements(interactive=False)
assert result.success is True
assert (spec_dir / "requirements.json").exists()
# Verify content
with open(spec_dir / "requirements.json") as f:
req = json.load(f)
assert req["task_description"] == "Add user authentication"
class TestPhaseContext:
"""Tests for phase_context method."""
@pytest.mark.asyncio
async def test_context_file_exists(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Context phase returns early if file exists."""
context_file = spec_dir / "context.json"
context_file.write_text(json.dumps({"task_description": "Test"}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_context()
assert result.success is True
assert result.phase == "context"
assert result.retries == 0
@pytest.mark.asyncio
async def test_context_discovery_success(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Context phase calls discovery script and succeeds."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
with patch('spec.context.run_context_discovery', return_value=(True, "Success")):
with patch('spec.context.get_context_stats', return_value={"files_to_modify": 5}):
result = await executor.phase_context()
assert result.success is True
@pytest.mark.asyncio
async def test_context_creates_minimal_on_failure(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Context phase creates minimal context when script fails."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
with patch('spec.context.run_context_discovery', return_value=(False, "Failed")):
with patch('spec.context.create_minimal_context') as mock_minimal:
result = await executor.phase_context()
mock_minimal.assert_called_once()
assert result.success is True # Creates minimal context as fallback
class TestPhaseQuickSpec:
"""Tests for phase_quick_spec method."""
@pytest.mark.asyncio
async def test_quick_spec_files_exist(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Quick spec phase returns early if files exist."""
(spec_dir / "spec.md").write_text("# Test Spec")
(spec_dir / "implementation_plan.json").write_text(json.dumps({"phases": []}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_quick_spec()
assert result.success is True
assert result.phase == "quick_spec"
assert result.retries == 0
@pytest.mark.asyncio
async def test_quick_spec_runs_agent(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Quick spec phase runs agent to create spec."""
# Agent creates spec.md on success
async def agent_side_effect(*args, **kwargs):
(spec_dir / "spec.md").write_text("# Generated Spec")
return (True, "Done")
agent_fn = AsyncMock(side_effect=agent_side_effect)
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=agent_fn,
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_quick_spec()
assert result.success is True
assert agent_fn.called
class TestPhaseResearch:
"""Tests for phase_research method."""
@pytest.mark.asyncio
async def test_research_file_exists(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Research phase returns early if file exists."""
(spec_dir / "research.json").write_text(json.dumps({"findings": []}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_research()
assert result.success is True
assert result.phase == "research"
assert result.retries == 0
@pytest.mark.asyncio
async def test_research_skipped_no_requirements(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Research phase skipped when no requirements.json."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_research()
assert result.success is True
assert (spec_dir / "research.json").exists()
class TestPhaseSpecWriting:
"""Tests for phase_spec_writing method."""
@pytest.mark.asyncio
async def test_spec_writing_file_exists_valid(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Spec writing phase returns early if valid spec exists."""
(spec_dir / "spec.md").write_text("# Test Spec\n\n## Overview\n")
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(spec_valid=True),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_spec_writing()
assert result.success is True
assert result.phase == "spec_writing"
assert result.retries == 0
@pytest.mark.asyncio
async def test_spec_writing_regenerates_invalid(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Spec writing phase regenerates invalid spec."""
(spec_dir / "spec.md").write_text("Invalid spec")
async def agent_side_effect(*args, **kwargs):
(spec_dir / "spec.md").write_text("# Valid Spec\n\n## Overview\n")
return (True, "Done")
agent_fn = AsyncMock(side_effect=agent_side_effect)
# First call returns invalid, subsequent calls return valid
validator = mock_spec_validator(spec_valid=False)
from unittest.mock import MagicMock as Mock
from dataclasses import dataclass
@dataclass
class MockResult:
valid: bool
checkpoint: str = "spec_document"
errors: list = None
fixes: list = None
def __post_init__(self):
self.errors = self.errors or []
self.fixes = self.fixes or []
call_count = [0]
def validate_spec_side_effect():
call_count[0] += 1
if call_count[0] == 1:
return MockResult(valid=False, errors=["Invalid"])
return MockResult(valid=True)
validator.validate_spec_document = Mock(side_effect=validate_spec_side_effect)
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=validator,
run_agent_fn=agent_fn,
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_spec_writing()
assert result.success is True
assert agent_fn.called
class TestPhaseSelfCritique:
"""Tests for phase_self_critique method."""
@pytest.mark.asyncio
async def test_self_critique_no_spec(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Self-critique fails if spec.md doesn't exist."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_self_critique()
assert result.success is False
assert "spec.md does not exist" in result.errors[0]
@pytest.mark.asyncio
async def test_self_critique_already_completed(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Self-critique returns early if already completed."""
(spec_dir / "spec.md").write_text("# Test Spec")
(spec_dir / "critique_report.json").write_text(json.dumps({
"issues_fixed": True,
"no_issues_found": False,
}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_self_critique()
assert result.success is True
assert result.retries == 0
class TestPhasePlanning:
"""Tests for phase_planning method."""
@pytest.mark.asyncio
async def test_planning_file_exists_valid(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Planning phase returns early if valid plan exists."""
(spec_dir / "implementation_plan.json").write_text(json.dumps({
"phases": [{"phase": 1, "subtasks": []}]
}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(plan_valid=True),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_planning()
assert result.success is True
assert result.phase == "planning"
assert result.retries == 0
class TestPhaseValidation:
"""Tests for phase_validation method."""
@pytest.mark.asyncio
async def test_validation_all_pass(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Validation phase passes when all validations pass."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(
spec_valid=True,
plan_valid=True,
context_valid=True,
all_valid=True,
),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_validation()
assert result.success is True
assert result.phase == "validation"
@pytest.mark.asyncio
async def test_validation_retries_on_failure(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Validation phase retries with auto-fix agent on failure."""
# Create agent mock that simulates failure
agent_fn = mock_run_agent_fn(success=False, output="Fix failed")
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(all_valid=False),
run_agent_fn=agent_fn,
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
result = await executor.phase_validation()
assert result.success is False
assert result.retries == MAX_RETRIES
class TestRunScript:
"""Tests for _run_script helper method."""
def test_run_script_not_found(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""_run_script returns False when script not found."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
success, output = executor._run_script("nonexistent.py", [])
assert success is False
assert "not found" in output.lower()
class TestMaxRetriesConstant:
"""Tests for MAX_RETRIES configuration."""
def test_max_retries_is_positive(self):
"""MAX_RETRIES is a positive integer."""
assert MAX_RETRIES > 0
assert isinstance(MAX_RETRIES, int)
def test_max_retries_reasonable(self):
"""MAX_RETRIES is a reasonable value."""
assert 1 <= MAX_RETRIES <= 10
class TestPhaseWorkflow:
"""Integration tests for phase workflow patterns."""
@pytest.mark.asyncio
async def test_phases_are_idempotent(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Running a phase twice with existing output is idempotent."""
# Pre-create files
(spec_dir / "requirements.json").write_text(json.dumps({"task_description": "Test"}))
(spec_dir / "context.json").write_text(json.dumps({"task_description": "Test"}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
# Run phases twice
result1 = await executor.phase_requirements(interactive=False)
result2 = await executor.phase_requirements(interactive=False)
assert result1.success is True
assert result2.success is True
assert result1.retries == 0
assert result2.retries == 0
@pytest.mark.asyncio
async def test_phases_log_to_task_logger(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Phases log messages to task logger."""
(spec_dir / "project_index.json").write_text(json.dumps({"files": []}))
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
with patch('spec.discovery.run_discovery_script', return_value=(True, "Success")):
with patch('spec.discovery.get_project_index_stats', return_value={"file_count": 10}):
await executor.phase_discovery()
# Verify logger was called
assert mock_task_logger.log.called
@pytest.mark.asyncio
async def test_phases_print_status(
self,
temp_dir: Path,
spec_dir: Path,
mock_run_agent_fn,
mock_task_logger,
mock_ui_module,
mock_spec_validator,
):
"""Phases print status messages via UI module."""
executor = PhaseExecutor(
project_dir=temp_dir,
spec_dir=spec_dir,
task_description="Test task",
spec_validator=mock_spec_validator(),
run_agent_fn=mock_run_agent_fn(),
task_logger=mock_task_logger,
ui_module=mock_ui_module,
)
await executor.phase_requirements(interactive=False)
# Verify UI print_status was called
assert mock_ui_module.print_status.called
+603
View File
@@ -0,0 +1,603 @@
#!/usr/bin/env python3
"""
Tests for Spec Pipeline Integration
====================================
Tests the spec/pipeline.py module functionality including:
- SpecOrchestrator initialization
- Spec directory creation and naming
- Orphaned pending folder cleanup
- Specs directory path resolution
"""
import json
import pytest
import sys
import time
from datetime import datetime, timedelta
from pathlib import Path
from unittest.mock import MagicMock, patch, AsyncMock
# Add auto-claude directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude"))
# Mock modules that have external dependencies
mock_sdk = MagicMock()
mock_sdk.ClaudeSDKClient = MagicMock()
mock_sdk.ClaudeCodeOptions = MagicMock()
mock_types = MagicMock()
mock_types.HookMatcher = MagicMock()
sys.modules['claude_code_sdk'] = mock_sdk
sys.modules['claude_code_sdk.types'] = mock_types
# Mock init module to prevent side effects
mock_init = MagicMock()
mock_init.init_auto_claude_dir = MagicMock(return_value=(Path("/tmp"), False))
sys.modules['init'] = mock_init
# Mock other external dependencies
mock_client = MagicMock()
mock_client.create_client = MagicMock()
sys.modules['client'] = mock_client
mock_review = MagicMock()
mock_review.ReviewState = MagicMock()
mock_review.run_review_checkpoint = MagicMock()
sys.modules['review'] = mock_review
mock_task_logger = MagicMock()
mock_task_logger.LogEntryType = MagicMock()
mock_task_logger.LogPhase = MagicMock()
mock_task_logger.get_task_logger = MagicMock()
mock_task_logger.update_task_logger_path = MagicMock()
sys.modules['task_logger'] = mock_task_logger
mock_ui = MagicMock()
mock_ui.Icons = MagicMock()
mock_ui.box = MagicMock(return_value="")
mock_ui.highlight = MagicMock(return_value="")
mock_ui.icon = MagicMock(return_value="")
mock_ui.muted = MagicMock(return_value="")
mock_ui.print_key_value = MagicMock()
mock_ui.print_section = MagicMock()
mock_ui.print_status = MagicMock()
sys.modules['ui'] = mock_ui
mock_validate_spec = MagicMock()
mock_validate_spec.SpecValidator = MagicMock()
sys.modules['validate_spec'] = mock_validate_spec
# Now import the module under test
from spec.pipeline import SpecOrchestrator, get_specs_dir
class TestGetSpecsDir:
"""Tests for get_specs_dir function."""
def test_returns_specs_path(self, temp_dir: Path):
"""Returns path to specs directory."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
result = get_specs_dir(temp_dir)
assert result == temp_dir / ".auto-claude" / "specs"
def test_calls_init_auto_claude_dir(self, temp_dir: Path):
"""Initializes auto-claude directory."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
get_specs_dir(temp_dir)
mock_init.assert_called_once_with(temp_dir)
def test_dev_mode_param_ignored(self, temp_dir: Path):
"""dev_mode parameter is deprecated and ignored."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
result1 = get_specs_dir(temp_dir, dev_mode=False)
result2 = get_specs_dir(temp_dir, dev_mode=True)
assert result1 == result2
class TestSpecOrchestratorInit:
"""Tests for SpecOrchestrator initialization."""
def test_init_with_project_dir(self, temp_dir: Path):
"""Initializes with project directory."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
task_description="Test task",
)
assert orchestrator.project_dir == temp_dir
assert orchestrator.task_description == "Test task"
def test_init_creates_spec_dir(self, temp_dir: Path):
"""Creates spec directory if not exists."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
task_description="Test task",
)
assert orchestrator.spec_dir.exists()
def test_init_with_spec_name(self, temp_dir: Path):
"""Uses provided spec name."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
spec_name="my-feature",
)
assert orchestrator.spec_dir.name == "my-feature"
def test_init_with_spec_dir(self, temp_dir: Path):
"""Uses provided spec directory."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
custom_spec_dir = specs_dir / "custom-spec"
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
spec_dir=custom_spec_dir,
)
assert orchestrator.spec_dir == custom_spec_dir
def test_init_default_model(self, temp_dir: Path):
"""Uses default model."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.model == "claude-opus-4-5-20251101"
def test_init_custom_model(self, temp_dir: Path):
"""Uses custom model."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
model="claude-sonnet-4-20250514",
)
assert orchestrator.model == "claude-sonnet-4-20250514"
class TestCreateSpecDir:
"""Tests for spec directory creation."""
def test_creates_numbered_directory(self, temp_dir: Path):
"""Creates numbered spec directory."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.spec_dir.name.startswith("001-")
assert "pending" in orchestrator.spec_dir.name
def test_increments_number(self, temp_dir: Path):
"""Increments directory number."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create existing directories
(specs_dir / "001-first").mkdir()
(specs_dir / "002-second").mkdir()
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.spec_dir.name.startswith("003-")
def test_finds_highest_number(self, temp_dir: Path):
"""Finds highest existing number."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create non-sequential directories
(specs_dir / "001-first").mkdir()
(specs_dir / "005-fifth").mkdir()
(specs_dir / "003-third").mkdir()
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.spec_dir.name.startswith("006-")
class TestGenerateSpecName:
"""Tests for spec name generation."""
def test_generates_kebab_case(self, temp_dir: Path):
"""Generates kebab-case name."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
name = orchestrator._generate_spec_name("Add User Authentication")
assert name == "user-authentication"
def test_skips_common_words(self, temp_dir: Path):
"""Skips common words like 'the', 'a', 'add'."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
name = orchestrator._generate_spec_name("Create the new login page")
# Should skip 'create', 'the', 'new'
assert "login" in name
assert "page" in name
def test_limits_to_four_words(self, temp_dir: Path):
"""Limits name to four meaningful words."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
name = orchestrator._generate_spec_name(
"Implement user authentication system with OAuth providers and session management"
)
parts = name.split("-")
assert len(parts) <= 4
def test_handles_special_characters(self, temp_dir: Path):
"""Handles special characters in task description."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
name = orchestrator._generate_spec_name("Add OAuth2.0 (Google) authentication!")
assert "-" in name or name == "spec"
assert "!" not in name
assert "(" not in name
def test_returns_spec_for_empty_description(self, temp_dir: Path):
"""Returns 'spec' for empty description."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
name = orchestrator._generate_spec_name("")
assert name == "spec"
class TestCleanupOrphanedPendingFolders:
"""Tests for orphaned pending folder cleanup."""
def test_removes_empty_pending_folder(self, temp_dir: Path):
"""Removes empty pending folders older than 10 minutes."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create non-pending folders to establish numbering context
(specs_dir / "001-feature").mkdir()
(specs_dir / "003-another").mkdir()
# Create old EMPTY pending folder at 002
old_pending = specs_dir / "002-pending"
old_pending.mkdir()
# Set modification time to 15 minutes ago
old_time = time.time() - (15 * 60)
import os
os.utime(old_pending, (old_time, old_time))
# Store the inode to verify it's actually deleted and recreated
old_inode = old_pending.stat().st_ino
# Creating orchestrator triggers cleanup
# The cleanup removes 002-pending (empty and old)
# Then _create_spec_dir creates 004-pending (after 003)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
# The orchestrator should have created a new folder at 004
assert orchestrator.spec_dir.name.startswith("004-")
# The 002-pending folder no longer exists (cleaned up)
assert not old_pending.exists()
def test_keeps_folder_with_requirements(self, temp_dir: Path):
"""Keeps pending folder with requirements.json."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create pending folder with requirements
pending_with_req = specs_dir / "001-pending"
pending_with_req.mkdir()
(pending_with_req / "requirements.json").write_text("{}")
# Set modification time to 15 minutes ago
old_time = time.time() - (15 * 60)
import os
os.utime(pending_with_req, (old_time, old_time))
# Creating orchestrator triggers cleanup
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert pending_with_req.exists()
def test_keeps_folder_with_spec(self, temp_dir: Path):
"""Keeps pending folder with spec.md."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create pending folder with spec
pending_with_spec = specs_dir / "001-pending"
pending_with_spec.mkdir()
(pending_with_spec / "spec.md").write_text("# Spec")
# Set modification time to 15 minutes ago
old_time = time.time() - (15 * 60)
import os
os.utime(pending_with_spec, (old_time, old_time))
# Creating orchestrator triggers cleanup
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert pending_with_spec.exists()
def test_keeps_recent_pending_folder(self, temp_dir: Path):
"""Keeps pending folder younger than 10 minutes."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create recent pending folder (no need to modify time, it's fresh)
recent_pending = specs_dir / "001-pending"
recent_pending.mkdir()
# Creating orchestrator triggers cleanup
orchestrator = SpecOrchestrator(project_dir=temp_dir)
# Recent folder should still exist (unless orchestrator created 002-pending)
# The folder might be gone if orchestrator picked a different name
# So we check the spec dir count instead
assert any(d.name.endswith("-pending") for d in specs_dir.iterdir())
class TestRenameSpecDirFromRequirements:
"""Tests for renaming spec directory from requirements."""
def test_renames_from_task_description(self, temp_dir: Path):
"""Renames spec dir based on requirements task description."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
# Write requirements
requirements = {
"task_description": "Add user authentication system"
}
(orchestrator.spec_dir / "requirements.json").write_text(
json.dumps(requirements)
)
# Rename
result = orchestrator._rename_spec_dir_from_requirements()
assert result is True
assert "pending" not in orchestrator.spec_dir.name
assert "user" in orchestrator.spec_dir.name or "authentication" in orchestrator.spec_dir.name
def test_returns_false_no_requirements(self, temp_dir: Path):
"""Returns False when no requirements file."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
result = orchestrator._rename_spec_dir_from_requirements()
assert result is False
def test_returns_false_empty_task_description(self, temp_dir: Path):
"""Returns False when task description is empty."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
# Write requirements with empty task
requirements = {"task_description": ""}
(orchestrator.spec_dir / "requirements.json").write_text(
json.dumps(requirements)
)
result = orchestrator._rename_spec_dir_from_requirements()
assert result is False
def test_skips_rename_if_not_pending(self, temp_dir: Path):
"""Skips rename if directory is not a pending folder."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
# Create a named spec dir
named_dir = specs_dir / "001-my-feature"
named_dir.mkdir()
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
spec_dir=named_dir,
)
# Write requirements
requirements = {"task_description": "Different name task"}
(orchestrator.spec_dir / "requirements.json").write_text(
json.dumps(requirements)
)
result = orchestrator._rename_spec_dir_from_requirements()
# Should return True (no error) but not rename
assert result is True
assert orchestrator.spec_dir.name == "001-my-feature"
class TestComplexityOverride:
"""Tests for complexity override configuration."""
def test_sets_complexity_override(self, temp_dir: Path):
"""Sets complexity override."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
complexity_override="simple",
)
assert orchestrator.complexity_override == "simple"
def test_default_use_ai_assessment(self, temp_dir: Path):
"""Default uses AI assessment."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.use_ai_assessment is True
def test_disable_ai_assessment(self, temp_dir: Path):
"""Can disable AI assessment."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
use_ai_assessment=False,
)
assert orchestrator.use_ai_assessment is False
class TestSpecOrchestratorDevMode:
"""Tests for dev mode configuration."""
def test_default_dev_mode_false(self, temp_dir: Path):
"""Dev mode is False by default."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.dev_mode is False
def test_enable_dev_mode(self, temp_dir: Path):
"""Can enable dev mode."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(
project_dir=temp_dir,
dev_mode=True,
)
assert orchestrator.dev_mode is True
class TestSpecOrchestratorValidator:
"""Tests for SpecValidator integration."""
def test_creates_validator(self, temp_dir: Path):
"""Creates SpecValidator instance."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.validator is not None
class TestSpecOrchestratorAssessment:
"""Tests for complexity assessment state."""
def test_assessment_initially_none(self, temp_dir: Path):
"""Assessment is None initially."""
with patch('spec.pipeline.init_auto_claude_dir') as mock_init:
mock_init.return_value = (temp_dir / ".auto-claude", False)
specs_dir = temp_dir / ".auto-claude" / "specs"
specs_dir.mkdir(parents=True, exist_ok=True)
orchestrator = SpecOrchestrator(project_dir=temp_dir)
assert orchestrator.assessment is None
+14 -14
View File
@@ -137,7 +137,7 @@ class TestSetupWorkspace:
def test_setup_direct_mode(self, temp_git_repo: Path): def test_setup_direct_mode(self, temp_git_repo: Path):
"""Direct mode returns project dir and no manager.""" """Direct mode returns project dir and no manager."""
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.DIRECT, WorkspaceMode.DIRECT,
@@ -148,7 +148,7 @@ class TestSetupWorkspace:
def test_setup_isolated_mode(self, temp_git_repo: Path): def test_setup_isolated_mode(self, temp_git_repo: Path):
"""Isolated mode creates worktree and returns manager.""" """Isolated mode creates worktree and returns manager."""
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
TEST_SPEC_NAME, TEST_SPEC_NAME,
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -177,7 +177,7 @@ class TestWorkspaceUtilities:
def test_per_spec_worktree_naming(self, temp_git_repo: Path): def test_per_spec_worktree_naming(self, temp_git_repo: Path):
"""Per-spec architecture uses spec name for worktree directory.""" """Per-spec architecture uses spec name for worktree directory."""
spec_name = "my-spec-001" spec_name = "my-spec-001"
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
spec_name, spec_name,
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -194,7 +194,7 @@ class TestWorkspaceIntegration:
def test_isolated_workflow(self, temp_git_repo: Path): def test_isolated_workflow(self, temp_git_repo: Path):
"""Full isolated workflow: setup -> work -> finalize.""" """Full isolated workflow: setup -> work -> finalize."""
# Setup isolated workspace # Setup isolated workspace
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -212,7 +212,7 @@ class TestWorkspaceIntegration:
def test_direct_workflow(self, temp_git_repo: Path): def test_direct_workflow(self, temp_git_repo: Path):
"""Full direct workflow: setup -> work.""" """Full direct workflow: setup -> work."""
# Setup direct workspace # Setup direct workspace
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.DIRECT, WorkspaceMode.DIRECT,
@@ -230,7 +230,7 @@ class TestWorkspaceIntegration:
def test_isolated_merge(self, temp_git_repo: Path): def test_isolated_merge(self, temp_git_repo: Path):
"""Can merge isolated workspace back to main.""" """Can merge isolated workspace back to main."""
# Setup # Setup
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -258,7 +258,7 @@ class TestWorkspaceCleanup:
def test_cleanup_after_merge(self, temp_git_repo: Path): def test_cleanup_after_merge(self, temp_git_repo: Path):
"""Workspace is cleaned up after merge with delete_after=True.""" """Workspace is cleaned up after merge with delete_after=True."""
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -276,7 +276,7 @@ class TestWorkspaceCleanup:
def test_workspace_preserved_after_merge_no_delete(self, temp_git_repo: Path): def test_workspace_preserved_after_merge_no_delete(self, temp_git_repo: Path):
"""Workspace preserved after merge with delete_after=False.""" """Workspace preserved after merge with delete_after=False."""
working_dir, manager = setup_workspace( working_dir, manager, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -299,7 +299,7 @@ class TestWorkspaceReuse:
def test_reuse_existing_workspace(self, temp_git_repo: Path): def test_reuse_existing_workspace(self, temp_git_repo: Path):
"""Can reuse existing workspace on second setup.""" """Can reuse existing workspace on second setup."""
# First setup # First setup
working_dir1, manager1 = setup_workspace( working_dir1, manager1, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -309,7 +309,7 @@ class TestWorkspaceReuse:
(working_dir1 / "marker.txt").write_text("marker") (working_dir1 / "marker.txt").write_text("marker")
# Second setup (should reuse) # Second setup (should reuse)
working_dir2, manager2 = setup_workspace( working_dir2, manager2, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -342,7 +342,7 @@ class TestPerSpecWorktreeName:
def test_worktree_named_after_spec(self, temp_git_repo: Path): def test_worktree_named_after_spec(self, temp_git_repo: Path):
"""Worktree is named after the spec.""" """Worktree is named after the spec."""
spec_name = "spec-1" spec_name = "spec-1"
working_dir, _ = setup_workspace( working_dir, _, _ = setup_workspace(
temp_git_repo, temp_git_repo,
spec_name, spec_name,
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -353,13 +353,13 @@ class TestPerSpecWorktreeName:
def test_different_specs_get_different_worktrees(self, temp_git_repo: Path): def test_different_specs_get_different_worktrees(self, temp_git_repo: Path):
"""Different specs create separate worktrees.""" """Different specs create separate worktrees."""
working_dir1, _ = setup_workspace( working_dir1, _, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"spec-1", "spec-1",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
) )
working_dir2, _ = setup_workspace( working_dir2, _, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"spec-2", "spec-2",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
@@ -372,7 +372,7 @@ class TestPerSpecWorktreeName:
def test_worktree_path_in_worktrees_dir(self, temp_git_repo: Path): def test_worktree_path_in_worktrees_dir(self, temp_git_repo: Path):
"""Worktree is created in .worktrees directory.""" """Worktree is created in .worktrees directory."""
working_dir, _ = setup_workspace( working_dir, _, _ = setup_workspace(
temp_git_repo, temp_git_repo,
"test-spec", "test-spec",
WorkspaceMode.ISOLATED, WorkspaceMode.ISOLATED,
+59 -55
View File
@@ -63,41 +63,41 @@ class TestWorktreeCreation:
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info = manager.create("test-worker") info = manager.create_worktree("test-spec")
assert info.path.exists() assert info.path.exists()
assert info.branch == "auto-claude/test-worker" assert info.branch == "auto-claude/test-spec"
assert info.is_active is True assert info.is_active is True
assert (info.path / "README.md").exists() assert (info.path / "README.md").exists()
def test_create_worktree_with_custom_branch(self, temp_git_repo: Path): def test_create_worktree_with_spec_name(self, temp_git_repo: Path):
"""Can create worktree with custom branch name.""" """Worktree branch is derived from spec name."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info = manager.create("test-worker", branch_name="my-feature-branch") info = manager.create_worktree("my-feature-spec")
assert info.branch == "my-feature-branch" assert info.branch == "auto-claude/my-feature-spec"
def test_create_replaces_existing_worktree(self, temp_git_repo: Path): def test_get_or_create_replaces_existing_worktree(self, temp_git_repo: Path):
"""Creating worktree with same name replaces existing one.""" """get_or_create_worktree returns existing worktree."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info1 = manager.create("test-worker") info1 = manager.create_worktree("test-spec")
# Create a file in the worktree # Create a file in the worktree
(info1.path / "test-file.txt").write_text("test") (info1.path / "test-file.txt").write_text("test")
# Create again should work (replacing the old one) # get_or_create should return existing
info2 = manager.create("test-worker") info2 = manager.get_or_create_worktree("test-spec")
assert info2.path.exists() assert info2.path.exists()
# The test file should be gone (fresh worktree) # The test file should still be there (same worktree)
assert not (info2.path / "test-file.txt").exists() assert (info2.path / "test-file.txt").exists()
class TestStagingWorktree: class TestStagingWorktree:
"""Tests for staging worktree operations.""" """Tests for staging worktree operations (backward compatibility)."""
def test_get_or_create_staging_creates_new(self, temp_git_repo: Path): def test_get_or_create_staging_creates_new(self, temp_git_repo: Path):
"""Creates staging worktree if it doesn't exist.""" """Creates staging worktree if it doesn't exist."""
@@ -107,8 +107,9 @@ class TestStagingWorktree:
info = manager.get_or_create_staging("test-spec") info = manager.get_or_create_staging("test-spec")
assert info.path.exists() assert info.path.exists()
assert info.path.name == STAGING_WORKTREE_NAME # Staging is now per-spec, worktree is named after spec
assert "test-spec" in info.branch assert info.path.name == "test-spec"
assert "auto-claude/test-spec" in info.branch
def test_get_or_create_staging_returns_existing(self, temp_git_repo: Path): def test_get_or_create_staging_returns_existing(self, temp_git_repo: Path):
"""Returns existing staging worktree without recreating.""" """Returns existing staging worktree without recreating."""
@@ -148,7 +149,8 @@ class TestStagingWorktree:
path = manager.get_staging_path() path = manager.get_staging_path()
assert path is not None assert path is not None
assert path.name == STAGING_WORKTREE_NAME # Staging is now per-spec, path is named after spec
assert path.name == "test-spec"
def test_get_staging_info(self, temp_git_repo: Path): def test_get_staging_info(self, temp_git_repo: Path):
"""get_staging_info returns WorktreeInfo.""" """get_staging_info returns WorktreeInfo."""
@@ -170,9 +172,9 @@ class TestWorktreeRemoval:
"""Can remove a worktree.""" """Can remove a worktree."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info = manager.create("test-worker") info = manager.create_worktree("test-spec")
manager.remove("test-worker") manager.remove_worktree("test-spec")
assert not info.path.exists() assert not info.path.exists()
@@ -191,10 +193,10 @@ class TestWorktreeRemoval:
"""Removing worktree can also delete the branch.""" """Removing worktree can also delete the branch."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info = manager.create("test-worker") info = manager.create_worktree("test-spec")
branch_name = info.branch branch_name = info.branch
manager.remove("test-worker", delete_branch=True) manager.remove_worktree("test-spec", delete_branch=True)
# Verify branch is deleted # Verify branch is deleted
result = subprocess.run( result = subprocess.run(
@@ -257,14 +259,13 @@ class TestWorktreeCommitAndMerge:
subprocess.run(["git", "checkout", manager.base_branch], cwd=temp_git_repo, capture_output=True) subprocess.run(["git", "checkout", manager.base_branch], cwd=temp_git_repo, capture_output=True)
assert (temp_git_repo / "feature.txt").exists() assert (temp_git_repo / "feature.txt").exists()
def test_merge_branch_to_staging(self, temp_git_repo: Path): def test_merge_worktree(self, temp_git_repo: Path):
"""Can merge a branch into staging worktree.""" """Can merge a worktree back to main."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
manager.get_or_create_staging("test-spec")
# Create another worktree with changes # Create a worktree with changes
worker_info = manager.create("worker-1") worker_info = manager.create_worktree("worker-spec")
(worker_info.path / "worker-file.txt").write_text("worker content") (worker_info.path / "worker-file.txt").write_text("worker content")
subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True) subprocess.run(["git", "add", "."], cwd=worker_info.path, capture_output=True)
subprocess.run( subprocess.run(
@@ -272,14 +273,14 @@ class TestWorktreeCommitAndMerge:
cwd=worker_info.path, capture_output=True cwd=worker_info.path, capture_output=True
) )
# Merge worker branch into staging # Merge worktree back to main
result = manager.merge_branch_to_staging(worker_info.branch) result = manager.merge_worktree("worker-spec", delete_after=False)
assert result is True assert result is True
# Verify file is in staging # Verify file is in main branch
staging_path = manager.get_staging_path() subprocess.run(["git", "checkout", manager.base_branch], cwd=temp_git_repo, capture_output=True)
assert (staging_path / "worker-file.txt").exists() assert (temp_git_repo / "worker-file.txt").exists()
class TestChangeTracking: class TestChangeTracking:
@@ -313,7 +314,7 @@ class TestChangeTracking:
(info.path / "README.md").write_text("modified") (info.path / "README.md").write_text("modified")
manager.commit_in_staging("Changes") manager.commit_in_staging("Changes")
summary = manager.get_change_summary() summary = manager.get_change_summary("test-spec")
assert summary["new_files"] == 1 # new-file.txt assert summary["new_files"] == 1 # new-file.txt
assert summary["modified_files"] == 1 # README.md assert summary["modified_files"] == 1 # README.md
@@ -328,7 +329,7 @@ class TestChangeTracking:
(info.path / "added.txt").write_text("new file") (info.path / "added.txt").write_text("new file")
manager.commit_in_staging("Add file") manager.commit_in_staging("Add file")
files = manager.get_changed_files() files = manager.get_changed_files("test-spec")
assert len(files) > 0 assert len(files) > 0
file_names = [f[1] for f in files] file_names = [f[1] for f in files]
@@ -339,34 +340,34 @@ class TestWorktreeUtilities:
"""Tests for utility methods.""" """Tests for utility methods."""
def test_list_worktrees(self, temp_git_repo: Path): def test_list_worktrees(self, temp_git_repo: Path):
"""list_worktrees returns active worktrees.""" """list_all_worktrees returns active worktrees."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
manager.create("worker-1") manager.create_worktree("spec-1")
manager.create("worker-2") manager.create_worktree("spec-2")
worktrees = manager.list_worktrees() worktrees = manager.list_all_worktrees()
assert len(worktrees) == 2 assert len(worktrees) == 2
def test_get_info(self, temp_git_repo: Path): def test_get_info(self, temp_git_repo: Path):
"""get_info returns correct WorktreeInfo.""" """get_worktree_info returns correct WorktreeInfo."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
manager.create("test-worker") manager.create_worktree("test-spec")
info = manager.get_info("test-worker") info = manager.get_worktree_info("test-spec")
assert info is not None assert info is not None
assert info.branch == "auto-claude/test-worker" assert info.branch == "auto-claude/test-spec"
def test_get_worktree_path(self, temp_git_repo: Path): def test_get_worktree_path(self, temp_git_repo: Path):
"""get_worktree_path returns correct path.""" """get_worktree_path returns correct path."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
info = manager.create("test-worker") info = manager.create_worktree("test-spec")
path = manager.get_worktree_path("test-worker") path = manager.get_worktree_path("test-spec")
assert path == info.path assert path == info.path
@@ -374,25 +375,28 @@ class TestWorktreeUtilities:
"""cleanup_all removes all worktrees.""" """cleanup_all removes all worktrees."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
manager.create("worker-1") manager.create_worktree("spec-1")
manager.create("worker-2") manager.create_worktree("spec-2")
manager.get_or_create_staging("test-spec") manager.get_or_create_staging("test-spec")
manager.cleanup_all() manager.cleanup_all()
assert len(manager.list_worktrees()) == 0 assert len(manager.list_all_worktrees()) == 0
def test_cleanup_workers_only_preserves_staging(self, temp_git_repo: Path): def test_cleanup_stale_worktrees(self, temp_git_repo: Path):
"""cleanup_workers_only removes workers but keeps staging.""" """cleanup_stale_worktrees removes directories without git tracking."""
manager = WorktreeManager(temp_git_repo) manager = WorktreeManager(temp_git_repo)
manager.setup() manager.setup()
manager.create("worker-1")
manager.get_or_create_staging("test-spec")
manager.cleanup_workers_only() # Create a stale worktree directory (exists but not tracked by git)
stale_dir = manager.worktrees_dir / "stale-worktree"
stale_dir.mkdir(parents=True, exist_ok=True)
assert manager.staging_exists() is True # This should clean up the stale directory
assert manager.get_info("worker-1") is None manager.cleanup_stale_worktrees()
# Stale directory should be removed
assert not stale_dir.exists()
def test_get_test_commands_python(self, temp_git_repo: Path): def test_get_test_commands_python(self, temp_git_repo: Path):
"""get_test_commands detects Python project commands.""" """get_test_commands detects Python project commands."""
@@ -403,7 +407,7 @@ class TestWorktreeUtilities:
# Create requirements.txt # Create requirements.txt
(info.path / "requirements.txt").write_text("flask\n") (info.path / "requirements.txt").write_text("flask\n")
commands = manager.get_test_commands(info.path) commands = manager.get_test_commands("test-spec")
assert any("pip" in cmd for cmd in commands) assert any("pip" in cmd for cmd in commands)
@@ -416,6 +420,6 @@ class TestWorktreeUtilities:
# Create package.json # Create package.json
(info.path / "package.json").write_text('{"name": "test"}') (info.path / "package.json").write_text('{"name": "test"}')
commands = manager.get_test_commands(info.path) commands = manager.get_test_commands("test-spec")
assert any("npm" in cmd for cmd in commands) assert any("npm" in cmd for cmd in commands)