test_ffmpeg_server.py

import unittest
import os
import sys
import tempfile
import shutil
from unittest.mock import patch, MagicMock

# Add the parent directory to sys.path to import the ffmpeg_server module
sys.path.insert(0, os.path.abspath(os.path.dirname(__file__)))

# Import the module to test
import ffmpeg_server

class TestFFmpegServer(unittest.TestCase):
    
    def setUp(self):
        # Create a temporary directory for testing
        self.test_dir = tempfile.mkdtemp()
        # Save the original working directory
        self.original_working_dir = ffmpeg_server.WORKING_DIR
        # Set the working directory to our test directory
        ffmpeg_server.WORKING_DIR = self.test_dir
        # Create a test video file
        self.test_video = os.path.join(self.test_dir, "test_video.mp4")
        with open(self.test_video, "w") as f:
            f.write("dummy video content")
        # Create a test image file for watermark
        self.test_image = os.path.join(self.test_dir, "test_watermark.png")
        with open(self.test_image, "w") as f:
            f.write("dummy image content")
    
    def tearDown(self):
        # Restore the original working directory
        ffmpeg_server.WORKING_DIR = self.original_working_dir
        # Remove the temporary directory
        shutil.rmtree(self.test_dir)
    
    def test_is_safe_path(self):
        # Test path within working directory
        self.assertTrue(ffmpeg_server.is_safe_path(self.test_dir, self.test_video))
        # Test path outside working directory
        outside_path = os.path.join(os.path.dirname(self.test_dir), "outside.txt")
        self.assertFalse(ffmpeg_server.is_safe_path(self.test_dir, outside_path))
    
    @patch('subprocess.check_output')
    @patch('subprocess.run')
    def test_run_ffmpeg_command_success(self, mock_run, mock_check_output):
        # Mock the ffmpeg path
        mock_check_output.return_value = b'/usr/bin/ffmpeg\n'
        # Mock successful subprocess run
        mock_process = MagicMock()
        mock_process.stdout = "Success output"
        mock_process.stderr = ""
        mock_run.return_value = mock_process
        
        result = ffmpeg_server.run_ffmpeg_command(["ffmpeg", "-version"])
        
        self.assertTrue(result["success"])
        self.assertEqual(result["output"], "Success output")
        self.assertIsNone(result["error"])
    
    @patch('subprocess.check_output')
    @patch('subprocess.run')
    def test_run_ffmpeg_command_error(self, mock_run, mock_check_output):
        # Mock the ffmpeg path
        mock_check_output.return_value = b'/usr/bin/ffmpeg\n'
        # Mock subprocess run with error
        mock_run.side_effect = ffmpeg_server.subprocess.CalledProcessError(1, [], stderr="Error output")
        
        result = ffmpeg_server.run_ffmpeg_command(["ffmpeg", "-invalid"])
        
        self.assertFalse(result["success"])
        self.assertIsNone(result["output"])
        self.assertEqual(result["error"], "Error output")
    
    @patch('ffmpeg_server.run_ffmpeg_command')
    def test_trim_video_success(self, mock_run_command):
        # Mock successful ffmpeg command
        mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
        
        # Get relative paths for the test
        rel_input = os.path.basename(self.test_video)
        rel_output = "trimmed_video.mp4"
        
        result = ffmpeg_server.trim_video(rel_input, 10.0, 30.0, rel_output)
        
        self.assertTrue(result["success"])
        self.assertTrue("trimmed" in result["output"])
        # Verify the command was called with correct parameters
        args, _ = mock_run_command.call_args
        command = args[0]
        self.assertEqual(command[0], "ffmpeg")
        self.assertEqual(command[2], os.path.join(self.test_dir, rel_input))
        self.assertEqual(command[4], "10.0")
        self.assertEqual(command[6], "30.0")
    
    def test_trim_video_invalid_params(self):
        # Test with negative start time
        result = ffmpeg_server.trim_video(os.path.basename(self.test_video), -1.0, 30.0, "output.mp4")
        self.assertFalse(result["success"])
        self.assertIn("Invalid time parameters", result["error"])
        
        # Test with zero duration
        result = ffmpeg_server.trim_video(os.path.basename(self.test_video), 10.0, 0.0, "output.mp4")
        self.assertFalse(result["success"])
        self.assertIn("Invalid time parameters", result["error"])
    
    def test_trim_video_nonexistent_file(self):
        result = ffmpeg_server.trim_video("nonexistent.mp4", 10.0, 30.0, "output.mp4")
        self.assertFalse(result["success"])
        self.assertIn("does not exist", result["error"])
    
    @patch('ffmpeg_server.run_ffmpeg_command')
    def test_convert_video_format_success(self, mock_run_command):
        # Mock successful ffmpeg command
        mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
        
        # Get relative paths for the test
        rel_input = os.path.basename(self.test_video)
        rel_output = "converted_video"
        
        result = ffmpeg_server.convert_video_format(rel_input, "avi", rel_output)
        
        self.assertTrue(result["success"])
        self.assertTrue("converted" in result["output"])
        # Verify the command was called with correct parameters
        args, _ = mock_run_command.call_args
        command = args[0]
        self.assertEqual(command[0], "ffmpeg")
        self.assertEqual(command[2], os.path.join(self.test_dir, rel_input))
        self.assertEqual(command[4], "avi")
        self.assertTrue(command[-1].endswith(".avi"))
    
    @patch('ffmpeg_server.run_ffmpeg_command')
    def test_extract_audio_success(self, mock_run_command):
        # Mock successful ffmpeg command
        mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
        
        # Get relative paths for the test
        rel_input = os.path.basename(self.test_video)
        rel_output = "extracted_audio"
        
        result = ffmpeg_server.extract_audio(rel_input, "mp3", rel_output)
        
        self.assertTrue(result["success"])
        self.assertTrue("extracted" in result["output"])
        # Verify the command was called with correct parameters
        args, _ = mock_run_command.call_args
        command = args[0]
        self.assertEqual(command[0], "ffmpeg")
        self.assertEqual(command[2], os.path.join(self.test_dir, rel_input))
        self.assertEqual(command[4], "-acodec")
        self.assertTrue(command[-1].endswith(".mp3"))
    
    @patch('ffmpeg_server.run_ffmpeg_command')
    def test_add_watermark_success(self, mock_run_command):
        # Mock successful ffmpeg command
        mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
        
        # Get relative paths for the test
        rel_input = os.path.basename(self.test_video)
        rel_watermark = os.path.basename(self.test_image)
        rel_output = "watermarked_video.mp4"
        
        result = ffmpeg_server.add_watermark(rel_input, rel_watermark, "bottomright", rel_output)
        
        self.assertTrue(result["success"])
        self.assertTrue("watermark" in result["output"])
        # Verify the command was called with correct parameters
        args, _ = mock_run_command.call_args
        command = args[0]
        self.assertEqual(command[0], "ffmpeg")
        self.assertEqual(command[2], os.path.join(self.test_dir, rel_input))
        self.assertEqual(command[4], os.path.join(self.test_dir, rel_watermark))
        self.assertIn("overlay", command[6])
    
    def test_add_watermark_nonexistent_watermark(self):
        result = ffmpeg_server.add_watermark(
            os.path.basename(self.test_video), 
            "nonexistent.png", 
            "center", 
            "output.mp4"
        )
        self.assertFalse(result["success"])
        self.assertIn("Watermark file", result["error"])
    
    @patch('ffmpeg_server.run_ffmpeg_command')
    def test_adjust_video_quality_success(self, mock_run_command):
        # Mock successful ffmpeg command
        mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
        
        # Get relative paths for the test
        rel_input = os.path.basename(self.test_video)
        rel_output = "high_quality_video.mp4"
        
        result = ffmpeg_server.adjust_video_quality(rel_input, "high", rel_output)
        
        self.assertTrue(result["success"])
        self.assertTrue("quality" in result["output"])
        # Verify the command was called with correct parameters
        args, _ = mock_run_command.call_args
        command = args[0]
        self.assertEqual(command[0], "ffmpeg")
        self.assertEqual(command[2], os.path.join(self.test_dir, rel_input))
        self.assertEqual(command[3], "-crf")
        self.assertEqual(command[4], "18")
    
    def test_adjust_video_quality_invalid_quality(self):
        # Test with invalid quality - should use default
        with patch('ffmpeg_server.run_ffmpeg_command') as mock_run_command:
            mock_run_command.return_value = {"success": True, "output": "ffmpeg output", "error": None}
            
            result = ffmpeg_server.adjust_video_quality(
                os.path.basename(self.test_video), 
                "invalid_quality", 
                "output.mp4"
            )
            
            # Should still succeed but use default quality
            self.assertTrue(result["success"])
            args, _ = mock_run_command.call_args
            command = args[0]
            self.assertEqual(command[4], "23")  # Default medium quality CRF
    
    @patch('glob.glob')
    def test_batch_process_success(self, mock_glob):
        # Mock glob to return our test video
        mock_glob.return_value = [self.test_video]
        
        # Mock the trim_video function
        with patch('ffmpeg_server.trim_video') as mock_trim:
            mock_trim.return_value = {"success": True, "output": "Video trimmed", "error": None}
            
            result = ffmpeg_server.batch_process(
                "*.mp4",
                "trim",
                {"start_time": 10.0, "duration": 30.0}
            )
            
            self.assertTrue(result["success"])
            self.assertEqual(len(result["details"]), 1)
            self.assertTrue(mock_trim.called)
    
    def test_batch_process_no_matching_files(self):
        # Test with a pattern that doesn't match any files
        result = ffmpeg_server.batch_process("nomatch*.mp4", "trim", {"start_time": 10.0, "duration": 30.0})
        self.assertFalse(result["success"])
        self.assertIn("No files match pattern", result["error"])
    
    def test_batch_process_invalid_operation(self):
        # Create a test file that will match the pattern
        with open(os.path.join(self.test_dir, "test.mp4"), "w") as f:
            f.write("dummy content")
            
        result = ffmpeg_server.batch_process("*.mp4", "invalid_op", {})
        self.assertFalse(result["success"])
        self.assertIn("Unknown operation", result["error"])
    
    def test_list_media_files(self):
        # Create additional test media files
        media_files = [
            "test1.mp4", "test2.avi", "test3.mp3", "test4.jpg"
        ]
        for filename in media_files:
            with open(os.path.join(self.test_dir, filename), "w") as f:
                f.write("dummy content")
        
        # Test the function
        with patch('glob.glob') as mock_glob:
            # Make glob return our test files with full paths
            mock_glob.side_effect = lambda pattern: [
                os.path.join(self.test_dir, f) for f in media_files 
                if f.endswith(pattern[pattern.rindex("*")+1:])
            ]
            
            result = ffmpeg_server.list_media_files()
            
            self.assertTrue(result["success"])
            self.assertEqual(len(result["files"]), len(media_files))
            for filename in media_files:
                self.assertIn(filename, result["files"])

if __name__ == "__main__":
    unittest.main()