390 lines
13 KiB
Python
390 lines
13 KiB
Python
"""
|
|
Tests for FFmpeg assembler module.
|
|
"""
|
|
|
|
import pytest
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock
|
|
from src.assembly.assembler import (
|
|
FFmpegAssembler,
|
|
AssemblyConfig,
|
|
AssemblyResult,
|
|
TransitionType
|
|
)
|
|
|
|
|
|
class TestAssemblyConfig:
|
|
"""Test AssemblyConfig dataclass."""
|
|
|
|
def test_default_config(self):
|
|
"""Test default configuration values."""
|
|
config = AssemblyConfig()
|
|
assert config.fps == 24
|
|
assert config.container == "mp4"
|
|
assert config.codec == "h264"
|
|
assert config.crf == 18
|
|
assert config.preset == "medium"
|
|
assert config.transition == TransitionType.NONE
|
|
assert config.transition_duration_ms == 500
|
|
assert config.add_shot_labels is False
|
|
assert config.audio_track is None
|
|
|
|
def test_config_to_dict(self):
|
|
"""Test configuration serialization."""
|
|
config = AssemblyConfig(
|
|
fps=30,
|
|
codec="h265",
|
|
transition=TransitionType.FADE
|
|
)
|
|
d = config.to_dict()
|
|
assert d["fps"] == 30
|
|
assert d["codec"] == "h265"
|
|
assert d["transition"] == "fade"
|
|
assert d["audio_track"] is None
|
|
|
|
def test_config_with_audio(self):
|
|
"""Test configuration with audio track."""
|
|
audio_path = Path("/path/to/audio.mp3")
|
|
config = AssemblyConfig(audio_track=audio_path)
|
|
d = config.to_dict()
|
|
assert d["audio_track"] == str(audio_path)
|
|
|
|
|
|
class TestFFmpegAssemblerInit:
|
|
"""Test FFmpegAssembler initialization."""
|
|
|
|
def test_default_init(self):
|
|
"""Test default initialization."""
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
assembler = FFmpegAssembler()
|
|
assert assembler.ffmpeg_path == "ffmpeg"
|
|
|
|
def test_custom_ffmpeg_path(self):
|
|
"""Test initialization with custom ffmpeg path."""
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
assembler = FFmpegAssembler(ffmpeg_path="/usr/bin/ffmpeg")
|
|
assert assembler.ffmpeg_path == "/usr/bin/ffmpeg"
|
|
|
|
def test_ffmpeg_not_found(self):
|
|
"""Test behavior when ffmpeg is not available."""
|
|
with patch('subprocess.run', side_effect=FileNotFoundError()):
|
|
assembler = FFmpegAssembler()
|
|
# Should not raise, just have _check_ffmpeg return False
|
|
assert assembler._check_ffmpeg() is False
|
|
|
|
|
|
class TestFFmpegAssemblerAssemble:
|
|
"""Test video assembly functionality."""
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess(self):
|
|
"""Fixture to mock subprocess.run."""
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
yield mock_run
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Fixture for temporary directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
def test_assemble_no_files(self, mock_subprocess):
|
|
"""Test assembly with no input files."""
|
|
assembler = FFmpegAssembler()
|
|
result = assembler.assemble([], Path("output.mp4"))
|
|
|
|
assert result.success is False
|
|
assert "No shot files provided" in result.error_message
|
|
|
|
def test_assemble_missing_files(self, mock_subprocess, temp_dir):
|
|
"""Test assembly with missing input files."""
|
|
assembler = FFmpegAssembler()
|
|
missing_file = temp_dir / "nonexistent.mp4"
|
|
result = assembler.assemble([missing_file], temp_dir / "output.mp4")
|
|
|
|
assert result.success is False
|
|
assert "Missing shot files" in result.error_message
|
|
|
|
def test_assemble_single_file(self, mock_subprocess, temp_dir):
|
|
"""Test simple concatenation with single file."""
|
|
# Create dummy input file
|
|
input_file = temp_dir / "shot_001.mp4"
|
|
input_file.write_bytes(b"dummy video data")
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
assembler = FFmpegAssembler()
|
|
|
|
# Mock ffprobe for duration check
|
|
with patch.object(assembler, '_get_video_duration', return_value=4.0):
|
|
result = assembler.assemble([input_file], output_file)
|
|
|
|
assert result.success is True
|
|
assert result.output_path == output_file
|
|
assert result.num_shots == 1
|
|
|
|
# Verify ffmpeg was called
|
|
assert mock_subprocess.called
|
|
call_args = mock_subprocess.call_args[0][0]
|
|
assert "ffmpeg" in call_args[0]
|
|
assert "-f" in call_args
|
|
assert "concat" in call_args
|
|
|
|
def test_assemble_multiple_files(self, mock_subprocess, temp_dir):
|
|
"""Test concatenation with multiple files."""
|
|
# Create dummy input files
|
|
input_files = [
|
|
temp_dir / "shot_001.mp4",
|
|
temp_dir / "shot_002.mp4",
|
|
temp_dir / "shot_003.mp4"
|
|
]
|
|
for f in input_files:
|
|
f.write_bytes(b"dummy video data")
|
|
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch.object(assembler, '_get_video_duration', return_value=4.0):
|
|
result = assembler.assemble(input_files, output_file)
|
|
|
|
assert result.success is True
|
|
assert result.num_shots == 3
|
|
assert result.output_path == output_file
|
|
|
|
def test_assemble_with_config(self, mock_subprocess, temp_dir):
|
|
"""Test assembly with custom configuration."""
|
|
input_file = temp_dir / "shot_001.mp4"
|
|
input_file.write_bytes(b"dummy video data")
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
config = AssemblyConfig(
|
|
fps=30,
|
|
codec="h265",
|
|
crf=20,
|
|
preset="slow"
|
|
)
|
|
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch.object(assembler, '_get_video_duration', return_value=4.0):
|
|
result = assembler.assemble([input_file], output_file, config)
|
|
|
|
assert result.success is True
|
|
|
|
# Verify config was used
|
|
call_args = mock_subprocess.call_args[0][0]
|
|
assert "-r" in call_args
|
|
assert "30" in call_args
|
|
assert "libx265" in call_args
|
|
assert "-crf" in call_args
|
|
assert "20" in call_args
|
|
assert "slow" in call_args
|
|
|
|
def test_assemble_ffmpeg_error(self, mock_subprocess, temp_dir):
|
|
"""Test handling of ffmpeg error."""
|
|
input_file = temp_dir / "shot_001.mp4"
|
|
input_file.write_bytes(b"dummy video data")
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
# Make ffmpeg return error
|
|
mock_subprocess.return_value = MagicMock(
|
|
returncode=1,
|
|
stderr="Error: Invalid data"
|
|
)
|
|
|
|
assembler = FFmpegAssembler()
|
|
result = assembler.assemble([input_file], output_file)
|
|
|
|
assert result.success is False
|
|
assert "FFmpeg error" in result.error_message
|
|
|
|
|
|
class TestFFmpegAssemblerTransitions:
|
|
"""Test transition functionality."""
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess(self):
|
|
"""Fixture to mock subprocess.run."""
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
yield mock_run
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Fixture for temporary directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
def test_fade_transition(self, mock_subprocess, temp_dir):
|
|
"""Test assembly with fade transition."""
|
|
input_files = [
|
|
temp_dir / "shot_001.mp4",
|
|
temp_dir / "shot_002.mp4"
|
|
]
|
|
for f in input_files:
|
|
f.write_bytes(b"dummy video data")
|
|
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
config = AssemblyConfig(
|
|
transition=TransitionType.FADE,
|
|
transition_duration_ms=500
|
|
)
|
|
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch.object(assembler, '_get_video_duration', return_value=4.0):
|
|
result = assembler.assemble(input_files, output_file, config)
|
|
|
|
assert result.success is True
|
|
|
|
# Verify filter_complex was used
|
|
call_args = mock_subprocess.call_args[0][0]
|
|
assert "-filter_complex" in call_args
|
|
assert "xfade" in str(call_args)
|
|
|
|
|
|
class TestFFmpegAssemblerAudio:
|
|
"""Test audio functionality."""
|
|
|
|
@pytest.fixture
|
|
def mock_subprocess(self):
|
|
"""Fixture to mock subprocess.run."""
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0, stderr="")
|
|
yield mock_run
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Fixture for temporary directory."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
def test_add_audio(self, mock_subprocess, temp_dir):
|
|
"""Test adding audio track."""
|
|
input_file = temp_dir / "shot_001.mp4"
|
|
input_file.write_bytes(b"dummy video data")
|
|
audio_file = temp_dir / "audio.mp3"
|
|
audio_file.write_bytes(b"dummy audio data")
|
|
output_file = temp_dir / "output.mp4"
|
|
|
|
config = AssemblyConfig(audio_track=audio_file)
|
|
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch.object(assembler, '_get_video_duration', return_value=4.0):
|
|
result = assembler.assemble([input_file], output_file, config)
|
|
|
|
assert result.success is True
|
|
|
|
# Verify audio was added (second ffmpeg call)
|
|
calls = mock_subprocess.call_args_list
|
|
# First call is for video, second should be for audio
|
|
assert len(calls) >= 2
|
|
|
|
|
|
class TestFFmpegAssemblerUtilities:
|
|
"""Test utility methods."""
|
|
|
|
def test_get_video_codec(self):
|
|
"""Test codec name mapping."""
|
|
assembler = FFmpegAssembler()
|
|
|
|
assert assembler._get_video_codec("h264") == "libx264"
|
|
assert assembler._get_video_codec("h265") == "libx265"
|
|
assert assembler._get_video_codec("vp9") == "libvpx-vp9"
|
|
assert assembler._get_video_codec("unknown") == "libx264" # Default
|
|
|
|
def test_get_video_duration(self):
|
|
"""Test duration extraction."""
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(
|
|
returncode=0,
|
|
stdout="4.500\n"
|
|
)
|
|
duration = assembler._get_video_duration(Path("test.mp4"))
|
|
assert duration == 4.5
|
|
|
|
def test_get_video_duration_error(self):
|
|
"""Test duration extraction with error."""
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=1)
|
|
duration = assembler._get_video_duration(Path("test.mp4"))
|
|
assert duration == 0.0
|
|
|
|
def test_extract_frame(self):
|
|
"""Test frame extraction."""
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = assembler.extract_frame(
|
|
Path("video.mp4"),
|
|
2.5,
|
|
Path("frame.jpg")
|
|
)
|
|
assert result is True
|
|
|
|
# Verify ffmpeg command
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "-ss" in call_args
|
|
assert "2.5" in call_args
|
|
assert "-vframes" in call_args
|
|
assert "1" in call_args
|
|
|
|
def test_burn_in_labels(self):
|
|
"""Test label burn-in."""
|
|
assembler = FFmpegAssembler()
|
|
|
|
with patch('subprocess.run') as mock_run:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
result = assembler.burn_in_labels(
|
|
Path("input.mp4"),
|
|
Path("output.mp4"),
|
|
["Shot 1", "Shot 2"],
|
|
font_size=24,
|
|
position="top-left"
|
|
)
|
|
assert result.success is True
|
|
|
|
# Verify drawtext filter
|
|
call_args = mock_run.call_args[0][0]
|
|
assert "-vf" in call_args
|
|
assert "drawtext" in str(call_args)
|
|
|
|
|
|
class TestAssemblyResult:
|
|
"""Test AssemblyResult dataclass."""
|
|
|
|
def test_success_result(self):
|
|
"""Test successful result."""
|
|
result = AssemblyResult(
|
|
success=True,
|
|
output_path=Path("output.mp4"),
|
|
duration_s=12.5,
|
|
num_shots=3
|
|
)
|
|
assert result.success is True
|
|
assert result.output_path == Path("output.mp4")
|
|
assert result.duration_s == 12.5
|
|
assert result.num_shots == 3
|
|
assert result.metadata == {}
|
|
|
|
def test_failure_result(self):
|
|
"""Test failure result."""
|
|
result = AssemblyResult(
|
|
success=False,
|
|
error_message="FFmpeg failed"
|
|
)
|
|
assert result.success is False
|
|
assert result.error_message == "FFmpeg failed"
|
|
assert result.output_path is None
|