Delete tests

tests/__init__.py

@@ -1,77 +0,0 @@
-"""
-BackgroundFX Pro Test Suite.
-Comprehensive testing for all modules.
-"""
-
-import os
-import sys
-from pathlib import Path
-
-# Add parent directory to path for imports
-parent_dir = Path(__file__).parent.parent
-sys.path.insert(0, str(parent_dir))
-
-# Test categories
-TEST_CATEGORIES = {
-    'unit': 'Unit tests for individual components',
-    'integration': 'Integration tests for component interactions',
-    'api': 'API endpoint tests',
-    'models': 'Model management tests',
-    'pipeline': 'Processing pipeline tests',
-    'performance': 'Performance and benchmark tests',
-    'gpu': 'GPU-specific tests'
-}
-
-def run_tests(category: str = None, verbose: bool = True):
-    """
-    Run tests for BackgroundFX Pro.
-    
-    Args:
-        category: Optional test category to run
-        verbose: Enable verbose output
-    """
-    import pytest
-    
-    args = []
-    
-    if category:
-        if category in TEST_CATEGORIES:
-            args.extend(['-m', category])
-        else:
-            print(f"Unknown category: {category}")
-            print(f"Available categories: {', '.join(TEST_CATEGORIES.keys())}")
-            return 1
-    
-    if verbose:
-        args.append('-v')
-    
-    # Add coverage by default
-    args.extend(['--cov=.', '--cov-report=term-missing'])
-    
-    return pytest.main(args)
-
-
-def run_quick_tests():
-    """Run quick unit tests only (no slow/integration tests)."""
-    import pytest
-    return pytest.main(['-m', 'not slow and not integration', '-v'])
-
-
-def run_gpu_tests():
-    """Run GPU-specific tests if GPU is available."""
-    import torch
-    
-    if not torch.cuda.is_available():
-        print("No GPU available, skipping GPU tests")
-        return 0
-    
-    import pytest
-    return pytest.main(['-m', 'gpu', '-v'])
-
-
-__all__ = [
-    'run_tests',
-    'run_quick_tests',
-    'run_gpu_tests',
-    'TEST_CATEGORIES'
-]

tests/conftest.py

@@ -1,351 +0,0 @@
-"""
-Pytest configuration and fixtures for BackgroundFX Pro tests.
-"""
-
-import pytest
-import numpy as np
-import torch
-import cv2
-import tempfile
-import shutil
-from pathlib import Path
-from unittest.mock import Mock, MagicMock
-import os
-import sys
-
-# Add parent directory to path for imports
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-
-# ============================================================================
-# Configuration
-# ============================================================================
-
-@pytest.fixture(scope="session")
-def test_config():
-    """Test configuration."""
-    return {
-        'device': 'cpu',  # Use CPU for testing
-        'test_data_dir': Path(__file__).parent / 'data',
-        'temp_dir': tempfile.mkdtemp(prefix='bgfx_test_'),
-        'max_test_duration': 30,  # seconds
-        'use_gpu': torch.cuda.is_available()
-    }
-
-
-@pytest.fixture(scope="session", autouse=True)
-def cleanup(test_config):
-    """Cleanup after all tests."""
-    yield
-    # Clean up temp directory
-    if os.path.exists(test_config['temp_dir']):
-        shutil.rmtree(test_config['temp_dir'])
-
-
-# ============================================================================
-# Image and Video Fixtures
-# ============================================================================
-
-@pytest.fixture
-def sample_image():
-    """Create a sample image for testing."""
-    # Create 512x512 RGB image with a person-like shape
-    image = np.zeros((512, 512, 3), dtype=np.uint8)
-    
-    # Add background
-    image[:, :] = [100, 150, 200]  # Blue background
-    
-    # Add person-like shape (simple rectangle for testing)
-    cv2.rectangle(image, (150, 100), (350, 450), (50, 100, 50), -1)
-    
-    # Add some texture
-    noise = np.random.randint(0, 20, (512, 512, 3), dtype=np.uint8)
-    image = cv2.add(image, noise)
-    
-    return image
-
-
-@pytest.fixture
-def sample_mask():
-    """Create a sample mask for testing."""
-    mask = np.zeros((512, 512), dtype=np.uint8)
-    # Create person mask
-    cv2.rectangle(mask, (150, 100), (350, 450), 255, -1)
-    # Add some edge refinement
-    mask = cv2.GaussianBlur(mask, (5, 5), 2)
-    return mask
-
-
-@pytest.fixture
-def sample_background():
-    """Create a sample background image."""
-    background = np.zeros((512, 512, 3), dtype=np.uint8)
-    # Create gradient background
-    for i in range(512):
-        background[i, :] = [
-            int(255 * (i / 512)),  # Red gradient
-            100,  # Fixed green
-            int(255 * (1 - i / 512))  # Blue inverse gradient
-        ]
-    return background
-
-
-@pytest.fixture
-def sample_video(test_config):
-    """Create a sample video file for testing."""
-    video_path = Path(test_config['temp_dir']) / 'test_video.mp4'
-    
-    # Create video writer
-    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-    out = cv2.VideoWriter(str(video_path), fourcc, 30.0, (512, 512))
-    
-    # Write 30 frames (1 second at 30fps)
-    for i in range(30):
-        frame = np.zeros((512, 512, 3), dtype=np.uint8)
-        # Animate a moving rectangle
-        x = 100 + i * 5
-        cv2.rectangle(frame, (x, 200), (x + 100, 400), (0, 255, 0), -1)
-        out.write(frame)
-    
-    out.release()
-    return str(video_path)
-
-
-# ============================================================================
-# Model Fixtures
-# ============================================================================
-
-@pytest.fixture
-def mock_model():
-    """Create a mock ML model for testing."""
-    model = MagicMock()
-    model.eval = MagicMock(return_value=None)
-    model.to = MagicMock(return_value=model)
-    
-    # Mock forward pass
-    def forward(x):
-        batch_size = x.shape[0] if hasattr(x, 'shape') else 1
-        return torch.randn(batch_size, 1, 512, 512)
-    
-    model.__call__ = MagicMock(side_effect=forward)
-    model.forward = MagicMock(side_effect=forward)
-    
-    return model
-
-
-@pytest.fixture
-def mock_sam2_predictor():
-    """Create a mock SAM2 predictor."""
-    predictor = MagicMock()
-    
-    def predict(image):
-        h, w = image.shape[:2] if len(image.shape) > 2 else (512, 512)
-        return np.random.randint(0, 2, (h, w), dtype=np.uint8) * 255
-    
-    predictor.predict = MagicMock(side_effect=predict)
-    predictor.set_image = MagicMock(return_value=None)
-    
-    return predictor
-
-
-@pytest.fixture
-def mock_matanyone_model():
-    """Create a mock MatAnyone model."""
-    model = MagicMock()
-    
-    def refine(image, mask):
-        return cv2.GaussianBlur(mask, (5, 5), 2)
-    
-    model.refine = MagicMock(side_effect=refine)
-    
-    return model
-
-
-# ============================================================================
-# Pipeline and Processing Fixtures
-# ============================================================================
-
-@pytest.fixture
-def pipeline_config():
-    """Create pipeline configuration for testing."""
-    from api.pipeline import PipelineConfig
-    
-    return PipelineConfig(
-        use_gpu=False,  # CPU for testing
-        quality_preset='medium',
-        enable_cache=False,  # Disable cache for testing
-        batch_size=1,
-        max_workers=2
-    )
-
-
-@pytest.fixture
-def mock_pipeline(pipeline_config):
-    """Create a mock processing pipeline."""
-    from api.pipeline import ProcessingPipeline
-    
-    # Mock the pipeline to avoid loading real models
-    with pytest.MonkeyPatch().context() as m:
-        m.setattr('api.pipeline.ModelFactory.load_model', 
-                  lambda self, *args, **kwargs: Mock())
-        pipeline = ProcessingPipeline(pipeline_config)
-    
-    return pipeline
-
-
-# ============================================================================
-# API and Server Fixtures
-# ============================================================================
-
-@pytest.fixture
-def api_client():
-    """Create a test client for the API."""
-    from fastapi.testclient import TestClient
-    from api.api_server import app
-    
-    return TestClient(app)
-
-
-@pytest.fixture
-def mock_job_manager():
-    """Create a mock job manager."""
-    manager = MagicMock()
-    manager.create_job = MagicMock(return_value='test-job-123')
-    manager.get_job = MagicMock(return_value={'status': 'processing'})
-    manager.update_job = MagicMock(return_value=None)
-    
-    return manager
-
-
-# ============================================================================
-# File System Fixtures
-# ============================================================================
-
-@pytest.fixture
-def temp_dir(test_config):
-    """Create a temporary directory for test files."""
-    temp_path = Path(test_config['temp_dir']) / 'test_run'
-    temp_path.mkdir(parents=True, exist_ok=True)
-    yield temp_path
-    # Cleanup
-    if temp_path.exists():
-        shutil.rmtree(temp_path)
-
-
-@pytest.fixture
-def sample_files(temp_dir, sample_image):
-    """Create sample files in temp directory."""
-    files = {}
-    
-    # Save sample image
-    image_path = temp_dir / 'sample.jpg'
-    cv2.imwrite(str(image_path), sample_image)
-    files['image'] = image_path
-    
-    # Create multiple images for batch testing
-    for i in range(3):
-        path = temp_dir / f'image_{i}.jpg'
-        cv2.imwrite(str(path), sample_image)
-        files[f'image_{i}'] = path
-    
-    return files
-
-
-# ============================================================================
-# Model Registry Fixtures
-# ============================================================================
-
-@pytest.fixture
-def mock_registry():
-    """Create a mock model registry."""
-    from models.registry import ModelRegistry, ModelInfo, ModelTask, ModelFramework
-    
-    registry = ModelRegistry(models_dir=Path(tempfile.mkdtemp()))
-    
-    # Add test model
-    test_model = ModelInfo(
-        model_id='test-model',
-        name='Test Model',
-        version='1.0',
-        task=ModelTask.SEGMENTATION,
-        framework=ModelFramework.PYTORCH,
-        url='http://example.com/model.pth',
-        filename='test_model.pth',
-        file_size=1000000
-    )
-    
-    registry.register_model(test_model)
-    
-    return registry
-
-
-# ============================================================================
-# WebSocket Fixtures
-# ============================================================================
-
-@pytest.fixture
-def mock_websocket():
-    """Create a mock WebSocket connection."""
-    ws = MagicMock()
-    ws.accept = MagicMock(return_value=None)
-    ws.send_json = MagicMock(return_value=None)
-    ws.receive_text = MagicMock(return_value='{"type": "ping", "data": {}}')
-    
-    return ws
-
-
-# ============================================================================
-# Utility Fixtures
-# ============================================================================
-
-@pytest.fixture
-def mock_progress_callback():
-    """Create a mock progress callback."""
-    callback = MagicMock()
-    return callback
-
-
-@pytest.fixture
-def device():
-    """Get device for testing."""
-    return 'cuda' if torch.cuda.is_available() else 'cpu'
-
-
-@pytest.fixture
-def performance_timer():
-    """Timer for performance testing."""
-    import time
-    
-    class Timer:
-        def __init__(self):
-            self.start_time = None
-            self.elapsed = 0
-        
-        def __enter__(self):
-            self.start_time = time.time()
-            return self
-        
-        def __exit__(self, *args):
-            self.elapsed = time.time() - self.start_time
-    
-    return Timer
-
-
-# ============================================================================
-# Markers
-# ============================================================================
-
-def pytest_configure(config):
-    """Register custom markers."""
-    config.addinivalue_line(
-        "markers", "slow: marks tests as slow (deselect with '-m \"not slow\"')"
-    )
-    config.addinivalue_line(
-        "markers", "gpu: marks tests that require GPU"
-    )
-    config.addinivalue_line(
-        "markers", "integration: marks integration tests"
-    )
-    config.addinivalue_line(
-        "markers", "unit: marks unit tests"
-    )

