wrolpi/modules/videos/test/test_transcode_audio.py
lrnselfreliance 15d3792b0e
Some checks are pending
CodeQL / Analyze (push) Waiting to run
Controller Tests / controller-tests (push) Waiting to run
Jest Tests / jest-tests (push) Waiting to run
videos: transcode can remove the audio stream (silent video) (#737)
A video's audio stream can be dropped the same way its video stream could
already be: a new REMOVE_AUDIO audio target maps only the video stream with
-an, needs a video container, and is refused for audio files.  Output
verification insists the result has no audio stream.

Removing both streams is refused by the validator, the API, resolve_output,
and the modal: choosing Remove for one stream resets the other to Keep and
withdraws the other's Remove option.

Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-26 17:49:14 -06:00

423 lines
21 KiB
Python

"""Audio-only transcodes: removing the video stream from a video, and transcoding audio files.
Also the mirror: removing the audio stream from a video (`REMOVE_AUDIO`)."""
import json
import pathlib
import shutil
from unittest import mock
import pytest
from modules.videos.transcode import transcode_video_file, validate_transcode_request, verify_transcode_output, \
REMOVE_VIDEO, REMOVE_AUDIO, TRANSCODE_AUDIO_CONTAINERS, default_audio_container, transcode_video_job
from wrolpi import jobs
from wrolpi.cmd import CommandResult
from wrolpi.vars import PROJECT_DIR
MP4 = 'mov,mp4,m4a,3gp,3g2,mj2'
def make_probe(vcodec, acodec, duration=10.0, format_name=MP4):
streams = []
if vcodec:
streams.append({'codec_type': 'video', 'codec_name': vcodec})
if acodec:
streams.append({'codec_type': 'audio', 'codec_name': acodec})
return {'streams': streams, 'format': {'format_name': format_name, 'duration': str(duration)}}
class FakeProbes:
def __init__(self, source, output):
self.source, self.output = source, output
def __call__(self, path: pathlib.Path):
return self.output if '.transcode.' in path.name else self.source
def fake_ffmpeg(cmd, **kwargs):
"""Writes the output file; the tail-decode check writes nothing."""
if cmd[-1] != '-':
pathlib.Path(cmd[-1]).write_bytes(b'output data')
return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1)
async def async_fake_ffmpeg(cmd, **kwargs):
return fake_ffmpeg(cmd, **kwargs)
# ------------------------------------------------------------------ validation
def test_validate_remove_video():
"""Removing the video needs an audio container; an audio container needs a compatible codec."""
validate_transcode_request(REMOVE_VIDEO, 'aac', 'm4a')
validate_transcode_request(REMOVE_VIDEO, None, 'ogg') # Copy: checked against the source later.
validate_transcode_request(REMOVE_VIDEO, 'mp3', 'mp3')
with pytest.raises(ValueError, match='audio container'):
validate_transcode_request(REMOVE_VIDEO, 'aac', 'mp4')
with pytest.raises(ValueError, match='ogg'):
validate_transcode_request(REMOVE_VIDEO, 'aac', 'ogg')
with pytest.raises(ValueError, match='mp3'):
validate_transcode_request(REMOVE_VIDEO, 'opus', 'mp3')
with pytest.raises(ValueError, match='video'):
validate_transcode_request('h264', 'aac', 'm4a') # A video stream cannot live in m4a.
assert set(TRANSCODE_AUDIO_CONTAINERS) == {'m4a', 'ogg', 'mp3'}
def test_default_audio_container():
assert default_audio_container('aac') == 'm4a'
assert default_audio_container('opus') == 'ogg'
assert default_audio_container('vorbis') == 'ogg'
assert default_audio_container('mp3') == 'mp3'
assert default_audio_container('flac') == 'ogg'
assert default_audio_container(None) == 'm4a'
# ------------------------------------------------------------------ verification
def test_verify_audio_only_output():
source = make_probe('vp9', 'opus', 100.0, 'matroska,webm')
good = make_probe(None, 'aac', 100.0)
verify_transcode_output(source, good, REMOVE_VIDEO, 'aac', 'm4a')
# The video stream must be gone.
with pytest.raises(RuntimeError, match='video stream'):
verify_transcode_output(source, make_probe('vp9', 'aac', 100.0), REMOVE_VIDEO, 'aac', 'm4a')
# An embedded cover is not a video stream.
verify_transcode_output(source, make_probe('mjpeg', 'aac', 100.0), REMOVE_VIDEO, 'aac', 'm4a')
# Audio is required.
with pytest.raises(RuntimeError, match='audio'):
verify_transcode_output(source, make_probe(None, None, 100.0), REMOVE_VIDEO, 'aac', 'm4a')
def test_verify_container_names():
"""ffprobe names containers differently than the file suffix does."""
source = make_probe('h264', 'aac', 10.0)
verify_transcode_output(source, make_probe('h264', 'aac', 10.0, 'matroska,webm'), None, None, 'mkv')
verify_transcode_output(source, make_probe(None, 'opus', 10.0, 'ogg'), REMOVE_VIDEO, 'opus', 'ogg')
verify_transcode_output(source, make_probe(None, 'mp3', 10.0, 'mp3'), REMOVE_VIDEO, 'mp3', 'mp3')
verify_transcode_output(source, make_probe(None, 'aac', 10.0, MP4), REMOVE_VIDEO, None, 'm4a')
with pytest.raises(RuntimeError, match='container'):
verify_transcode_output(source, make_probe(None, 'mp3', 10.0, 'mp3'), REMOVE_VIDEO, 'mp3', 'ogg')
def test_verify_audio_source():
"""An audio-only source has no video to demand; the output is audio-only too."""
source = make_probe(None, 'mp3', 100.0, 'mp3')
verify_transcode_output(source, make_probe(None, 'aac', 100.0), None, 'aac', 'm4a')
verify_transcode_output(source, make_probe(None, 'mp3', 100.0, 'mp3'), None, None, 'mp3')
# ------------------------------------------------------------------ transcode_video_file
@pytest.mark.asyncio
async def test_remove_video(test_directory, async_client):
"""Removing the video: no video map, `-vn`, the audio is required, the output is an audio file
named for its container, and the original is replaced."""
video_path = test_directory / 'movie.mp4'
video_path.write_bytes(b'fake video data')
probes = FakeProbes(make_probe('h264', 'aac', 10.0), make_probe(None, 'aac', 10.0))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
result = await transcode_video_file(video_path, target_vcodec=REMOVE_VIDEO, target_acodec=None,
container='m4a')
assert result == test_directory / 'movie.m4a' and result.is_file()
assert not video_path.exists()
cmd = mock_run.call_args_list[0][0][0]
assert '-vn' in cmd
assert '0:V:0' not in cmd
assert cmd[cmd.index('-map') + 1] == '0:a:0', 'Audio is required, not optional'
assert '-c:v' not in cmd
assert cmd[cmd.index('-c:a') + 1] == 'copy'
assert '+faststart' in cmd, 'm4a is an mp4; fast start applies'
assert str(cmd[-1]).endswith('movie.transcode.m4a')
@pytest.mark.asyncio
async def test_remove_video_to_ogg_no_faststart(test_directory, async_client):
video_path = test_directory / 'movie.webm'
video_path.write_bytes(b'fake video data')
probes = FakeProbes(make_probe('vp9', 'opus', 10.0, 'matroska,webm'), make_probe(None, 'opus', 10.0, 'ogg'))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
result = await transcode_video_file(video_path, target_vcodec=REMOVE_VIDEO, container='ogg')
assert result == test_directory / 'movie.ogg'
cmd = mock_run.call_args_list[0][0][0]
assert '-movflags' not in cmd, 'movflags is an mp4 muxer option'
@pytest.mark.asyncio
async def test_audio_source_is_audio_output(test_directory, async_client):
"""An audio file has no video to keep or remove: the output is audio-only whatever the video
target says, and an unsuitable container is replaced by the codec's natural one."""
audio_path = test_directory / 'song.webm'
audio_path.write_bytes(b'fake audio data')
probes = FakeProbes(make_probe(None, 'opus', 10.0, 'matroska,webm'), make_probe(None, 'opus', 10.0, 'ogg'))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
result = await transcode_video_file(audio_path, target_vcodec=None, target_acodec=None, container='mp4')
assert result == test_directory / 'song.ogg', 'opus copied: ogg, not the video container asked for'
cmd = mock_run.call_args_list[0][0][0]
assert '-vn' in cmd and '0:V:0' not in cmd
@pytest.mark.asyncio
async def test_copy_audio_into_incompatible_container_is_refused(test_directory, async_client):
"""Copying opus into m4a cannot work; refuse before ffmpeg runs."""
video_path = test_directory / 'movie.webm'
video_path.write_bytes(b'fake video data')
probes = FakeProbes(make_probe('vp9', 'opus', 10.0, 'matroska,webm'), make_probe(None, 'opus', 10.0))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
with pytest.raises(RuntimeError, match='m4a'):
await transcode_video_file(video_path, target_vcodec=REMOVE_VIDEO, target_acodec=None, container='m4a')
mock_run.assert_not_called()
assert video_path.is_file()
@pytest.mark.asyncio
async def test_video_target_on_audio_source_is_refused(test_directory, async_client):
audio_path = test_directory / 'song.mp3'
audio_path.write_bytes(b'fake audio data')
probes = FakeProbes(make_probe(None, 'mp3', 10.0, 'mp3'), make_probe(None, 'mp3', 10.0, 'mp3'))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
with pytest.raises(RuntimeError, match='no video stream'):
await transcode_video_file(audio_path, target_vcodec='h264', container='mp4')
mock_run.assert_not_called()
# ------------------------------------------------------------------ Job + API
@pytest.mark.asyncio
async def test_remove_video_job(test_session, test_directory, async_client, video_factory):
"""The FileGroup follows the video into its audio-only life: same id, new suffix and mimetype."""
from modules.videos.models import Video
from wrolpi.files.models import FileGroup
video = video_factory()
video.file_group.length = 5
test_session.commit()
file_group_id = video.file_group_id
video_path = video.video_path
probes = FakeProbes(make_probe('h264', 'aac', 5.3), make_probe(None, 'mp3', 5.3, 'mp3'))
async def fake_run_command(cmd, stdout_callback=None, **kwargs):
if cmd[-1] != '-':
shutil.copy(PROJECT_DIR / 'test/big_buck_bunny.mp3', cmd[-1])
return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1)
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('wrolpi.cmd.run_command', side_effect=fake_run_command):
job_id = transcode_video_job.enqueue(file_group_id=file_group_id, video_codec=REMOVE_VIDEO,
audio_codec='mp3', container='mp3')
record = await jobs.wait_for_job(job_id)
assert record['status'] == jobs.COMPLETE, record
new_path = video_path.with_suffix('.mp3')
assert new_path.is_file() and not video_path.exists()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, file_group_id)
assert video.video_path == new_path
assert video.file_group.mimetype.startswith('audio/')
assert test_session.query(FileGroup).count() == 1
@pytest.mark.asyncio
async def test_transcode_api_audio_file(async_client, test_session, audio_factory):
"""An audio file can be transcoded (audio codec / audio container); a video target is a 400."""
audio = audio_factory()
test_session.commit()
request, response = await async_client.post(f'/api/videos/{audio.file_group_id}/transcode',
content=json.dumps({'video_codec': 'h264', 'container': 'mp4'}))
assert response.status == 400, response.json
assert 'audio' in response.json['message'].lower() or 'audio' in response.json['error'].lower()
request, response = await async_client.post(f'/api/videos/{audio.file_group_id}/transcode',
content=json.dumps({'audio_codec': 'aac', 'container': 'm4a'}))
assert response.status == 200, response.json
jobs.cancel_job(response.json['job_id'])
# Remove-video on an audio file is a no-op wording-wise but allowed (it is already audio).
request, response = await async_client.post(f'/api/videos/{audio.file_group_id}/transcode',
content=json.dumps({'video_codec': REMOVE_VIDEO, 'container': 'mp3'}))
assert response.status == 200, response.json
jobs.cancel_job(response.json['job_id'])
@pytest.mark.asyncio
async def test_transcode_api_remove_video_needs_audio_container(async_client, test_session, video_factory):
video = video_factory()
test_session.commit()
request, response = await async_client.post(f'/api/videos/{video.file_group_id}/transcode',
content=json.dumps({'video_codec': REMOVE_VIDEO, 'container': 'mp4'}))
assert response.status == 400, response.json
request, response = await async_client.post(f'/api/videos/{video.file_group_id}/transcode',
content=json.dumps({'video_codec': REMOVE_VIDEO, 'audio_codec': 'aac',
'container': 'm4a'}))
assert response.status == 200, response.json
jobs.cancel_job(response.json['job_id'])
# ------------------------------------------------------------------ remove audio
def test_validate_remove_audio():
"""Removing the audio needs a video container; both streams cannot be removed."""
validate_transcode_request(None, REMOVE_AUDIO, 'mp4')
validate_transcode_request('h264', REMOVE_AUDIO, 'mkv')
validate_transcode_request(None, REMOVE_AUDIO, 'mp4', fragmented=True)
with pytest.raises(ValueError, match='video container'):
validate_transcode_request(None, REMOVE_AUDIO, 'm4a')
with pytest.raises(ValueError, match='video container'):
validate_transcode_request(None, REMOVE_AUDIO, 'mp3')
with pytest.raises(ValueError, match='both'):
validate_transcode_request(REMOVE_VIDEO, REMOVE_AUDIO, 'mp4')
with pytest.raises(ValueError, match='both'):
validate_transcode_request(REMOVE_VIDEO, REMOVE_AUDIO, 'm4a')
def test_verify_remove_audio_output():
"""The output must have the video stream and must not have an audio stream."""
source = make_probe('vp9', 'opus', 100.0)
verify_transcode_output(source, make_probe('h264', None, 100.0), 'h264', REMOVE_AUDIO, 'mp4')
verify_transcode_output(source, make_probe('vp9', None, 100.0), None, REMOVE_AUDIO, 'mp4')
with pytest.raises(RuntimeError, match='still has an audio stream'):
verify_transcode_output(source, make_probe('vp9', 'opus', 100.0), None, REMOVE_AUDIO, 'mp4')
with pytest.raises(RuntimeError, match='no video stream'):
verify_transcode_output(source, make_probe(None, None, 100.0), None, REMOVE_AUDIO, 'mp4')
# A silent source can have its (absent) audio removed; the result is unchanged.
verify_transcode_output(make_probe('vp9', None, 100.0), make_probe('vp9', None, 100.0), None, REMOVE_AUDIO, 'mp4')
@pytest.mark.asyncio
async def test_remove_audio(test_directory, async_client):
"""Removing the audio: the video is mapped, `-an` drops the audio, no `-c:a`, the output keeps a
video container and the video stream is copied unless a target is given."""
video_path = test_directory / 'movie.webm'
video_path.write_bytes(b'fake video data')
probes = FakeProbes(make_probe('vp9', 'opus', 10.0, 'matroska,webm'), make_probe('vp9', None, 10.0, 'matroska,webm'))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
result = await transcode_video_file(video_path, target_vcodec=None, target_acodec=REMOVE_AUDIO,
container='mkv')
assert result == test_directory / 'movie.mkv' and result.is_file()
assert not video_path.exists()
cmd = mock_run.call_args_list[0][0][0]
assert '-an' in cmd
assert '-vn' not in cmd
assert '0:a:0?' not in cmd and '0:a:0' not in cmd
assert cmd[cmd.index('-map') + 1] == '0:V:0'
assert '-c:a' not in cmd
assert cmd[cmd.index('-c:v') + 1] == 'copy'
assert str(cmd[-1]).endswith('movie.transcode.mkv')
@pytest.mark.asyncio
async def test_remove_audio_with_video_target(test_directory, async_client):
video_path = test_directory / 'movie.webm'
video_path.write_bytes(b'fake video data')
probes = FakeProbes(make_probe('vp9', 'opus', 10.0, 'matroska,webm'), make_probe('h264', None, 10.0))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
result = await transcode_video_file(video_path, target_vcodec='h264', target_acodec=REMOVE_AUDIO,
container='mp4')
assert result == test_directory / 'movie.mp4'
cmd = mock_run.call_args_list[0][0][0]
assert '-an' in cmd
assert cmd[cmd.index('-c:v') + 1] == 'libx264'
assert '-c:a' not in cmd
assert '+faststart' in cmd
@pytest.mark.asyncio
async def test_remove_audio_from_audio_source_is_refused(test_directory, async_client):
"""An audio file with its audio removed is nothing; ffmpeg is never run."""
audio_path = test_directory / 'song.mp3'
audio_path.write_bytes(b'fake audio data')
probes = FakeProbes(make_probe(None, 'mp3', 10.0, 'mp3'), make_probe(None, 'mp3', 10.0, 'mp3'))
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('modules.videos.transcode.run_command', side_effect=async_fake_ffmpeg) as mock_run:
with pytest.raises(RuntimeError, match='no video stream'):
await transcode_video_file(audio_path, target_acodec=REMOVE_AUDIO, container='mp4')
mock_run.assert_not_called()
assert audio_path.is_file()
def test_resolve_output_refuses_removing_both():
"""Even without `validate_transcode_request`, ffmpeg is never asked for nothing."""
from modules.videos.transcode import resolve_output
with pytest.raises(RuntimeError, match='both'):
resolve_output(make_probe('h264', 'aac'), REMOVE_VIDEO, REMOVE_AUDIO, 'mp4')
with pytest.raises(RuntimeError, match='both'):
resolve_output(make_probe(None, 'aac'), REMOVE_VIDEO, REMOVE_AUDIO, 'mp3')
@pytest.mark.asyncio
async def test_remove_audio_job(test_session, test_directory, async_client, video_factory):
"""The Job removes the audio in place: same FileGroup, still a video."""
from modules.videos.models import Video
from wrolpi.files.models import FileGroup
video = video_factory()
video.file_group.length = 5
test_session.commit()
file_group_id = video.file_group_id
video_path = video.video_path
probes = FakeProbes(make_probe('h264', 'aac', 5.3), make_probe('h264', None, 5.3))
async def fake_run_command(cmd, stdout_callback=None, **kwargs):
if cmd[-1] != '-':
shutil.copy(PROJECT_DIR / 'test/big_buck_bunny_720p_1mb.mp4', cmd[-1])
return CommandResult(return_code=0, cancelled=False, stdout=b'', stderr=b'', elapsed=1)
with mock.patch('modules.videos.transcode.probe_for_verify', side_effect=probes), \
mock.patch('wrolpi.cmd.run_command', side_effect=fake_run_command):
job_id = transcode_video_job.enqueue(file_group_id=file_group_id, video_codec=None,
audio_codec=REMOVE_AUDIO, container='mp4')
record = await jobs.wait_for_job(job_id)
assert record['status'] == jobs.COMPLETE, record
assert video_path.with_suffix('.mp4').is_file()
test_session.expire_all()
video = Video.find_by_file_group_id(test_session, file_group_id)
assert video.file_group.mimetype.startswith('video/')
assert test_session.query(FileGroup).count() == 1
@pytest.mark.asyncio
async def test_transcode_api_remove_audio(async_client, test_session, video_factory, audio_factory):
"""Remove-audio is queued for a video with a video container; refused on an audio file, with an
audio container, and together with remove-video."""
video = video_factory()
audio = audio_factory()
test_session.commit()
request, response = await async_client.post(f'/api/videos/{video.file_group_id}/transcode',
content=json.dumps({'audio_codec': REMOVE_AUDIO, 'container': 'mp4'}))
assert response.status == 200, response.json
assert 'Remove audio' in jobs.get_job(response.json['job_id'])['description']
jobs.cancel_job(response.json['job_id'])
request, response = await async_client.post(f'/api/videos/{video.file_group_id}/transcode',
content=json.dumps({'audio_codec': REMOVE_AUDIO, 'container': 'm4a'}))
assert response.status == 400, response.json
request, response = await async_client.post(f'/api/videos/{video.file_group_id}/transcode',
content=json.dumps({'video_codec': REMOVE_VIDEO,
'audio_codec': REMOVE_AUDIO, 'container': 'mp4'}))
assert response.status == 400, response.json
request, response = await async_client.post(f'/api/videos/{audio.file_group_id}/transcode',
content=json.dumps({'audio_codec': REMOVE_AUDIO, 'container': 'mp4'}))
assert response.status == 400, response.json