341 lines
11 KiB
Python
341 lines
11 KiB
Python
"""
|
|
Environment Management Utilities
|
|
|
|
Utilities for managing test environments, database setup, and service mocking.
|
|
"""
|
|
|
|
import os
|
|
import tempfile
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Dict, List, Optional, Any
|
|
import logging
|
|
|
|
|
|
class TestEnvironmentManager:
|
|
"""Manages test environment setup and cleanup."""
|
|
|
|
def __init__(self):
|
|
"""Initialize environment manager."""
|
|
self.logger = logging.getLogger("TestEnvironmentManager")
|
|
self.temp_dirs: List[Path] = []
|
|
self.temp_files: List[Path] = []
|
|
self.original_env: Dict[str, str] = {}
|
|
self.processes: List[subprocess.Popen] = []
|
|
|
|
def setup_test_environment(self, config) -> None:
|
|
"""Setup complete test environment."""
|
|
|
|
self.logger.info("Setting up test environment...")
|
|
|
|
# Setup database
|
|
if config.test_database_url != "sqlite:///:memory:":
|
|
self._setup_test_database(config.test_database_url)
|
|
|
|
# Setup environment variables
|
|
self._setup_environment_variables(config)
|
|
|
|
# Setup temporary directories
|
|
self._setup_temp_directories()
|
|
|
|
# Setup service mocks
|
|
if config.mock_external_apis:
|
|
self._setup_service_mocks()
|
|
|
|
self.logger.info("Test environment setup complete")
|
|
|
|
def cleanup_test_environment(self) -> None:
|
|
"""Cleanup test environment."""
|
|
|
|
self.logger.info("Cleaning up test environment...")
|
|
|
|
# Stop processes
|
|
for process in self.processes:
|
|
try:
|
|
process.terminate()
|
|
process.wait(timeout=5)
|
|
except (subprocess.TimeoutExpired, ProcessLookupError):
|
|
try:
|
|
process.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
|
|
# Remove temporary files
|
|
for temp_file in self.temp_files:
|
|
try:
|
|
if temp_file.exists():
|
|
temp_file.unlink()
|
|
except Exception as e:
|
|
self.logger.warning(f"Failed to remove temp file {temp_file}: {e}")
|
|
|
|
# Remove temporary directories
|
|
for temp_dir in self.temp_dirs:
|
|
try:
|
|
if temp_dir.exists():
|
|
shutil.rmtree(temp_dir)
|
|
except Exception as e:
|
|
self.logger.warning(f"Failed to remove temp dir {temp_dir}: {e}")
|
|
|
|
# Restore environment variables
|
|
self._restore_environment_variables()
|
|
|
|
self.logger.info("Test environment cleanup complete")
|
|
|
|
def _setup_test_database(self, database_url: str) -> None:
|
|
"""Setup test database."""
|
|
|
|
if database_url.startswith("sqlite:///"):
|
|
# SQLite file database
|
|
db_path = database_url.replace("sqlite:///", "")
|
|
if db_path != ":memory:":
|
|
db_file = Path(db_path)
|
|
|
|
# Create temporary database file
|
|
if not db_file.parent.exists():
|
|
db_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Track for cleanup
|
|
self.temp_files.append(db_file)
|
|
|
|
# Set database URL environment variables
|
|
self._set_env_var("DATABASE_URL", database_url)
|
|
self._set_env_var("TEST_DATABASE_URL", database_url)
|
|
|
|
def _setup_environment_variables(self, config) -> None:
|
|
"""Setup test environment variables."""
|
|
|
|
# Core test variables
|
|
self._set_env_var("TESTING", "true")
|
|
self._set_env_var("TEST_MODE", "true")
|
|
|
|
# API mocking
|
|
if config.mock_external_apis:
|
|
self._set_env_var("MOCK_EXTERNAL_APIS", "true")
|
|
self._set_env_var("MOCK_OPENAI_API", "true")
|
|
self._set_env_var("MOCK_ANTHROPIC_API", "true")
|
|
|
|
# Timeout settings
|
|
self._set_env_var("TEST_TIMEOUT", str(config.test_timeout))
|
|
self._set_env_var("NETWORK_TIMEOUT", str(config.network_timeout))
|
|
|
|
# Custom test environment variables
|
|
for key, value in config.test_env_vars.items():
|
|
self._set_env_var(key, value)
|
|
|
|
# Security keys for testing
|
|
self._set_env_var("JWT_SECRET_KEY", "test_secret_key_for_testing_only_not_secure")
|
|
self._set_env_var("SECRET_KEY", "test_secret_key")
|
|
|
|
def _setup_temp_directories(self) -> None:
|
|
"""Setup temporary directories for tests."""
|
|
|
|
# Test data directory
|
|
test_data_dir = Path(tempfile.mkdtemp(prefix="test_data_"))
|
|
self.temp_dirs.append(test_data_dir)
|
|
self._set_env_var("TEST_DATA_DIR", str(test_data_dir))
|
|
|
|
# Test uploads directory
|
|
test_uploads_dir = Path(tempfile.mkdtemp(prefix="test_uploads_"))
|
|
self.temp_dirs.append(test_uploads_dir)
|
|
self._set_env_var("TEST_UPLOADS_DIR", str(test_uploads_dir))
|
|
|
|
# Test cache directory
|
|
test_cache_dir = Path(tempfile.mkdtemp(prefix="test_cache_"))
|
|
self.temp_dirs.append(test_cache_dir)
|
|
self._set_env_var("TEST_CACHE_DIR", str(test_cache_dir))
|
|
|
|
def _setup_service_mocks(self) -> None:
|
|
"""Setup external service mocks."""
|
|
|
|
# Mock API endpoints
|
|
mock_endpoints = {
|
|
"MOCK_ANTHROPIC_API_URL": "http://localhost:9999/mock/anthropic",
|
|
"MOCK_OPENAI_API_URL": "http://localhost:9999/mock/openai",
|
|
"MOCK_YOUTUBE_API_URL": "http://localhost:9999/mock/youtube"
|
|
}
|
|
|
|
for key, value in mock_endpoints.items():
|
|
self._set_env_var(key, value)
|
|
|
|
def _set_env_var(self, key: str, value: str) -> None:
|
|
"""Set environment variable with backup."""
|
|
|
|
# Backup original value if it exists
|
|
if key in os.environ and key not in self.original_env:
|
|
self.original_env[key] = os.environ[key]
|
|
|
|
# Set new value
|
|
os.environ[key] = value
|
|
|
|
def _restore_environment_variables(self) -> None:
|
|
"""Restore original environment variables."""
|
|
|
|
# Get all test-related environment variables
|
|
test_env_vars = [key for key in os.environ.keys() if
|
|
key.startswith(('TEST_', 'MOCK_')) or
|
|
key in ['TESTING', 'JWT_SECRET_KEY', 'SECRET_KEY']]
|
|
|
|
# Remove test environment variables
|
|
for key in test_env_vars:
|
|
if key in self.original_env:
|
|
# Restore original value
|
|
os.environ[key] = self.original_env[key]
|
|
else:
|
|
# Remove test variable
|
|
os.environ.pop(key, None)
|
|
|
|
def create_temp_file(self, content: str = "", suffix: str = "") -> Path:
|
|
"""Create a temporary file and track it for cleanup."""
|
|
|
|
temp_file = Path(tempfile.NamedTemporaryFile(
|
|
delete=False,
|
|
suffix=suffix,
|
|
mode='w',
|
|
encoding='utf-8'
|
|
).name)
|
|
|
|
if content:
|
|
temp_file.write_text(content)
|
|
|
|
self.temp_files.append(temp_file)
|
|
return temp_file
|
|
|
|
def create_temp_dir(self, prefix: str = "test_") -> Path:
|
|
"""Create a temporary directory and track it for cleanup."""
|
|
|
|
temp_dir = Path(tempfile.mkdtemp(prefix=prefix))
|
|
self.temp_dirs.append(temp_dir)
|
|
return temp_dir
|
|
|
|
def start_mock_server(self, port: int = 9999) -> subprocess.Popen:
|
|
"""Start a mock server for API testing."""
|
|
|
|
# This is a placeholder for starting mock servers
|
|
# In a real implementation, you might start a mock HTTP server
|
|
# using tools like httpretty, responses, or a custom mock server
|
|
|
|
self.logger.info(f"Mock server would start on port {port}")
|
|
|
|
# Mock process (doesn't actually start anything)
|
|
# In practice, you'd start a real mock server process
|
|
mock_process = None # subprocess.Popen(...)
|
|
|
|
if mock_process:
|
|
self.processes.append(mock_process)
|
|
|
|
return mock_process
|
|
|
|
|
|
def check_test_dependencies() -> Dict[str, bool]:
|
|
"""Check if test dependencies are available."""
|
|
|
|
dependencies = {}
|
|
|
|
# Check Python packages
|
|
required_packages = [
|
|
"pytest",
|
|
"pytest-asyncio",
|
|
"pytest-cov",
|
|
"pytest-xdist",
|
|
"pytest-timeout",
|
|
"httpx",
|
|
"fastapi",
|
|
"sqlalchemy"
|
|
]
|
|
|
|
for package in required_packages:
|
|
try:
|
|
__import__(package.replace("-", "_"))
|
|
dependencies[package] = True
|
|
except ImportError:
|
|
dependencies[package] = False
|
|
|
|
# Check for Node.js (for frontend tests)
|
|
try:
|
|
result = subprocess.run(
|
|
["node", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5
|
|
)
|
|
dependencies["nodejs"] = result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
dependencies["nodejs"] = False
|
|
|
|
# Check for npm (for frontend tests)
|
|
try:
|
|
result = subprocess.run(
|
|
["npm", "--version"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=5
|
|
)
|
|
dependencies["npm"] = result.returncode == 0
|
|
except (subprocess.TimeoutExpired, FileNotFoundError):
|
|
dependencies["npm"] = False
|
|
|
|
return dependencies
|
|
|
|
|
|
def install_test_dependencies() -> bool:
|
|
"""Install missing test dependencies."""
|
|
|
|
logger = logging.getLogger("DependencyInstaller")
|
|
|
|
# Install Python dependencies
|
|
python_deps = [
|
|
"pytest>=7.0.0",
|
|
"pytest-asyncio>=0.21.0",
|
|
"pytest-cov>=4.0.0",
|
|
"pytest-xdist>=3.0.0",
|
|
"pytest-timeout>=2.1.0",
|
|
"pytest-mock>=3.10.0"
|
|
]
|
|
|
|
try:
|
|
logger.info("Installing Python test dependencies...")
|
|
subprocess.run([
|
|
"pip", "install", "--upgrade"
|
|
] + python_deps, check=True)
|
|
|
|
logger.info("Python dependencies installed successfully")
|
|
return True
|
|
|
|
except subprocess.CalledProcessError as e:
|
|
logger.error(f"Failed to install Python dependencies: {e}")
|
|
return False
|
|
|
|
|
|
def validate_test_environment() -> List[str]:
|
|
"""Validate test environment and return list of issues."""
|
|
|
|
issues = []
|
|
|
|
# Check dependencies
|
|
dependencies = check_test_dependencies()
|
|
for dep, available in dependencies.items():
|
|
if not available:
|
|
issues.append(f"Missing dependency: {dep}")
|
|
|
|
# Check directories
|
|
required_dirs = [
|
|
Path("backend/tests"),
|
|
Path("frontend/src")
|
|
]
|
|
|
|
for dir_path in required_dirs:
|
|
if not dir_path.exists():
|
|
issues.append(f"Missing directory: {dir_path}")
|
|
|
|
# Check permissions
|
|
test_reports_dir = Path("test_reports")
|
|
try:
|
|
test_reports_dir.mkdir(exist_ok=True)
|
|
test_file = test_reports_dir / "test_write_permission"
|
|
test_file.write_text("test")
|
|
test_file.unlink()
|
|
except Exception as e:
|
|
issues.append(f"Cannot write to test reports directory: {e}")
|
|
|
|
return issues |