Page MenuHome

test_commands_create_video.py
No OneTemporary

File Metadata

Created
Thu, Dec 12, 2:39 PM

test_commands_create_video.py

import logging
import shutil
import typing
from pathlib import Path
import platform
import shlex
import subprocess
import sys
import tempfile
from unittest import mock
from tests.test_runner import AbstractCommandTest
log = logging.getLogger(__name__)
class CreateVideoTest(AbstractCommandTest):
settings: typing.Dict[str, typing.Any] = {
'ffmpeg_cmd': f'{Path(sys.executable).absolute().as_posix()!r} -hide_banner',
'input_files': '/tmp/*.png',
'output_file': '/tmp/merged.mkv',
'fps': 24,
}
def setUp(self):
super().setUp()
from flamenco_worker.commands import CreateVideoCommand
self.cmd = CreateVideoCommand(
worker=self.fworker,
task_id='12345',
command_idx=0,
)
self.settings = self.settings.copy()
def test_validate(self):
self.assertIn('not found on $PATH', self.cmd.validate({'ffmpeg_cmd': '/does/not/exist'}))
self.assertIsNone(self.cmd.validate(self.settings))
def test_validate_without_ffmpeg(self):
settings = self.settings.copy()
del settings['ffmpeg_cmd']
self.assertIsNone(self.cmd.validate(settings))
self.assertEqual(['ffmpeg'], settings['ffmpeg_cmd'],
'The default setting should be stored in the dict after validation')
def test_build_ffmpeg_cmd_windows(self):
self.cmd.validate(self.settings)
with mock.patch('platform.system') as mock_system:
mock_system.return_value = 'Windows'
cliargs = self.cmd._build_ffmpeg_command(self.settings)
self.assertEqual([
Path(sys.executable).absolute().as_posix(), '-hide_banner',
'-r', '24',
'-f', 'concat',
'-i', Path(self.settings['input_files']).absolute().with_name('ffmpeg-input.txt').as_posix(),
'-c:v', 'h264',
'-crf', '23',
'-g', '18',
'-vf', 'pad=ceil(iw/2)*2:ceil(ih/2)*2',
'-y',
'-bf', '0',
'/tmp/merged.mkv',
], cliargs)
def test_build_ffmpeg_cmd_linux(self):
self.cmd.validate(self.settings)
with mock.patch('platform.system') as mock_system:
mock_system.return_value = 'Linux'
cliargs = self.cmd._build_ffmpeg_command(self.settings)
self.assertEqual([
Path(sys.executable).absolute().as_posix(), '-hide_banner',
'-r', '24',
'-pattern_type', 'glob',
'-i', '/tmp/*.png',
'-c:v', 'h264',
'-crf', '23',
'-g', '18',
'-vf', 'pad=ceil(iw/2)*2:ceil(ih/2)*2',
'-y',
'-bf', '0',
'/tmp/merged.mkv',
], cliargs)
def test_run_ffmpeg(self):
with tempfile.TemporaryDirectory() as tempdir:
outfile = Path(tempdir) / 'merged.mkv'
frame_dir = Path(__file__).with_name('test_frames')
settings: typing.Dict[str, typing.Any] = {
**self.settings,
'ffmpeg_cmd': 'ffmpeg', # use the real FFmpeg for this test.
'input_files': f'{frame_dir.as_posix()}/*.png',
'output_file': outfile.as_posix(),
}
self.loop.run_until_complete(self.cmd.run(settings))
self.assertTrue(outfile.exists())
ffprobe_cmd = [shutil.which('ffprobe'), '-v', 'error',
'-show_entries', 'format=duration',
'-of', 'default=noprint_wrappers=1:nokey=1',
outfile.as_posix()]
log.debug('Running %s', ' '.join(shlex.quote(arg) for arg in ffprobe_cmd))
probe_out = subprocess.check_output(ffprobe_cmd)
probed_duration = float(probe_out)
fps: int = settings['fps']
expect_duration = len(list(frame_dir.glob('*.png'))) / fps
self.assertAlmostEqual(expect_duration, probed_duration, places=3)

Event Timeline