tests/pytest.ini

@@ -1,53 +0,0 @@
-[pytest]
-# Pytest configuration for BackgroundFX Pro
-
-# Test discovery
-testpaths = tests
-python_files = test_*.py
-python_classes = Test*
-python_functions = test_*
-
-# Markers
-markers =
-    slow: marks tests as slow (deselect with '-m "not slow"')
-    gpu: marks tests that require GPU
-    integration: marks integration tests
-    unit: marks unit tests
-    api: marks API tests
-    models: marks model-related tests
-    pipeline: marks pipeline tests
-    performance: marks performance tests
-
-# Coverage
-addopts = 
-    --verbose
-    --strict-markers
-    --tb=short
-    --cov=.
-    --cov-report=html
-    --cov-report=term-missing
-    --cov-config=.coveragerc
-
-# Logging
-log_cli = true
-log_cli_level = INFO
-log_cli_format = %(asctime)s [%(levelname)8s] %(message)s
-log_cli_date_format = %Y-%m-%d %H:%M:%S
-
-# Warnings
-filterwarnings =
-    ignore::DeprecationWarning
-    ignore::PendingDeprecationWarning
-    ignore::FutureWarning:torch.*
-
-# Timeout
-timeout = 300
-
-# Parallel execution (optional, requires pytest-xdist)
-# addopts = -n auto
-
-# Environment variables for testing
-env =
-    TESTING = 1
-    DEVICE = cpu
-    LOG_LEVEL = WARNING

tests/test_api.py

