Create tests/test_app.py

tests/test_app.py

@@ -0,0 +1,332 @@
+"""
+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