224 lines
6.2 KiB
Python
224 lines
6.2 KiB
Python
"""
|
|
Global pytest fixtures and configuration for Clean Tracks tests.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import shutil
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, MagicMock
|
|
|
|
import pytest
|
|
from flask import Flask
|
|
from flask_socketio import SocketIO
|
|
|
|
# Add src directory to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / 'src'))
|
|
|
|
|
|
# Test data directory
|
|
TEST_DATA_DIR = Path(__file__).parent / 'test_data'
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def test_data_dir():
|
|
"""Provide path to test data directory."""
|
|
return TEST_DATA_DIR
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
"""Create a temporary directory for test files."""
|
|
temp_dir = tempfile.mkdtemp()
|
|
yield Path(temp_dir)
|
|
# Cleanup
|
|
shutil.rmtree(temp_dir, ignore_errors=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_audio_file(test_data_dir):
|
|
"""Provide path to sample audio file for testing."""
|
|
audio_file = test_data_dir / 'sample.mp3'
|
|
if not audio_file.exists():
|
|
# Create a minimal valid MP3 file for testing
|
|
# (In real tests, you'd have actual test audio files)
|
|
audio_file.parent.mkdir(parents=True, exist_ok=True)
|
|
audio_file.write_bytes(b'ID3') # Minimal MP3 header
|
|
return audio_file
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_whisper_model():
|
|
"""Mock Whisper model for testing without loading actual model."""
|
|
mock_model = Mock()
|
|
mock_model.transcribe = Mock(return_value={
|
|
'text': 'This is a test transcription with some bad words.',
|
|
'segments': [
|
|
{
|
|
'id': 0,
|
|
'start': 0.0,
|
|
'end': 5.0,
|
|
'text': 'This is a test transcription with some bad words.',
|
|
'words': [
|
|
{'word': 'This', 'start': 0.0, 'end': 0.5, 'confidence': 0.9},
|
|
{'word': 'is', 'start': 0.5, 'end': 0.8, 'confidence': 0.95},
|
|
{'word': 'a', 'start': 0.8, 'end': 1.0, 'confidence': 0.92},
|
|
{'word': 'test', 'start': 1.0, 'end': 1.5, 'confidence': 0.88},
|
|
]
|
|
}
|
|
]
|
|
})
|
|
return mock_model
|
|
|
|
|
|
@pytest.fixture
|
|
def app():
|
|
"""Create Flask application for testing."""
|
|
from api import create_app
|
|
|
|
test_config = {
|
|
'TESTING': True,
|
|
'SECRET_KEY': 'test-secret-key',
|
|
'DATABASE_URL': 'sqlite:///:memory:',
|
|
'UPLOAD_FOLDER': '/tmp/test-uploads',
|
|
'MAX_CONTENT_LENGTH': 10 * 1024 * 1024, # 10MB for tests
|
|
'CORS_ORIGINS': '*'
|
|
}
|
|
|
|
app, socketio = create_app(test_config)
|
|
|
|
# Create application context
|
|
with app.app_context():
|
|
yield app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(app):
|
|
"""Create Flask test client."""
|
|
return app.test_client()
|
|
|
|
|
|
@pytest.fixture
|
|
def socketio_client(app):
|
|
"""Create Socket.IO test client."""
|
|
socketio = app.socketio
|
|
return socketio.test_client(app)
|
|
|
|
|
|
@pytest.fixture
|
|
def runner(app):
|
|
"""Create Flask CLI runner for testing CLI commands."""
|
|
return app.test_cli_runner()
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_audio_processor():
|
|
"""Mock AudioProcessor for testing."""
|
|
mock_processor = Mock()
|
|
mock_processor.process_file = Mock(return_value={
|
|
'words_detected': 5,
|
|
'words_censored': 5,
|
|
'audio_duration': 30.0,
|
|
'output_file': 'output.mp3'
|
|
})
|
|
return mock_processor
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_word_list():
|
|
"""Mock word list for testing."""
|
|
return [
|
|
{'word': 'badword1', 'severity': 'high', 'category': 'profanity'},
|
|
{'word': 'badword2', 'severity': 'medium', 'category': 'inappropriate'},
|
|
{'word': 'badword3', 'severity': 'low', 'category': 'slang'}
|
|
]
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_job_manager():
|
|
"""Mock JobManager for WebSocket testing."""
|
|
from api.websocket_enhanced import JobManager, JobMetrics
|
|
|
|
manager = JobManager()
|
|
job_id = manager.create_job()
|
|
return manager, job_id
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_websocket_emitter():
|
|
"""Mock WebSocket event emitter."""
|
|
emitter = Mock()
|
|
emitter.emit_progress = Mock()
|
|
emitter.emit_completed = Mock()
|
|
emitter.emit_error = Mock()
|
|
return emitter
|
|
|
|
|
|
# Pytest plugins and hooks
|
|
|
|
def pytest_configure(config):
|
|
"""Configure pytest with custom settings."""
|
|
# Set environment variables for testing
|
|
os.environ['TESTING'] = 'true'
|
|
os.environ['DEBUG'] = 'false'
|
|
|
|
# Create test data directory if it doesn't exist
|
|
TEST_DATA_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
def pytest_unconfigure(config):
|
|
"""Clean up after pytest."""
|
|
# Remove test environment variables
|
|
os.environ.pop('TESTING', None)
|
|
|
|
|
|
def pytest_collection_modifyitems(config, items):
|
|
"""Modify test collection to add markers."""
|
|
for item in items:
|
|
# Add markers based on test location
|
|
if 'unit' in str(item.fspath):
|
|
item.add_marker(pytest.mark.unit)
|
|
elif 'integration' in str(item.fspath):
|
|
item.add_marker(pytest.mark.integration)
|
|
elif 'e2e' in str(item.fspath):
|
|
item.add_marker(pytest.mark.e2e)
|
|
|
|
# Add markers based on test name
|
|
if 'test_cli' in item.name:
|
|
item.add_marker(pytest.mark.cli)
|
|
elif 'test_websocket' in item.name:
|
|
item.add_marker(pytest.mark.websocket)
|
|
elif 'test_security' in item.name:
|
|
item.add_marker(pytest.mark.security)
|
|
elif 'test_performance' in item.name:
|
|
item.add_marker(pytest.mark.performance)
|
|
|
|
|
|
# Helper functions for tests
|
|
|
|
def create_test_audio_file(path: Path, duration: float = 10.0):
|
|
"""Create a test audio file."""
|
|
# In real implementation, this would create an actual audio file
|
|
# For now, just create a dummy file
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(b'RIFF' + b'\x00' * 100) # Minimal WAV header
|
|
return path
|
|
|
|
|
|
def create_test_word_list_file(path: Path, words: list):
|
|
"""Create a test word list CSV file."""
|
|
import csv
|
|
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(path, 'w', newline='') as f:
|
|
writer = csv.DictWriter(f, fieldnames=['word', 'severity', 'category'])
|
|
writer.writeheader()
|
|
writer.writerows(words)
|
|
return path
|
|
|
|
|
|
class AsyncMock(MagicMock):
|
|
"""Mock for async functions."""
|
|
async def __call__(self, *args, **kwargs):
|
|
return super(AsyncMock, self).__call__(*args, **kwargs) |