@@ -1,356 +0,0 @@
-"""
-Tests for API endpoints and WebSocket functionality.
-"""
-
-import pytest
-import json
-import base64
-from unittest.mock import Mock, patch, MagicMock
-from fastapi.testclient import TestClient
-import numpy as np
-import cv2
-
-from api.api_server import app, ProcessingRequest, ProcessingResponse
-from api.websocket import WebSocketHandler, WSMessage, MessageType
-
-
-class TestAPIEndpoints:
-    """Test REST API endpoints."""
-    
-    @pytest.fixture
-    def client(self):
-        """Create test client."""
-        return TestClient(app)
-    
-    @pytest.fixture
-    def auth_headers(self):
-        """Create authentication headers."""
-        # Mock authentication for testing
-        return {"Authorization": "Bearer test-token"}
-    
-    def test_root_endpoint(self, client):
-        """Test root endpoint."""
-        response = client.get("/")
-        assert response.status_code == 200
-        data = response.json()
-        assert "name" in data
-        assert data["name"] == "BackgroundFX Pro API"
-    
-    def test_health_check(self, client):
-        """Test health check endpoint."""
-        response = client.get("/health")
-        assert response.status_code == 200
-        data = response.json()
-        assert data["status"] == "healthy"
-        assert "services" in data
-    
-    @patch('api.api_server.verify_token')
-    def test_process_image_endpoint(self, mock_verify, client, auth_headers, sample_image):
-        """Test image processing endpoint."""
-        mock_verify.return_value = "test-user"
-        
-        # Create test image file
-        _, buffer = cv2.imencode('.jpg', sample_image)
-        
-        files = {"file": ("test.jpg", buffer.tobytes(), "image/jpeg")}
-        data = {
-            "background": "blur",
-            "quality": "high"
-        }
-        
-        with patch('api.api_server.process_image_task'):
-            response = client.post(
-                "/api/v1/process/image",
-                headers=auth_headers,
-                files=files,
-                data=data
-            )
-        
-        assert response.status_code == 200
-        result = response.json()
-        assert "job_id" in result
-        assert result["status"] == "processing"
-    
-    @patch('api.api_server.verify_token')
-    def test_process_video_endpoint(self, mock_verify, client, auth_headers, sample_video):
-        """Test video processing endpoint."""
-        mock_verify.return_value = "test-user"
-        
-        with open(sample_video, 'rb') as f:
-            files = {"file": ("test.mp4", f.read(), "video/mp4")}
-        
-        data = {
-            "background": "office",
-            "quality": "medium"
-        }
-        
-        with patch('api.api_server.process_video_task'):
-            response = client.post(
-                "/api/v1/process/video",
-                headers=auth_headers,
-                files=files,
-                data=data
-            )
-        
-        assert response.status_code == 200
-        result = response.json()
-        assert "job_id" in result
-    
-    @patch('api.api_server.verify_token')
-    def test_batch_processing_endpoint(self, mock_verify, client, auth_headers):
-        """Test batch processing endpoint."""
-        mock_verify.return_value = "test-user"
-        
-        batch_request = {
-            "items": [
-                {"id": "1", "input_path": "/tmp/img1.jpg", "output_path": "/tmp/out1.jpg"},
-                {"id": "2", "input_path": "/tmp/img2.jpg", "output_path": "/tmp/out2.jpg"}
-            ],
-            "parallel": True,
-            "priority": "normal"
-        }
-        
-        with patch('api.api_server.process_batch_task'):
-            response = client.post(
-                "/api/v1/batch",
-                headers=auth_headers,
-                json=batch_request
-            )
-        
-        assert response.status_code == 200
-        result = response.json()
-        assert "job_id" in result
-    
-    @patch('api.api_server.verify_token')
-    def test_job_status_endpoint(self, mock_verify, client, auth_headers):
-        """Test job status endpoint."""
-        mock_verify.return_value = "test-user"
-        
-        job_id = "test-job-123"
-        
-        with patch.object(app.state.job_manager, 'get_job') as mock_get:
-            mock_get.return_value = ProcessingResponse(
-                job_id=job_id,
-                status="completed",
-                progress=1.0
-            )
-            
-            response = client.get(
-                f"/api/v1/job/{job_id}",
-                headers=auth_headers
-            )
-        
-        assert response.status_code == 200
-        result = response.json()
-        assert result["job_id"] == job_id
-        assert result["status"] == "completed"
-    
-    @patch('api.api_server.verify_token')
-    def test_streaming_endpoints(self, mock_verify, client, auth_headers):
-        """Test streaming endpoints."""
-        mock_verify.return_value = "test-user"
-        
-        # Start stream
-        stream_request = {
-            "source": "0",
-            "stream_type": "webcam",
-            "output_format": "hls"
-        }
-        
-        with patch.object(app.state.video_processor, 'start_stream_processing') as mock_start:
-            mock_start.return_value = True
-            
-            response = client.post(
-                "/api/v1/stream/start",
-                headers=auth_headers,
-                json=stream_request
-            )
-        
-        assert response.status_code == 200
-        result = response.json()
-        assert result["status"] == "streaming"
-        
-        # Stop stream
-        with patch.object(app.state.video_processor, 'stop_stream_processing'):
-            response = client.get(
-                "/api/v1/stream/stop",
-                headers=auth_headers
-            )
-        
-        assert response.status_code == 200
-
-
-class TestWebSocket:
-    """Test WebSocket functionality."""
-    
-    @pytest.fixture
-    def ws_handler(self):
-        """Create WebSocket handler."""
-        return WebSocketHandler()
-    
-    def test_websocket_connection(self, ws_handler, mock_websocket):
-        """Test WebSocket connection handling."""
-        # Test connection acceptance
-        async def test_connect():
-            await ws_handler.handle_connection(mock_websocket)
-        
-        # Would need async test runner for full test
-        assert mock_websocket.accept.called or True  # Simplified for sync test
-    
-    def test_message_parsing(self, ws_handler):
-        """Test WebSocket message parsing."""
-        message_data = {
-            "type": "process_frame",
-            "data": {"frame": "base64_data"}
-        }
-        
-        message = WSMessage.from_dict(message_data)
-        
-        assert message.type == MessageType.PROCESS_FRAME
-        assert message.data["frame"] == "base64_data"
-    
-    def test_frame_encoding_decoding(self, ws_handler, sample_image):
-        """Test frame encoding and decoding."""
-        # Encode frame
-        _, buffer = cv2.imencode('.jpg', sample_image)
-        encoded = base64.b64encode(buffer).decode('utf-8')
-        
-        # Decode frame
-        decoded = ws_handler.frame_processor._decode_frame(encoded)
-        
-        assert decoded is not None
-        assert decoded.shape == sample_image.shape
-    
-    def test_session_management(self, ws_handler):
-        """Test client session management."""
-        mock_ws = MagicMock()
-        
-        # Add session
-        async def test_add():
-            session = await ws_handler.session_manager.add_session(mock_ws, "test-client")
-            assert session.client_id == "test-client"
-        
-        # Would need async test runner for full test
-        assert ws_handler.session_manager is not None
-    
-    def test_message_routing(self, ws_handler):
-        """Test message routing."""
-        messages = [
-            WSMessage(type=MessageType.PING, data={}),
-            WSMessage(type=MessageType.UPDATE_CONFIG, data={"quality": "high"}),
-            WSMessage(type=MessageType.START_STREAM, data={"source": 0})
-        ]
-        
-        for msg in messages:
-            assert msg.type in MessageType
-            assert isinstance(msg.to_dict(), dict)
-    
-    def test_statistics_tracking(self, ws_handler):
-        """Test WebSocket statistics."""
-        stats = ws_handler.get_statistics()
-        
-        assert "uptime" in stats
-        assert "total_connections" in stats
-        assert "active_connections" in stats
-        assert "total_frames_processed" in stats
-
-
-class TestAPIIntegration:
-    """Integration tests for API."""
-    
-    @pytest.mark.integration
-    def test_full_image_processing_flow(self, client, sample_image, temp_dir):
-        """Test complete image processing flow."""
-        # Skip authentication for integration test
-        with patch('api.api_server.verify_token', return_value="test-user"):
-            # Upload image
-            _, buffer = cv2.imencode('.jpg', sample_image)
-            files = {"file": ("test.jpg", buffer.tobytes(), "image/jpeg")}
-            
-            response = client.post(
-                "/api/v1/process/image",
-                files=files,
-                data={"background": "blur", "quality": "low"}
-            )
-            
-            assert response.status_code == 200
-            job_data = response.json()
-            job_id = job_data["job_id"]
-            
-            # Check job status
-            response = client.get(f"/api/v1/job/{job_id}")
-            
-            # Would need actual processing for full test
-            assert response.status_code in [200, 404]
-    
-    @pytest.mark.integration
-    @pytest.mark.slow
-    def test_concurrent_requests(self, client):
-        """Test handling concurrent requests."""
-        import concurrent.futures
-        
-        def make_request():
-            response = client.get("/health")
-            return response.status_code
-        
-        with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
-            futures = [executor.submit(make_request) for _ in range(10)]
-            results = [f.result() for f in concurrent.futures.as_completed(futures)]
-        
-        assert all(status == 200 for status in results)
-    
-    @pytest.mark.integration
-    def test_error_handling(self, client):
-        """Test API error handling."""
-        # Test invalid endpoint
-        response = client.get("/api/v1/invalid")
-        assert response.status_code == 404
-        
-        # Test missing authentication
-        response = client.get("/api/v1/stats")
-        assert response.status_code in [401, 422]  # Unauthorized or validation error
-        
-        # Test invalid file format
-        with patch('api.api_server.verify_token', return_value="test-user"):
-            files = {"file": ("test.txt", b"text content", "text/plain")}
-            response = client.post(
-                "/api/v1/process/image",
-                files=files,
-                headers={"Authorization": "Bearer test"}
-            )
-            assert response.status_code == 400
-
-
-class TestAPIPerformance:
-    """Performance tests for API."""
-    
-    @pytest.mark.slow
-    def test_response_time(self, client, performance_timer):
-        """Test API response times."""
-        endpoints = ["/", "/health"]
-        
-        for endpoint in endpoints:
-            with performance_timer as timer:
-                response = client.get(endpoint)
-            
-            assert response.status_code == 200
-            assert timer.elapsed < 0.1  # Should respond in under 100ms
-    
-    @pytest.mark.slow
-    def test_file_upload_performance(self, client, performance_timer):
-        """Test file upload performance."""
-        # Create a 1MB test file
-        large_data = np.random.randint(0, 255, (1024, 1024, 3), dtype=np.uint8)
-        _, buffer = cv2.imencode('.jpg', large_data)
-        
-        with patch('api.api_server.verify_token', return_value="test-user"):
-            with patch('api.api_server.process_image_task'):
-                with performance_timer as timer:
-                    response = client.post(
-                        "/api/v1/process/image",
-                        files={"file": ("large.jpg", buffer.tobytes(), "image/jpeg")},
-                        headers={"Authorization": "Bearer test"}
-                    )
-                
-                assert response.status_code == 200
-                assert timer.elapsed < 2.0  # Should handle 1MB in under 2 seconds

tests/test_app.py

