185 lines
7.3 KiB
Python
185 lines
7.3 KiB
Python
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.tests.helpers import make_test_temp_dir
|
|
|
|
|
|
class TestFFProbe:
|
|
"""Test the ffprobe module functionality."""
|
|
|
|
def setup_method(self):
|
|
"""Set up test files."""
|
|
self.temp_dir = str(make_test_temp_dir("ffprobe"))
|
|
self.test_file = Path(self.temp_dir) / "test_video.mp4"
|
|
self.test_file.touch()
|
|
|
|
def teardown_method(self):
|
|
"""Clean up test files."""
|
|
import shutil
|
|
|
|
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffprobe_with_existing_file(self):
|
|
"""Test ffprobe with an existing file."""
|
|
from app.features.streaming.library.ffprobe import FFProbeResult, ffprobe
|
|
|
|
# Mock subprocess to avoid actual ffprobe execution
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
# Mock the subprocess result
|
|
mock_process = AsyncMock()
|
|
mock_process.wait.return_value = 0
|
|
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"")
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
with patch("anyio.open_file") as mock_open_file:
|
|
mock_file = AsyncMock()
|
|
mock_open_file.return_value.__aenter__.return_value = mock_file
|
|
|
|
result = await ffprobe(str(self.test_file))
|
|
assert isinstance(result, FFProbeResult)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffprobe_with_nonexistent_file(self):
|
|
"""Test ffprobe with a non-existent file."""
|
|
from app.features.streaming.library.ffprobe import ffprobe
|
|
|
|
nonexistent_file = Path(self.temp_dir) / "does_not_exist.mp4"
|
|
|
|
with pytest.raises(OSError, match="No such media file"):
|
|
await ffprobe(str(nonexistent_file))
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffprobe_caching_behavior(self):
|
|
"""Test that ffprobe results are cached with enhanced async timed_lru_cache."""
|
|
from app.features.streaming.library.ffprobe import ffprobe
|
|
|
|
assert hasattr(ffprobe, "cache_clear"), (
|
|
"Test that the function has been decorated with caching - ffprobe should have cache_clear method from timed_lru_cache"
|
|
)
|
|
assert hasattr(ffprobe, "cache_info"), "ffprobe should have cache_info method from timed_lru_cache"
|
|
|
|
# Clear cache to start fresh
|
|
ffprobe.cache_clear()
|
|
|
|
# Mock subprocess to avoid actual ffprobe execution
|
|
call_count = 0
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
|
|
def create_mock_process(*_args, **_kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
mock_process = AsyncMock()
|
|
mock_process.wait.return_value = 0
|
|
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}, "streams": []}', b"")
|
|
return mock_process
|
|
|
|
mock_subprocess.side_effect = create_mock_process
|
|
|
|
with patch("anyio.open_file") as mock_open_file:
|
|
mock_file = AsyncMock()
|
|
mock_open_file.return_value.__aenter__.return_value = mock_file
|
|
|
|
# First call should execute the function
|
|
result1 = await ffprobe(str(self.test_file))
|
|
assert isinstance(result1.metadata, dict)
|
|
first_call_count = call_count
|
|
|
|
# Second call with same argument should use cached result
|
|
result2 = await ffprobe(str(self.test_file))
|
|
assert isinstance(result2.metadata, dict)
|
|
|
|
assert call_count == first_call_count, (
|
|
"The subprocess should not be called again for the actual ffprobe execution (it may be called for the -h check, but the main execution should be cached) - Second call should use cached result"
|
|
)
|
|
|
|
assert result1.metadata == result2.metadata, (
|
|
"Results should be equivalent (same data, may not be same object due to async nature)"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ffprobe_with_path_object(self):
|
|
"""Test ffprobe with Path object input."""
|
|
from app.features.streaming.library.ffprobe import ffprobe
|
|
|
|
# Mock subprocess to avoid actual ffprobe execution
|
|
with patch("asyncio.create_subprocess_exec") as mock_subprocess:
|
|
mock_process = AsyncMock()
|
|
mock_process.wait.return_value = 0
|
|
mock_process.communicate.return_value = (b'{"format": {"duration": "10.0"}}', b"")
|
|
mock_subprocess.return_value = mock_process
|
|
|
|
with patch("anyio.open_file") as mock_open_file:
|
|
mock_file = AsyncMock()
|
|
mock_open_file.return_value.__aenter__.return_value = mock_file
|
|
|
|
# Test with Path object
|
|
result = await ffprobe(self.test_file)
|
|
assert result.metadata == {"duration": "10.0"}
|
|
|
|
def test_ffprobe_result_properties(self):
|
|
"""Test FFProbeResult object properties."""
|
|
from app.features.streaming.library.ffprobe import FFStream, FFProbeResult
|
|
|
|
result = FFProbeResult()
|
|
|
|
assert result.video == [], "Test empty result"
|
|
assert result.audio == []
|
|
assert result.subtitle == []
|
|
assert result.attachment == []
|
|
assert result.metadata == {}
|
|
|
|
# Test adding streams
|
|
video_stream = FFStream(
|
|
{"index": 0, "codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080}
|
|
)
|
|
|
|
audio_stream = FFStream({"index": 1, "codec_type": "audio", "codec_name": "aac", "channels": 2})
|
|
|
|
result.video.append(video_stream)
|
|
result.audio.append(audio_stream)
|
|
|
|
assert len(result.video) == 1
|
|
assert len(result.audio) == 1
|
|
assert result.video[0].is_video()
|
|
assert result.audio[0].is_audio()
|
|
|
|
def test_stream_object_methods(self):
|
|
"""Test Stream object methods."""
|
|
from app.features.streaming.library.ffprobe import FFStream
|
|
|
|
# Test video stream
|
|
video_stream = FFStream(
|
|
{"codec_type": "video", "codec_name": "h264", "width": 1920, "height": 1080, "duration": "10.5"}
|
|
)
|
|
|
|
assert video_stream.is_video()
|
|
assert not video_stream.is_audio()
|
|
assert not video_stream.is_subtitle()
|
|
assert not video_stream.is_attachment()
|
|
|
|
# Test audio stream
|
|
audio_stream = FFStream({"codec_type": "audio", "codec_name": "aac", "channels": 2, "sample_rate": "44100"})
|
|
|
|
assert not audio_stream.is_video()
|
|
assert audio_stream.is_audio()
|
|
assert not audio_stream.is_subtitle()
|
|
assert not audio_stream.is_attachment()
|
|
|
|
# Test subtitle stream
|
|
subtitle_stream = FFStream({"codec_type": "subtitle", "codec_name": "subrip"})
|
|
|
|
assert not subtitle_stream.is_video()
|
|
assert not subtitle_stream.is_audio()
|
|
assert subtitle_stream.is_subtitle()
|
|
assert not subtitle_stream.is_attachment()
|
|
|
|
# Test attachment stream
|
|
attachment_stream = FFStream({"codec_type": "attachment", "codec_name": "ttf"})
|
|
|
|
assert not attachment_stream.is_video()
|
|
assert not attachment_stream.is_audio()
|
|
assert not attachment_stream.is_subtitle()
|
|
assert attachment_stream.is_attachment()
|