Daniil
138 lines
4.1 KiB
Python
138 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
from cpv3.modules.media.service import apply_silence_cuts, remove_silence
|
|
|
|
|
|
class _TempDownload:
|
|
def __init__(self, path: str) -> None:
|
|
self.path = path
|
|
|
|
def cleanup(self) -> None:
|
|
return None
|
|
|
|
|
|
class _FakeProcess:
|
|
returncode = 0
|
|
|
|
async def communicate(self) -> tuple[bytes, bytes]:
|
|
return b"", b""
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_remove_silence_uses_single_input_trim_filter(tmp_path) -> None:
|
|
input_path = tmp_path / "input.mp4"
|
|
input_path.write_bytes(b"video")
|
|
|
|
file_info = SimpleNamespace(
|
|
file_path="uploads/test-file.txt",
|
|
file_url="http://example.com/uploads/test-file.txt",
|
|
file_size=1024,
|
|
filename="test-file.txt",
|
|
)
|
|
storage = SimpleNamespace(
|
|
download_to_temp=AsyncMock(return_value=_TempDownload(str(input_path))),
|
|
upload_fileobj=AsyncMock(return_value="uploads/test-file.txt"),
|
|
get_file_info=AsyncMock(return_value=file_info),
|
|
)
|
|
|
|
captured: dict[str, tuple[str, ...]] = {}
|
|
|
|
async def fake_create_subprocess_exec(
|
|
*cmd: str, stdout=None, stderr=None
|
|
) -> _FakeProcess:
|
|
captured["cmd"] = cmd
|
|
return _FakeProcess()
|
|
|
|
with (
|
|
patch(
|
|
"cpv3.modules.media.service._compute_non_silent_segments",
|
|
return_value=[(0, 1000), (2000, 3000)],
|
|
),
|
|
patch(
|
|
"cpv3.modules.media.service.asyncio.create_subprocess_exec",
|
|
side_effect=fake_create_subprocess_exec,
|
|
),
|
|
):
|
|
await remove_silence(
|
|
storage,
|
|
file_key="uploads/input.mp4",
|
|
out_folder="processed",
|
|
)
|
|
|
|
cmd = list(captured["cmd"])
|
|
assert cmd.count("-i") == 1
|
|
|
|
filter_complex = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "trim=start=0.000:end=1.000" in filter_complex
|
|
assert "atrim=start=0.000:end=1.000" in filter_complex
|
|
assert "trim=start=2.000:end=3.000" in filter_complex
|
|
assert "atrim=start=2.000:end=3.000" in filter_complex
|
|
|
|
|
|
async def _run_sync_immediately(func):
|
|
return func()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_silence_cuts_uses_single_input_trim_filter(tmp_path) -> None:
|
|
input_path = tmp_path / "input.mp4"
|
|
input_path.write_bytes(b"video")
|
|
|
|
file_info = SimpleNamespace(
|
|
file_path="uploads/test-file.txt",
|
|
file_url="http://example.com/uploads/test-file.txt",
|
|
file_size=1024,
|
|
filename="test-file.txt",
|
|
)
|
|
storage = SimpleNamespace(
|
|
download_to_temp=AsyncMock(return_value=_TempDownload(str(input_path))),
|
|
upload_fileobj=AsyncMock(return_value="uploads/test-file.txt"),
|
|
get_file_info=AsyncMock(return_value=file_info),
|
|
)
|
|
|
|
captured: dict[str, tuple[str, ...]] = {}
|
|
|
|
async def fake_create_subprocess_exec(
|
|
*cmd: str, stdout=None, stderr=None
|
|
) -> _FakeProcess:
|
|
captured["cmd"] = cmd
|
|
return _FakeProcess()
|
|
|
|
fake_audio = MagicMock()
|
|
fake_audio.__len__.return_value = 5000
|
|
|
|
with (
|
|
patch(
|
|
"cpv3.modules.media.service.anyio.to_thread.run_sync",
|
|
side_effect=_run_sync_immediately,
|
|
),
|
|
patch("pydub.AudioSegment.from_file", return_value=fake_audio),
|
|
patch(
|
|
"cpv3.modules.media.service.asyncio.create_subprocess_exec",
|
|
side_effect=fake_create_subprocess_exec,
|
|
),
|
|
):
|
|
await apply_silence_cuts(
|
|
storage,
|
|
file_key="uploads/input.mp4",
|
|
out_folder="processed",
|
|
cuts=[
|
|
{"start_ms": 1000, "end_ms": 2000},
|
|
{"start_ms": 3000, "end_ms": 3500},
|
|
],
|
|
)
|
|
|
|
cmd = list(captured["cmd"])
|
|
assert cmd.count("-i") == 1
|
|
|
|
filter_complex = cmd[cmd.index("-filter_complex") + 1]
|
|
assert "trim=start=0.000:end=1.000" in filter_complex
|
|
assert "atrim=start=0.000:end=1.000" in filter_complex
|
|
assert "trim=start=2.000:end=3.000" in filter_complex
|
|
assert "atrim=start=3.500:end=5.000" in filter_complex
|