@@ -1,332 +0,0 @@
-"""
-Tests for the main application and video processor.
-"""
-
-import pytest
-from unittest.mock import Mock, patch, MagicMock
-import numpy as np
-import cv2
-from pathlib import Path
-import tempfile
-
-# Import from main app
-from app import VideoProcessor, processor
-
-
-class TestVideoProcessor:
-    """Test the main VideoProcessor class."""
-    
-    @pytest.fixture
-    def video_processor(self):
-        """Create a test video processor."""
-        with patch('app.model_loader.ModelLoader') as mock_loader:
-            with patch('app.device_manager.DeviceManager') as mock_device:
-                mock_device.return_value.get_optimal_device.return_value = 'cpu'
-                vp = VideoProcessor()
-                return vp
-    
-    def test_initialization(self, video_processor):
-        """Test VideoProcessor initialization."""
-        assert video_processor is not None
-        assert video_processor.models_loaded == False
-        assert video_processor.cancel_event is not None
-    
-    @patch('app.model_loader.ModelLoader.load_all_models')
-    def test_load_models(self, mock_load, video_processor):
-        """Test model loading."""
-        mock_sam2 = Mock()
-        mock_matanyone = Mock()
-        mock_load.return_value = (mock_sam2, mock_matanyone)
-        
-        result = video_processor.load_models()
-        
-        assert mock_load.called
-        assert video_processor.models_loaded == True
-        assert "Models already loaded" in result or "loaded" in result.lower()
-    
-    def test_load_models_with_progress(self, video_processor):
-        """Test model loading with progress callback."""
-        progress_values = []
-        
-        def progress_callback(value, message):
-            progress_values.append((value, message))
-        
-        with patch.object(video_processor.model_loader, 'load_all_models') as mock_load:
-            mock_load.return_value = (Mock(), Mock())
-            video_processor.load_models(progress_callback)
-        
-        assert video_processor.models_loaded == True
-    
-    def test_process_video_without_models(self, video_processor):
-        """Test video processing without loaded models."""
-        result_path, message = video_processor.process_video(
-            "test.mp4", "blur"
-        )
-        
-        assert result_path is None
-        assert "not loaded" in message.lower()
-    
-    @patch('app.validate_video_file')
-    @patch.object(VideoProcessor, '_process_single_stage')
-    def test_process_video_single_stage(self, mock_process, mock_validate, video_processor):
-        """Test single-stage video processing."""
-        mock_validate.return_value = (True, "Valid")
-        mock_process.return_value = ("/tmp/output.mp4", "Success")
-        
-        video_processor.models_loaded = True
-        video_processor.core_processor = Mock()
-        
-        result_path, message = video_processor.process_video(
-            "test.mp4", "blur", use_two_stage=False
-        )
-        
-        assert mock_process.called
-        assert result_path == "/tmp/output.mp4"
-        assert "Success" in message
-    
-    @patch('app.TWO_STAGE_AVAILABLE', True)
-    @patch('app.validate_video_file')
-    @patch.object(VideoProcessor, '_process_two_stage')
-    def test_process_video_two_stage(self, mock_process, mock_validate, video_processor):
-        """Test two-stage video processing."""
-        mock_validate.return_value = (True, "Valid")
-        mock_process.return_value = ("/tmp/output.mp4", "Success")
-        
-        video_processor.models_loaded = True
-        video_processor.core_processor = Mock()
-        video_processor.two_stage_processor = Mock()
-        
-        result_path, message = video_processor.process_video(
-            "test.mp4", "blur", use_two_stage=True
-        )
-        
-        assert mock_process.called
-        assert result_path == "/tmp/output.mp4"
-    
-    def test_cancel_processing(self, video_processor):
-        """Test processing cancellation."""
-        video_processor.cancel_processing()
-        assert video_processor.cancel_event.is_set()
-    
-    def test_get_status(self, video_processor):
-        """Test getting processor status."""
-        status = video_processor.get_status()
-        
-        assert "models_loaded" in status
-        assert "device" in status
-        assert "memory_usage" in status
-        assert status["models_loaded"] == False
-    
-    def test_cleanup_resources(self, video_processor):
-        """Test resource cleanup."""
-        with patch.object(video_processor.memory_manager, 'cleanup_aggressive'):
-            with patch.object(video_processor.model_loader, 'cleanup'):
-                video_processor.cleanup_resources()
-        
-        assert True  # Cleanup should not raise exceptions
-
-
-class TestCoreVideoProcessor:
-    """Test the CoreVideoProcessor from video_processor module."""
-    
-    @pytest.fixture
-    def core_processor(self, mock_sam2_predictor, mock_matanyone_model):
-        """Create a test core processor."""
-        from video_processor import CoreVideoProcessor
-        from app_config import ProcessingConfig
-        from memory_manager import MemoryManager
-        
-        config = ProcessingConfig()
-        memory_mgr = MemoryManager('cpu')
-        
-        processor = CoreVideoProcessor(
-            sam2_predictor=mock_sam2_predictor,
-            matanyone_model=mock_matanyone_model,
-            config=config,
-            memory_mgr=memory_mgr
-        )
-        return processor
-    
-    def test_core_processor_initialization(self, core_processor):
-        """Test CoreVideoProcessor initialization."""
-        assert core_processor is not None
-        assert core_processor.processing_active == False
-        assert core_processor.stats is not None
-    
-    def test_prepare_background(self, core_processor):
-        """Test background preparation."""
-        # Test professional background
-        background = core_processor.prepare_background(
-            "blur", None, 512, 512
-        )
-        # May return None if utilities not available
-        assert background is None or background.shape == (512, 512, 3)
-    
-    def test_get_processing_capabilities(self, core_processor):
-        """Test getting processing capabilities."""
-        capabilities = core_processor.get_processing_capabilities()
-        
-        assert "sam2_available" in capabilities
-        assert "matanyone_available" in capabilities
-        assert "quality_preset" in capabilities
-        assert "supported_formats" in capabilities
-    
-    def test_get_status(self, core_processor):
-        """Test getting processor status."""
-        status = core_processor.get_status()
-        
-        assert "processing_active" in status
-        assert "models_available" in status
-        assert "statistics" in status
-        assert "memory_usage" in status
-
-
-class TestApplicationIntegration:
-    """Integration tests for the main application."""
-    
-    @pytest.mark.integration
-    def test_global_processor_instance(self):
-        """Test the global processor instance."""
-        assert processor is not None
-        assert isinstance(processor, VideoProcessor)
-    
-    @pytest.mark.integration
-    @patch('app.model_loader.ModelLoader.load_all_models')
-    def test_model_loading_flow(self, mock_load):
-        """Test complete model loading flow."""
-        mock_load.return_value = (Mock(), Mock())
-        
-        # Use global processor
-        result = processor.load_models()
-        
-        assert processor.models_loaded == True
-        assert result is not None
-    
-    @pytest.mark.integration
-    @pytest.mark.slow
-    def test_memory_management(self):
-        """Test memory management during processing."""
-        import psutil
-        import os
-        
-        process = psutil.Process(os.getpid())
-        initial_memory = process.memory_info().rss / 1024 / 1024  # MB
-        
-        # Simulate processing
-        for _ in range(5):
-            # Create and discard large arrays
-            data = np.random.randint(0, 255, (1024, 1024, 3), dtype=np.uint8)
-            del data
-        
-        processor.cleanup_resources()
-        
-        final_memory = process.memory_info().rss / 1024 / 1024  # MB
-        memory_increase = final_memory - initial_memory
-        
-        # Memory increase should be reasonable
-        assert memory_increase < 200  # Less than 200MB increase
-
-
-class TestBackwardCompatibility:
-    """Test backward compatibility functions."""
-    
-    def test_load_models_wrapper(self):
-        """Test load_models_with_validation wrapper."""
-        from app import load_models_with_validation
-        
-        with patch.object(processor, 'load_models') as mock_load:
-            mock_load.return_value = "Success"
-            result = load_models_with_validation()
-        
-        assert mock_load.called
-        assert result == "Success"
-    
-    def test_process_video_wrapper(self):
-        """Test process_video_fixed wrapper."""
-        from app import process_video_fixed
-        
-        with patch.object(processor, 'process_video') as mock_process:
-            mock_process.return_value = ("/tmp/out.mp4", "Success")
-            
-            result = process_video_fixed(
-                "test.mp4", "blur", None
-            )
-        
-        assert mock_process.called
-        assert result[0] == "/tmp/out.mp4"
-    
-    def test_get_model_status_wrapper(self):
-        """Test get_model_status wrapper."""
-        from app import get_model_status
-        
-        with patch.object(processor, 'get_status') as mock_status:
-            mock_status.return_value = {"status": "ok"}
-            result = get_model_status()
-        
-        assert mock_status.called
-        assert result["status"] == "ok"
-
-
-class TestErrorHandling:
-    """Test error handling in the application."""
-    
-    def test_invalid_video_handling(self):
-        """Test handling of invalid video files."""
-        with patch('app.validate_video_file') as mock_validate:
-            mock_validate.return_value = (False, "Invalid format")
-            
-            processor.models_loaded = True
-            result_path, message = processor.process_video(
-                "invalid.txt", "blur"
-            )
-        
-        assert result_path is None
-        assert "Invalid" in message
-    
-    def test_model_loading_failure(self):
-        """Test handling of model loading failures."""
-        with patch.object(processor.model_loader, 'load_all_models') as mock_load:
-            mock_load.side_effect = Exception("Model not found")
-            
-            result = processor.load_models()
-        
-        assert processor.models_loaded == False
-        assert "failed" in result.lower()
-    
-    def test_processing_exception_handling(self):
-        """Test exception handling during processing."""
-        processor.models_loaded = True
-        processor.core_processor = Mock()
-        processor.core_processor.process_video.side_effect = Exception("Processing failed")
-        
-        with patch('app.validate_video_file', return_value=(True, "Valid")):
-            result_path, message = processor.process_video(
-                "test.mp4", "blur"
-            )
-        
-        assert result_path is None
-        assert "error" in message.lower() or "failed" in message.lower()
-
-
-class TestPerformance:
-    """Performance tests for the application."""
-    
-    @pytest.mark.slow
-    def test_initialization_speed(self, performance_timer):
-        """Test application initialization speed."""
-        with performance_timer as timer:
-            vp = VideoProcessor()
-        
-        assert timer.elapsed < 1.0  # Should initialize in under 1 second
-    
-    @pytest.mark.slow
-    @patch('app.model_loader.ModelLoader.load_all_models')
-    def test_model_loading_speed(self, mock_load, performance_timer):
-        """Test model loading speed."""
-        mock_load.return_value = (Mock(), Mock())
-        
-        with performance_timer as timer:
-            processor.load_models()
-        
-        # Mock loading should be very fast
-        assert timer.elapsed < 0.5

