electron-rare/Aperant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
142cd67c88924ea1db46b1c6bfdc78a1c56cd655
Aperant/tests/conftest.py
T
AndyMik90 142cd67c88 Getting ready for the lint gods
2025-12-15 19:49:58 +01:00

843 lines
26 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Pytest Configuration and Shared Fixtures
=========================================
Provides common test fixtures for the Auto-Build Framework test suite.
"""
import json
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from typing import Generator
import pytest
# Add auto-claude directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent / "auto-claude"))
# =============================================================================
# DIRECTORY FIXTURES
# =============================================================================
@pytest.fixture
def temp_dir() -> Generator[Path, None, None]:
"""Create a temporary directory that's cleaned up after the test."""
temp_path = Path(tempfile.mkdtemp())
yield temp_path
shutil.rmtree(temp_path, ignore_errors=True)
@pytest.fixture
def temp_git_repo(temp_dir: Path) -> Generator[Path, None, None]:
"""Create a temporary git repository with initial commit."""
# Initialize git repo
subprocess.run(["git", "init"], cwd=temp_dir, capture_output=True, check=True)
subprocess.run(
["git", "config", "user.email", "test@example.com"],
cwd=temp_dir, capture_output=True
)
subprocess.run(
["git", "config", "user.name", "Test User"],
cwd=temp_dir, capture_output=True
)
# Create initial commit
test_file = temp_dir / "README.md"
test_file.write_text("# Test Project\n")
subprocess.run(["git", "add", "."], cwd=temp_dir, capture_output=True)
subprocess.run(
["git", "commit", "-m", "Initial commit"],
cwd=temp_dir, capture_output=True
)
yield temp_dir
@pytest.fixture
def spec_dir(temp_dir: Path) -> Path:
"""Create a spec directory inside temp_dir."""
spec_path = temp_dir / "spec"
spec_path.mkdir(parents=True)
return spec_path
# =============================================================================
# PROJECT STRUCTURE FIXTURES
# =============================================================================
@pytest.fixture
def python_project(temp_git_repo: Path) -> Path:
"""Create a sample Python project structure."""
# Create pyproject.toml
pyproject = {
"project": {
"name": "test-project",
"version": "0.1.0",
"dependencies": [
"flask>=2.0",
"pytest>=7.0",
"sqlalchemy>=2.0",
],
},
"tool": {
"pytest": {"testpaths": ["tests"]},
"ruff": {"line-length": 100},
},
}
import tomllib
# Write as TOML (we'll write manually since tomllib is read-only)
toml_content = """[project]
name = "test-project"
version = "0.1.0"
dependencies = [
"flask>=2.0",
"pytest>=7.0",
"sqlalchemy>=2.0",
]
[tool.pytest]
testpaths = ["tests"]
[tool.ruff]
line-length = 100
"""
(temp_git_repo / "pyproject.toml").write_text(toml_content)
# Create Python files
(temp_git_repo / "app").mkdir()
(temp_git_repo / "app" / "__init__.py").write_text("# App module\n")
(temp_git_repo / "app" / "main.py").write_text("def main():\n pass\n")
# Create .env file
(temp_git_repo / ".env").write_text("DATABASE_URL=postgresql://localhost/test\n")
# Commit changes
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
subprocess.run(
["git", "commit", "-m", "Add Python project structure"],
cwd=temp_git_repo, capture_output=True
)
return temp_git_repo
@pytest.fixture
def node_project(temp_git_repo: Path) -> Path:
"""Create a sample Node.js project structure."""
package_json = {
"name": "test-project",
"version": "1.0.0",
"scripts": {
"dev": "next dev",
"build": "next build",
"test": "jest",
"lint": "eslint .",
},
"dependencies": {
"next": "^14.0.0",
"react": "^18.0.0",
"prisma": "^5.0.0",
},
"devDependencies": {
"jest": "^29.0.0",
"eslint": "^8.0.0",
"typescript": "^5.0.0",
},
}
(temp_git_repo / "package.json").write_text(json.dumps(package_json, indent=2))
(temp_git_repo / "tsconfig.json").write_text('{"compilerOptions": {}}')
# Create source files
(temp_git_repo / "src").mkdir()
(temp_git_repo / "src" / "index.ts").write_text("export const main = () => {};\n")
# Commit changes
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
subprocess.run(
["git", "commit", "-m", "Add Node.js project structure"],
cwd=temp_git_repo, capture_output=True
)
return temp_git_repo
@pytest.fixture
def docker_project(temp_git_repo: Path) -> Path:
"""Create a project with Docker configuration."""
# Dockerfile
dockerfile = """FROM python:3.11-slim
WORKDIR /app
COPY . .
RUN pip install -r requirements.txt
CMD ["python", "main.py"]
"""
(temp_git_repo / "Dockerfile").write_text(dockerfile)
# docker-compose.yml
compose = """services:
app:
build: .
ports:
- "8000:8000"
postgres:
image: postgres:15
environment:
POSTGRES_DB: test
redis:
image: redis:7
"""
(temp_git_repo / "docker-compose.yml").write_text(compose)
# requirements.txt
(temp_git_repo / "requirements.txt").write_text("flask\nredis\npsycopg2-binary\n")
# Commit changes
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
subprocess.run(
["git", "commit", "-m", "Add Docker configuration"],
cwd=temp_git_repo, capture_output=True
)
return temp_git_repo
# =============================================================================
# IMPLEMENTATION PLAN FIXTURES
# =============================================================================
@pytest.fixture
def sample_implementation_plan() -> dict:
"""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",
"type": "setup",
"chunks": [
{
"id": "chunk-1-1",
"description": "Add avatar fields to User model",
"service": "backend",
"status": "completed",
"files_to_modify": ["app/models/user.py"],
"files_to_create": ["migrations/add_avatar.py"],
},
{
"id": "chunk-1-2",
"description": "POST /api/users/avatar endpoint",
"service": "backend",
"status": "pending",
"files_to_modify": ["app/routes/users.py"],
},
],
"depends_on": [],
},
{
"phase": 2,
"name": "Worker Pipeline",
"type": "implementation",
"chunks": [
{
"id": "chunk-2-1",
"description": "Image processing task",
"service": "worker",
"status": "pending",
"files_to_create": ["app/tasks/images.py"],
},
],
"depends_on": [1],
},
{
"phase": 3,
"name": "Frontend",
"type": "implementation",
"chunks": [
{
"id": "chunk-3-1",
"description": "AvatarUpload component",
"service": "frontend",
"status": "pending",
"files_to_create": ["src/components/AvatarUpload.tsx"],
},
],
"depends_on": [1],
},
],
"final_acceptance": [
"User can upload avatar from profile page",
"Avatar is automatically resized",
],
}
@pytest.fixture
def implementation_plan_file(spec_dir: Path, sample_implementation_plan: dict) -> Path:
"""Create an implementation_plan.json file in the spec directory."""
plan_file = spec_dir / "implementation_plan.json"
plan_file.write_text(json.dumps(sample_implementation_plan, indent=2))
return plan_file
# =============================================================================
# SPEC FIXTURES
# =============================================================================
@pytest.fixture
def sample_spec() -> str:
"""Return a sample spec content."""
return """# Avatar Upload Feature
## Overview
Allow users to upload and manage their profile avatars.
## Requirements
1. Users can upload PNG, JPG, or WebP images
2. Images are automatically resized to 200x200
3. Original images are stored for future cropping
4. Upload progress is shown in UI
## Acceptance Criteria
- [ ] POST /api/users/avatar endpoint accepts image uploads
- [ ] Images are processed asynchronously by worker
- [ ] Frontend shows upload progress
- [ ] Avatar displays correctly after upload
"""
@pytest.fixture
def spec_file(spec_dir: Path, sample_spec: str) -> Path:
"""Create a spec.md file in the spec directory."""
spec_file = spec_dir / "spec.md"
spec_file.write_text(sample_spec)
return spec_file
# =============================================================================
# QA FIXTURES
# =============================================================================
@pytest.fixture
def qa_signoff_approved() -> dict:
"""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() -> dict:
"""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"},
],
}
# =============================================================================
# HELPER FUNCTIONS
# =============================================================================
@pytest.fixture
def make_commit(temp_git_repo: Path):
"""Factory fixture to create commits."""
def _make_commit(filename: str, content: str, message: str) -> str:
filepath = temp_git_repo / filename
filepath.parent.mkdir(parents=True, exist_ok=True)
filepath.write_text(content)
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
subprocess.run(
["git", "commit", "-m", message],
cwd=temp_git_repo, capture_output=True
)
result = subprocess.run(
["git", "rev-parse", "HEAD"],
cwd=temp_git_repo, capture_output=True, text=True
)
return result.stdout.strip()
return _make_commit
@pytest.fixture
def stage_files(temp_git_repo: Path):
"""Factory fixture to stage files without committing."""
def _stage_files(files: dict[str, str]) -> None:
for filename, content in files.items():
filepath = temp_git_repo / filename
filepath.parent.mkdir(parents=True, exist_ok=True)
filepath.write_text(content)
subprocess.run(["git", "add", "."], cwd=temp_git_repo, capture_output=True)
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
Reference in New Issue View Git Blame Copy Permalink