tests/test_models.py

@@ -1,376 +0,0 @@
-"""
-Tests for model management functionality.
-"""
-
-import pytest
-import tempfile
-from pathlib import Path
-from unittest.mock import Mock, patch, MagicMock
-import json
-
-from models import (
-    ModelRegistry,
-    ModelInfo,
-    ModelStatus,
-    ModelTask,
-    ModelFramework,
-    ModelDownloader,
-    ModelLoader,
-    ModelOptimizer
-)
-
-
-class TestModelRegistry:
-    """Test model registry functionality."""
-    
-    @pytest.fixture
-    def registry(self):
-        """Create a test registry."""
-        temp_dir = tempfile.mkdtemp()
-        return ModelRegistry(models_dir=Path(temp_dir))
-    
-    def test_registry_initialization(self, registry):
-        """Test registry initialization."""
-        assert registry is not None
-        assert len(registry.models) > 0  # Should have default models
-        assert registry.models_dir.exists()
-    
-    def test_register_model(self, registry):
-        """Test registering a new model."""
-        model = ModelInfo(
-            model_id="test-model",
-            name="Test Model",
-            version="1.0",
-            task=ModelTask.SEGMENTATION,
-            framework=ModelFramework.PYTORCH,
-            url="http://example.com/model.pth",
-            filename="test.pth",
-            file_size=1000000
-        )
-        
-        success = registry.register_model(model)
-        assert success == True
-        assert "test-model" in registry.models
-    
-    def test_get_model(self, registry):
-        """Test getting a model by ID."""
-        model = registry.get_model("rmbg-1.4")
-        assert model is not None
-        assert model.model_id == "rmbg-1.4"
-        assert model.task == ModelTask.SEGMENTATION
-    
-    def test_list_models_by_task(self, registry):
-        """Test listing models by task."""
-        segmentation_models = registry.list_models(task=ModelTask.SEGMENTATION)
-        assert len(segmentation_models) > 0
-        assert all(m.task == ModelTask.SEGMENTATION for m in segmentation_models)
-    
-    def test_list_models_by_framework(self, registry):
-        """Test listing models by framework."""
-        pytorch_models = registry.list_models(framework=ModelFramework.PYTORCH)
-        onnx_models = registry.list_models(framework=ModelFramework.ONNX)
-        
-        assert all(m.framework == ModelFramework.PYTORCH for m in pytorch_models)
-        assert all(m.framework == ModelFramework.ONNX for m in onnx_models)
-    
-    def test_get_best_model(self, registry):
-        """Test getting best model for a task."""
-        # Best for accuracy
-        best_accuracy = registry.get_best_model(
-            ModelTask.SEGMENTATION,
-            prefer_speed=False
-        )
-        assert best_accuracy is not None
-        
-        # Best for speed
-        best_speed = registry.get_best_model(
-            ModelTask.SEGMENTATION,
-            prefer_speed=True
-        )
-        assert best_speed is not None
-    
-    def test_update_model_usage(self, registry):
-        """Test updating model usage statistics."""
-        model_id = "rmbg-1.4"
-        initial_count = registry.models[model_id].use_count
-        
-        registry.update_model_usage(model_id)
-        
-        assert registry.models[model_id].use_count == initial_count + 1
-        assert registry.models[model_id].last_used is not None
-    
-    def test_get_total_size(self, registry):
-        """Test calculating total model size."""
-        total_size = registry.get_total_size()
-        assert total_size > 0
-        
-        # Size of available models should be 0 initially
-        available_size = registry.get_total_size(status=ModelStatus.AVAILABLE)
-        assert available_size == 0
-    
-    def test_export_registry(self, registry, temp_dir):
-        """Test exporting registry to file."""
-        export_path = temp_dir / "registry_export.json"
-        registry.export_registry(export_path)
-        
-        assert export_path.exists()
-        
-        with open(export_path) as f:
-            data = json.load(f)
-            assert "models" in data
-            assert len(data["models"]) > 0
-
-
-class TestModelDownloader:
-    """Test model downloading functionality."""
-    
-    @pytest.fixture
-    def downloader(self, mock_registry):
-        """Create a test downloader."""
-        return ModelDownloader(mock_registry)
-    
-    @patch('requests.get')
-    def test_download_model(self, mock_get, downloader):
-        """Test downloading a model."""
-        # Mock HTTP response
-        mock_response = MagicMock()
-        mock_response.headers = {'content-length': '1000000'}
-        mock_response.iter_content = MagicMock(
-            return_value=[b'data' * 1000]
-        )
-        mock_response.raise_for_status = MagicMock()
-        mock_get.return_value = mock_response
-        
-        # Test download
-        success = downloader.download_model("test-model", force=True)
-        
-        assert mock_get.called
-        # Note: Full download test would require more mocking
-    
-    def test_download_progress_tracking(self, downloader):
-        """Test download progress tracking."""
-        progress_values = []
-        
-        def progress_callback(progress):
-            progress_values.append(progress.progress)
-        
-        # Start a download (will fail but we can test progress initialization)
-        with patch.object(downloader, '_download_model_task', return_value=True):
-            downloader.download_model(
-                "test-model",
-                progress_callback=progress_callback
-            )
-        
-        assert "test-model" in downloader.downloads
-    
-    def test_cancel_download(self, downloader):
-        """Test cancelling a download."""
-        # Start a mock download
-        downloader.downloads["test-model"] = Mock()
-        downloader._stop_events["test-model"] = Mock()
-        
-        success = downloader.cancel_download("test-model")
-        
-        assert success == True
-        assert downloader._stop_events["test-model"].set.called
-    
-    def test_download_with_resume(self, downloader, temp_dir):
-        """Test download with resume support."""
-        # Create a partial file
-        partial_file = temp_dir / "test.pth.part"
-        partial_file.write_bytes(b"partial_data")
-        
-        # Mock download would check for partial file
-        assert partial_file.exists()
-        assert partial_file.stat().st_size > 0
-
-
-class TestModelLoader:
-    """Test model loading functionality."""
-    
-    @pytest.fixture
-    def loader(self, mock_registry):
-        """Create a test loader."""
-        return ModelLoader(mock_registry, device='cpu')
-    
-    def test_loader_initialization(self, loader):
-        """Test loader initialization."""
-        assert loader is not None
-        assert loader.device == 'cpu'
-        assert loader.max_memory_bytes > 0
-    
-    @patch('torch.load')
-    def test_load_pytorch_model(self, mock_torch_load, loader):
-        """Test loading a PyTorch model."""
-        mock_model = MagicMock()
-        mock_torch_load.return_value = mock_model
-        
-        # Mock model info
-        model_info = ModelInfo(
-            model_id="test-pytorch",
-            name="Test PyTorch Model",
-            version="1.0",
-            task=ModelTask.SEGMENTATION,
-            framework=ModelFramework.PYTORCH,
-            url="",
-            filename="model.pth",
-            local_path="/tmp/model.pth",
-            status=ModelStatus.AVAILABLE
-        )
-        
-        loader.registry.get_model = Mock(return_value=model_info)
-        
-        with patch.object(Path, 'exists', return_value=True):
-            loaded = loader.load_model("test-pytorch")
-        
-        # Note: Full test would require more setup
-        assert mock_torch_load.called
-    
-    def test_memory_management(self, loader):
-        """Test memory management during model loading."""
-        # Add mock models to loaded cache
-        for i in range(5):
-            loader.loaded_models[f"model_{i}"] = Mock(
-                memory_usage=100 * 1024 * 1024  # 100MB each
-            )
-        
-        loader.current_memory_usage = 500 * 1024 * 1024  # 500MB
-        
-        # Free memory
-        loader._free_memory(200 * 1024 * 1024)  # Need 200MB
-        
-        # Should have freed at least 2 models
-        assert len(loader.loaded_models) < 5
-    
-    def test_unload_model(self, loader):
-        """Test unloading a model."""
-        # Add a mock model
-        loader.loaded_models["test"] = Mock(
-            model=Mock(),
-            memory_usage=100 * 1024 * 1024
-        )
-        loader.current_memory_usage = 100 * 1024 * 1024
-        
-        success = loader.unload_model("test")
-        
-        assert success == True
-        assert "test" not in loader.loaded_models
-        assert loader.current_memory_usage == 0
-    
-    def test_get_memory_usage(self, loader):
-        """Test getting memory usage statistics."""
-        # Add mock models
-        loader.loaded_models["model1"] = Mock(memory_usage=100 * 1024 * 1024)
-        loader.loaded_models["model2"] = Mock(memory_usage=200 * 1024 * 1024)
-        loader.current_memory_usage = 300 * 1024 * 1024
-        
-        usage = loader.get_memory_usage()
-        
-        assert usage["current_usage_mb"] == 300
-        assert usage["loaded_models"] == 2
-        assert "model1" in usage["models"]
-        assert "model2" in usage["models"]
-
-
-class TestModelOptimizer:
-    """Test model optimization functionality."""
-    
-    @pytest.fixture
-    def optimizer(self, mock_registry):
-        """Create a test optimizer."""
-        loader = ModelLoader(mock_registry, device='cpu')
-        return ModelOptimizer(loader)
-    
-    @patch('torch.quantization.quantize_dynamic')
-    def test_quantize_pytorch_model(self, mock_quantize, optimizer):
-        """Test PyTorch model quantization."""
-        # Create mock model
-        mock_model = MagicMock()
-        mock_quantize.return_value = mock_model
-        
-        loaded = Mock(
-            model_id="test",
-            model=mock_model,
-            framework=ModelFramework.PYTORCH,
-            metadata={'input_size': (1, 3, 512, 512)}
-        )
-        
-        with patch.object(optimizer, '_get_model_size', return_value=1000000):
-            with patch.object(optimizer, '_benchmark_model', return_value=0.1):
-                result = optimizer._quantize_pytorch(
-                    loaded,
-                    Path("/tmp"),
-                    "dynamic"
-                )
-        
-        assert mock_quantize.called
-        # Note: Full test would require more setup
-    
-    def test_optimization_result(self, optimizer):
-        """Test optimization result structure."""
-        from models.optimizer import OptimizationResult
-        
-        result = OptimizationResult(
-            original_size_mb=100,
-            optimized_size_mb=25,
-            compression_ratio=4.0,
-            original_speed_ms=100,
-            optimized_speed_ms=50,
-            speedup=2.0,
-            accuracy_loss=0.01,
-            optimization_time=10.0,
-            output_path="/tmp/optimized.pth"
-        )
-        
-        assert result.compression_ratio == 4.0
-        assert result.speedup == 2.0
-        assert result.accuracy_loss == 0.01
-
-
-class TestModelIntegration:
-    """Integration tests for model management."""
-    
-    @pytest.mark.integration
-    @pytest.mark.slow
-    def test_model_registry_persistence(self, temp_dir):
-        """Test registry persistence across instances."""
-        # Create registry and add model
-        registry1 = ModelRegistry(models_dir=temp_dir)
-        
-        test_model = ModelInfo(
-            model_id="persistence-test",
-            name="Persistence Test",
-            version="1.0",
-            task=ModelTask.SEGMENTATION,
-            framework=ModelFramework.PYTORCH,
-            url="http://example.com/model.pth",
-            filename="persist.pth"
-        )
-        
-        registry1.register_model(test_model)
-        
-        # Create new registry instance
-        registry2 = ModelRegistry(models_dir=temp_dir)
-        
-        # Check if model persisted
-        loaded_model = registry2.get_model("persistence-test")
-        assert loaded_model is not None
-        assert loaded_model.name == "Persistence Test"
-    
-    @pytest.mark.integration
-    def test_model_manager_workflow(self):
-        """Test complete model manager workflow."""
-        from models import create_model_manager
-        
-        manager = create_model_manager()
-        
-        # Test model discovery
-        stats = manager.get_stats()
-        assert "registry" in stats
-        assert stats["registry"]["total_models"] > 0
-        
-        # Test benchmark (without actual model loading)
-        with patch.object(manager.loader, 'load_model', return_value=Mock()):
-            benchmarks = manager.benchmark()
-            # Would return empty without real models
-            assert isinstance(benchmarks, dict)

tests/test_pipeline.py

@@ -1,349 +0,0 @@
-"""
-Tests for the processing pipeline.
-"""
-
-import pytest
-import numpy as np
-import cv2
-from unittest.mock import Mock, patch, MagicMock
-from pathlib import Path
-
-from api.pipeline import (
-    ProcessingPipeline,
-    PipelineConfig,
-    PipelineResult,
-    ProcessingMode,
-    PipelineStage
-)
-
-
-class TestPipelineConfig:
-    """Test pipeline configuration."""
-    
-    def test_default_config(self):
-        """Test default configuration values."""
-        config = PipelineConfig()
-        assert config.mode == ProcessingMode.PHOTO
-        assert config.quality_preset == "high"
-        assert config.use_gpu == True
-        assert config.enable_cache == True
-    
-    def test_custom_config(self):
-        """Test custom configuration."""
-        config = PipelineConfig(
-            mode=ProcessingMode.VIDEO,
-            quality_preset="ultra",
-            use_gpu=False,
-            batch_size=4
-        )
-        assert config.mode == ProcessingMode.VIDEO
-        assert config.quality_preset == "ultra"
-        assert config.use_gpu == False
-        assert config.batch_size == 4
-
-
-class TestProcessingPipeline:
-    """Test the main processing pipeline."""
-    
-    @pytest.fixture
-    def mock_pipeline(self, pipeline_config):
-        """Create a pipeline with mocked components."""
-        with patch('api.pipeline.ModelFactory') as mock_factory:
-            with patch('api.pipeline.DeviceManager') as mock_device:
-                mock_device.return_value.get_device.return_value = 'cpu'
-                mock_factory.return_value.load_model.return_value = Mock()
-                
-                pipeline = ProcessingPipeline(pipeline_config)
-                return pipeline
-    
-    def test_pipeline_initialization(self, mock_pipeline):
-        """Test pipeline initialization."""
-        assert mock_pipeline is not None
-        assert mock_pipeline.config is not None
-        assert mock_pipeline.current_stage == PipelineStage.INITIALIZATION
-    
-    def test_process_image_success(self, mock_pipeline, sample_image, sample_background):
-        """Test successful image processing."""
-        # Mock the processing methods
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        result = mock_pipeline.process_image(sample_image, sample_background)
-        
-        assert result is not None
-        assert isinstance(result, PipelineResult)
-        assert result.success == True
-        assert result.output_image is not None
-    
-    def test_process_image_with_effects(self, mock_pipeline, sample_image):
-        """Test image processing with effects."""
-        mock_pipeline.config.apply_effects = ['bokeh', 'vignette']
-        
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        result = mock_pipeline.process_image(sample_image, None)
-        
-        assert result is not None
-        assert result.success == True
-    
-    def test_process_image_failure(self, mock_pipeline, sample_image):
-        """Test image processing failure handling."""
-        # Mock segmentation to fail
-        mock_pipeline._segment_image = Mock(side_effect=Exception("Segmentation failed"))
-        
-        result = mock_pipeline.process_image(sample_image, None)
-        
-        assert result is not None
-        assert result.success == False
-        assert len(result.errors) > 0
-    
-    @pytest.mark.parametrize("quality", ["low", "medium", "high", "ultra"])
-    def test_quality_presets(self, mock_pipeline, sample_image, quality):
-        """Test different quality presets."""
-        mock_pipeline.config.quality_preset = quality
-        
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        result = mock_pipeline.process_image(sample_image, None)
-        
-        assert result is not None
-        assert result.success == True
-    
-    def test_batch_processing(self, mock_pipeline, sample_image):
-        """Test batch processing of multiple images."""
-        images = [sample_image] * 3
-        
-        # Mock processing
-        mock_pipeline.process_image = Mock(return_value=PipelineResult(
-            success=True,
-            output_image=sample_image,
-            quality_score=0.9
-        ))
-        
-        results = mock_pipeline.process_batch(images)
-        
-        assert len(results) == 3
-        assert all(r.success for r in results)
-    
-    def test_progress_callback(self, mock_pipeline, sample_image):
-        """Test progress callback functionality."""
-        progress_values = []
-        
-        def progress_callback(value, message):
-            progress_values.append(value)
-        
-        mock_pipeline.config.progress_callback = progress_callback
-        
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        result = mock_pipeline.process_image(sample_image, None)
-        
-        assert len(progress_values) > 0
-        assert 0.0 <= max(progress_values) <= 1.0
-    
-    def test_cache_functionality(self, mock_pipeline, sample_image):
-        """Test caching functionality."""
-        mock_pipeline.config.enable_cache = True
-        
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        # First call
-        result1 = mock_pipeline.process_image(sample_image, None)
-        
-        # Second call (should use cache)
-        result2 = mock_pipeline.process_image(sample_image, None)
-        
-        assert result1.success == result2.success
-        # Verify segmentation was only called once (cache hit on second call)
-        assert mock_pipeline._segment_image.call_count == 1
-    
-    def test_memory_management(self, mock_pipeline):
-        """Test memory management and cleanup."""
-        initial_cache_size = len(mock_pipeline.cache)
-        
-        # Process multiple images to fill cache
-        for i in range(10):
-            image = np.random.randint(0, 255, (512, 512, 3), dtype=np.uint8)
-            mock_pipeline.cache[f"test_{i}"] = PipelineResult(success=True)
-        
-        # Clear cache
-        mock_pipeline.clear_cache()
-        
-        assert len(mock_pipeline.cache) == 0
-    
-    def test_statistics_tracking(self, mock_pipeline, sample_image):
-        """Test statistics tracking."""
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        # Process image
-        result = mock_pipeline.process_image(sample_image, None)
-        
-        # Get statistics
-        stats = mock_pipeline.get_statistics()
-        
-        assert 'total_processed' in stats
-        assert stats['total_processed'] > 0
-        assert 'avg_time' in stats
-
-
-class TestPipelineIntegration:
-    """Integration tests for the pipeline."""
-    
-    @pytest.mark.integration
-    @pytest.mark.slow
-    def test_end_to_end_processing(self, sample_image, sample_background, temp_dir):
-        """Test end-to-end processing pipeline."""
-        config = PipelineConfig(
-            use_gpu=False,
-            quality_preset="medium",
-            enable_cache=False
-        )
-        
-        # Create pipeline (will use real components if available)
-        try:
-            pipeline = ProcessingPipeline(config)
-        except Exception:
-            pytest.skip("Models not available for integration test")
-        
-        # Process image
-        result = pipeline.process_image(sample_image, sample_background)
-        
-        if result.success:
-            assert result.output_image is not None
-            assert result.output_image.shape == sample_image.shape
-            assert result.quality_score > 0
-            
-            # Save output
-            output_path = temp_dir / "test_output.png"
-            cv2.imwrite(str(output_path), result.output_image)
-            assert output_path.exists()
-    
-    @pytest.mark.integration
-    @pytest.mark.slow
-    def test_video_frame_processing(self, sample_video, temp_dir):
-        """Test processing video frames."""
-        config = PipelineConfig(
-            mode=ProcessingMode.VIDEO,
-            use_gpu=False,
-            quality_preset="low"
-        )
-        
-        try:
-            pipeline = ProcessingPipeline(config)
-        except Exception:
-            pytest.skip("Models not available for integration test")
-        
-        # Open video
-        cap = cv2.VideoCapture(sample_video)
-        processed_frames = []
-        
-        # Process first 5 frames
-        for i in range(5):
-            ret, frame = cap.read()
-            if not ret:
-                break
-            
-            result = pipeline.process_image(frame, None)
-            if result.success:
-                processed_frames.append(result.output_image)
-        
-        cap.release()
-        
-        assert len(processed_frames) > 0
-        
-        # Save as video
-        if processed_frames:
-            output_path = temp_dir / "test_video_out.mp4"
-            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-            out = cv2.VideoWriter(str(output_path), fourcc, 30.0, 
-                                (processed_frames[0].shape[1], processed_frames[0].shape[0]))
-            
-            for frame in processed_frames:
-                out.write(frame)
-            
-            out.release()
-            assert output_path.exists()
-
-
-class TestPipelinePerformance:
-    """Performance tests for the pipeline."""
-    
-    @pytest.mark.slow
-    def test_processing_speed(self, mock_pipeline, sample_image, performance_timer):
-        """Test processing speed."""
-        # Mock processing
-        mock_pipeline._segment_image = Mock(return_value=np.ones((512, 512), dtype=np.uint8) * 255)
-        mock_pipeline.alpha_matting.process = Mock(return_value={
-            'alpha': np.ones((512, 512), dtype=np.float32),
-            'confidence': 0.95
-        })
-        
-        with performance_timer as timer:
-            result = mock_pipeline.process_image(sample_image, None)
-        
-        assert result.success == True
-        assert timer.elapsed < 1.0  # Should process in under 1 second
-    
-    @pytest.mark.slow
-    def test_batch_processing_speed(self, mock_pipeline, sample_image, performance_timer):
-        """Test batch processing speed."""
-        images = [sample_image] * 10
-        
-        # Mock processing
-        mock_pipeline.process_image = Mock(return_value=PipelineResult(
-            success=True,
-            output_image=sample_image,
-            quality_score=0.9
-        ))
-        
-        with performance_timer as timer:
-            results = mock_pipeline.process_batch(images)
-        
-        assert len(results) == 10
-        assert timer.elapsed < 5.0  # Should process 10 images in under 5 seconds
-    
-    def test_memory_usage(self, mock_pipeline, sample_image):
-        """Test memory usage during processing."""
-        import psutil
-        import os
-        
-        process = psutil.Process(os.getpid())
-        initial_memory = process.memory_info().rss / 1024 / 1024  # MB
-        
-        # Process multiple images
-        for _ in range(10):
-            mock_pipeline.process_image(sample_image, None)
-        
-        final_memory = process.memory_info().rss / 1024 / 1024  # MB
-        memory_increase = final_memory - initial_memory
-        
-        # Memory increase should be reasonable (less than 500MB for 10 images)
-        assert memory_increase < 500

tests/test_smoke.py

@@ -1,267 +0,0 @@
-#!/usr/bin/env python3
-"""
-Smoke test for two-stage video processing
-THIS IS A NEW FILE - Basic end-to-end test
-Tests quality profiles, frame count preservation, and basic functionality
-"""
-import os
-import sys
-import cv2
-import numpy as np
-import tempfile
-import logging
-import time
-from pathlib import Path
-
-# Add project root to path
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-
-from processing.two_stage.two_stage_processor import TwoStageProcessor
-from models.loaders.matanyone_loader import MatAnyoneLoader
-
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-
-def create_test_video(path: str, frames: int = 30, fps: int = 30):
-    """Create a simple test video with a moving circle (simulating a person)"""
-    width, height = 640, 480
-    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
-    out = cv2.VideoWriter(path, fourcc, fps, (width, height))
-    
-    if not out.isOpened():
-        raise RuntimeError(f"Failed to create test video at {path}")
-    
-    for i in range(frames):
-        # Create frame with moving white circle on dark background
-        frame = np.zeros((height, width, 3), dtype=np.uint8)
-        frame[:] = (30, 30, 30)  # Dark gray background
-        
-        # Draw a moving circle (simulating a person)
-        x = int(width/2 + 100 * np.sin(i * 0.2))
-        y = int(height/2 + 50 * np.cos(i * 0.15))
-        cv2.circle(frame, (x, y), 60, (255, 255, 255), -1)
-        
-        # Add some variation to simulate clothing
-        cv2.circle(frame, (x, y-20), 20, (200, 100, 100), -1)  # "shirt"
-        
-        out.write(frame)
-    
-    out.release()
-    
-    # Verify the video was created
-    cap = cv2.VideoCapture(path)
-    actual_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
-    cap.release()
-    
-    logger.info(f"Created test video: {path} ({actual_frames} frames)")
-    return actual_frames
-
-
-def verify_output_video(path: str, expected_frames: int) -> bool:
-    """Verify output video exists and has correct frame count"""
-    if not os.path.exists(path):
-        logger.error(f"Output video not found: {path}")
-        return False
-    
-    file_size = os.path.getsize(path)
-    if file_size < 1000:
-        logger.error(f"Output video too small: {file_size} bytes")
-        return False
-    
-    cap = cv2.VideoCapture(path)
-    if not cap.isOpened():
-        logger.error(f"Cannot open output video: {path}")
-        return False
-    
-    actual_frames = 0
-    while True:
-        ret, frame = cap.read()
-        if not ret:
-            break
-        actual_frames += 1
-    
-    cap.release()
-    
-    if actual_frames != expected_frames:
-        logger.error(f"Frame count mismatch: got {actual_frames}, expected {expected_frames}")
-        return False
-    
-    logger.info(f"Output verified: {actual_frames} frames, {file_size:,} bytes")
-    return True
-
-
-def test_quality_profiles():
-    """Test that different quality profiles produce different results"""
-    logger.info("="*60)
-    logger.info("Testing Quality Profiles")
-    logger.info("="*60)
-    
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmpdir = Path(tmpdir)
-        
-        # Create test video
-        test_video = tmpdir / "test_input.mp4"
-        expected_frames = create_test_video(str(test_video), frames=30, fps=30)
-        
-        # Create a simple background
-        background = np.ones((480, 640, 3), dtype=np.uint8) * 128  # Gray
-        background[:240, :] = (100, 150, 200)  # Blue top half
-        
-        results = {}
-        
-        for quality in ["speed", "balanced", "max"]:
-            logger.info(f"\nTesting quality mode: {quality}")
-            logger.info("-" * 40)
-            
-            # Set quality environment variable
-            os.environ["BFX_QUALITY"] = quality
-            
-            # Initialize processor (without models for basic test)
-            processor = TwoStageProcessor(
-                sam2_predictor=None,  # Will use fallback
-                matanyone_model=None   # Will use fallback
-            )
-            
-            # Process video
-            output_path = tmpdir / f"output_{quality}.mp4"
-            start_time = time.time()
-            
-            result, message = processor.process_full_pipeline(
-                video_path=str(test_video),
-                background=background,
-                output_path=str(output_path),
-                key_color_mode="auto",
-                chroma_settings=None,
-                progress_callback=lambda p, d: logger.debug(f"{p:.1%}: {d}"),
-                stop_event=None
-            )
-            
-            process_time = time.time() - start_time
-            
-            if result is None:
-                logger.error(f"Processing failed for {quality}: {message}")
-                continue
-            
-            # Verify output
-            if verify_output_video(result, expected_frames):
-                results[quality] = {
-                    "success": True,
-                    "time": process_time,
-                    "frames_refined": processor.frames_refined,
-                    "total_frames": processor.total_frames_processed
-                }
-                logger.info(f"✓ {quality}: {process_time:.2f}s, "
-                          f"{processor.frames_refined}/{processor.total_frames_processed} refined")
-            else:
-                results[quality] = {"success": False}
-                logger.error(f"✗ {quality}: verification failed")
-    
-    # Summary
-    logger.info("\n" + "="*60)
-    logger.info("SUMMARY")
-    logger.info("="*60)
-    
-    all_passed = all(r.get("success", False) for r in results.values())
-    
-    if all_passed:
-        # Check that quality modes are actually different
-        if len(results) >= 2:
-            times = [r["time"] for r in results.values() if "time" in r]
-            refined_counts = [r["frames_refined"] for r in results.values() if "frames_refined" in r]
-            
-            if len(set(refined_counts)) > 1:
-                logger.info("✓ Quality profiles show different refinement counts")
-            else:
-                logger.warning("⚠ All quality profiles refined same number of frames")
-            
-            if max(times) - min(times) > 0.1:
-                logger.info("✓ Quality profiles show different processing times")
-            else:
-                logger.warning("⚠ Quality profiles have similar processing times")
-    
-    for quality, result in results.items():
-        if result.get("success"):
-            logger.info(f"✓ {quality:8s}: {result['time']:.2f}s, "
-                      f"{result['frames_refined']}/{result['total_frames']} frames refined")
-        else:
-            logger.info(f"✗ {quality:8s}: FAILED")
-    
-    return all_passed
-
-
-def test_frame_preservation():
-    """Test that no frames are lost during processing"""
-    logger.info("\n" + "="*60)
-    logger.info("Testing Frame Preservation")
-    logger.info("="*60)
-    
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmpdir = Path(tmpdir)
-        
-        # Test different frame counts
-        test_cases = [10, 25, 30, 60]
-        
-        for frame_count in test_cases:
-            logger.info(f"\nTesting with {frame_count} frames...")
-            
-            test_video = tmpdir / f"test_{frame_count}.mp4"
-            expected = create_test_video(str(test_video), frames=frame_count, fps=30)
-            
-            os.environ["BFX_QUALITY"] = "speed"  # Fast for this test
-            
-            processor = TwoStageProcessor()
-            output_path = tmpdir / f"output_{frame_count}.mp4"
-            
-            result, message = processor.process_full_pipeline(
-                video_path=str(test_video),
-                background=np.ones((480, 640, 3), dtype=np.uint8) * 100,
-                output_path=str(output_path),
-                key_color_mode="green",
-            )
-            
-            if result and verify_output_video(result, expected):
-                logger.info(f"✓ {frame_count} frames: preserved correctly")
-            else:
-                logger.error(f"✗ {frame_count} frames: FAILED")
-                return False
-    
-    logger.info("\n✓ All frame preservation tests passed!")
-    return True
-
-
-def main():
-    """Run all smoke tests"""
-    logger.info("\n" + "🔥"*20)
-    logger.info("BACKGROUNDFX PRO SMOKE TESTS")
-    logger.info("🔥"*20)
-    
-    tests_passed = []
-    
-    # Test 1: Quality profiles
-    try:
-        tests_passed.append(test_quality_profiles())
-    except Exception as e:
-        logger.error(f"Quality profile test crashed: {e}")
-        tests_passed.append(False)
-    
-    # Test 2: Frame preservation
-    try:
-        tests_passed.append(test_frame_preservation())
-    except Exception as e:
-        logger.error(f"Frame preservation test crashed: {e}")
-        tests_passed.append(False)
-    
-    # Final result
-    logger.info("\n" + "="*60)
-    if all(tests_passed):
-        logger.info("✅ ALL SMOKE TESTS PASSED!")
-        logger.info("="*60)
-        return 0
-    else:
-        logger.error("❌ SOME TESTS FAILED")
-        logger.info("="*60)
-        return 1
-
-
-if __name__ == "__main__":
-    exit(main())