diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml new file mode 100644 index 0000000..e3b2ddd --- /dev/null +++ b/.github/workflows/tests.yml @@ -0,0 +1,39 @@ +name: Tests with pytest + +on: + push: + branches: + - master + pull_request: + branches: + - master + +permissions: + contents: read + +jobs: + pytest: + name: Run tests + runs-on: ubuntu-latest + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + persist-credentials: false + + - name: Set up Python + uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0 + with: + python-version: '3.11' + + - name: Install system dependencies + run: | + sudo apt-get update + sudo apt-get install -y --no-install-recommends ffmpeg mkvtoolnix + + - name: Install requirements + run: pip install -r requirements.txt -r requirements.dev.txt + + - name: Run tests + run: pytest tests/ -v diff --git a/Taskfile.yml b/Taskfile.yml index 4c881fd..a1d88ad 100644 --- a/Taskfile.yml +++ b/Taskfile.yml @@ -91,6 +91,19 @@ tasks: - docker run -it --gpus all -u $(id -u):$(id -g) -v ${PWD}/{{.INPUT_DIRECTORY}}:/app/input -v ${PWD}/{{.OUTPUT_DIRECTORY}}:/app/output -v ${PWD}/{{.FONTS_DIRECTORY}}:/app/fonts --rm {{.IMAGE}}:{{.TARGET}} {{.CLI_ARGS}} # Development tools + test: + desc: Run tests + summary: | + Run the pytest test suite inside the dev container. + + Usage: + - task test + - task test -- -v + - task test -- --cov=ffconv + - task test -- tests/test_helper.py + cmds: + - $DOCKER_COMPOSE_RUN dev python3 -m pytest tests/ {{.CLI_ARGS}} + ruff: desc: Run ruff cmds: diff --git a/requirements.dev.txt b/requirements.dev.txt index cd474af..9592fba 100644 --- a/requirements.dev.txt +++ b/requirements.dev.txt @@ -1,3 +1,5 @@ ruff==0.16.4 mypy==2.3.1 black==26.5.1 +pytest==9.1.1 +pytest-cov==7.1.0 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..218e81a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,200 @@ +import shutil +import subprocess + +import pytest + +from ffconv.cli import mkvmerge_identify_streams + + +def _binaries_available(): + return ( + shutil.which("ffmpeg") is not None + and shutil.which("ffprobe") is not None + and shutil.which("mkvmerge") is not None + ) + + +@pytest.fixture(scope="session") +def test_mkv(tmp_path_factory): + if not _binaries_available(): + pytest.skip("ffmpeg and mkvmerge required") + + tmp = tmp_path_factory.mktemp("media") + + srt_path = tmp / "sub.srt" + srt_path.write_text("1\n00:00:00,000 --> 00:00:02,000\nTest subtitle\n\n") + + mkv_path = tmp / "test.mkv" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "color=black:size=64x64:rate=24:duration=2", + "-f", + "lavfi", + "-i", + "anullsrc=r=44100:cl=mono", + "-i", + str(srt_path), + "-t", + "2", + "-map", + "0:v", + "-map", + "1:a", + "-map", + "2:s", + "-c:v", + "libx264", + "-crf", + "51", + "-preset", + "ultrafast", + "-c:a", + "aac", + "-ab", + "32k", + "-c:s", + "srt", + str(mkv_path), + ], + check=True, + capture_output=True, + ) + + return mkv_path + + +@pytest.fixture(scope="session") +def stream_mapping(test_mkv): + _, mapping = mkvmerge_identify_streams( + test_mkv, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + return mapping + + +@pytest.fixture(scope="session") +def test_mkv_ass(tmp_path_factory): + """MKV with ASS subtitles — the most common real-world subtitle format.""" + if not _binaries_available(): + pytest.skip("ffmpeg, ffprobe and mkvmerge required") + + tmp = tmp_path_factory.mktemp("media_ass") + + ass_path = tmp / "sub.ass" + ass_path.write_text( + "[Script Info]\n" + "ScriptType: v4.00+\n" + "PlayResX: 64\n" + "PlayResY: 64\n" + "\n" + "[V4+ Styles]\n" + "Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour," + " Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline," + " Shadow, Alignment, MarginL, MarginR, MarginV, Encoding\n" + "Style: Default,Arial,12,&H00FFFFFF,&H000000FF,&H00000000,&H00000000," + "0,0,0,0,100,100,0,0,1,2,2,2,10,10,10,1\n" + "\n" + "[Events]\n" + "Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text\n" + "Dialogue: 0,0:00:00.00,0:00:02.00,Default,,0,0,0,,Test subtitle\n" + ) + + mkv_path = tmp / "test_ass.mkv" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "color=black:size=64x64:rate=24:duration=2", + "-f", + "lavfi", + "-i", + "anullsrc=r=44100:cl=mono", + "-i", + str(ass_path), + "-t", + "2", + "-map", + "0:v", + "-map", + "1:a", + "-map", + "2:s", + "-c:v", + "libx264", + "-crf", + "51", + "-preset", + "ultrafast", + "-c:a", + "aac", + "-ab", + "32k", + "-c:s", + "ass", + str(mkv_path), + ], + check=True, + capture_output=True, + ) + + return mkv_path + + +@pytest.fixture(scope="session") +def test_mkv_vorbis(tmp_path_factory): + """MKV with Vorbis audio — exercises the auto-preset default (re-encode) path.""" + if not _binaries_available(): + pytest.skip("ffmpeg, ffprobe and mkvmerge required") + + tmp = tmp_path_factory.mktemp("media_vorbis") + + srt_path = tmp / "sub.srt" + srt_path.write_text("1\n00:00:00,000 --> 00:00:02,000\nTest subtitle\n\n") + + mkv_path = tmp / "test_vorbis.mkv" + subprocess.run( + [ + "ffmpeg", + "-y", + "-f", + "lavfi", + "-i", + "color=black:size=64x64:rate=24:duration=2", + "-f", + "lavfi", + "-i", + "anullsrc=r=44100:cl=mono", + "-i", + str(srt_path), + "-t", + "2", + "-map", + "0:v", + "-map", + "1:a", + "-map", + "2:s", + "-c:v", + "libx264", + "-crf", + "51", + "-preset", + "ultrafast", + "-c:a", + "libvorbis", + "-c:s", + "srt", + str(mkv_path), + ], + check=True, + capture_output=True, + ) + + return mkv_path diff --git a/tests/test_cli_validators.py b/tests/test_cli_validators.py new file mode 100644 index 0000000..82ae8d9 --- /dev/null +++ b/tests/test_cli_validators.py @@ -0,0 +1,98 @@ +import pytest + +from ffconv.cli import validate_stream_count, validate_stream_order +from ffconv.exception import StreamOrderError, StreamTypeMissingError + +FILE_DETAILS = {"file_name": "test.mkv", "batch_name": "batch1"} + + +class TestValidateStreamOrder: + def test_correct_order_passes(self): + streams = { + "video": {"count": 1}, + "audio": {"count": 1}, + "subtitles": {"count": 1}, + } + validate_stream_order(streams, FILE_DETAILS) # should not raise + + def test_audio_first_raises(self): + streams = { + "audio": {"count": 1}, + "video": {"count": 1}, + "subtitles": {"count": 1}, + } + with pytest.raises(StreamOrderError): + validate_stream_order(streams, FILE_DETAILS) + + def test_subtitles_before_audio_raises(self): + streams = { + "video": {"count": 1}, + "subtitles": {"count": 1}, + "audio": {"count": 1}, + } + with pytest.raises(StreamOrderError): + validate_stream_order(streams, FILE_DETAILS) + + def test_error_message_mentions_expected_and_actual_types(self): + streams = { + "audio": {"count": 1}, + "video": {"count": 1}, + "subtitles": {"count": 1}, + } + with pytest.raises(StreamOrderError) as exc_info: + validate_stream_order(streams, FILE_DETAILS) + assert "video" in str(exc_info.value) + assert "audio" in str(exc_info.value) + + +class TestValidateStreamCount: + def test_all_streams_present_passes(self): + streams = { + "video": {"count": 1}, + "audio": {"count": 1}, + "subtitles": {"count": 1}, + } + validate_stream_count(streams, FILE_DETAILS) # should not raise + + def test_missing_video_raises(self): + streams = { + "audio": {"count": 1}, + "subtitles": {"count": 1}, + } + with pytest.raises(StreamTypeMissingError): + validate_stream_count(streams, FILE_DETAILS) + + def test_missing_audio_raises(self): + streams = { + "video": {"count": 1}, + "subtitles": {"count": 1}, + } + with pytest.raises(StreamTypeMissingError): + validate_stream_count(streams, FILE_DETAILS) + + def test_missing_subtitles_raises(self): + streams = { + "video": {"count": 1}, + "audio": {"count": 1}, + } + with pytest.raises(StreamTypeMissingError): + validate_stream_count(streams, FILE_DETAILS) + + def test_empty_streams_raises(self): + with pytest.raises(StreamTypeMissingError): + validate_stream_count({}, FILE_DETAILS) + + def test_error_message_mentions_missing_type(self): + streams = {"video": {"count": 1}, "audio": {"count": 1}} + with pytest.raises(StreamTypeMissingError) as exc_info: + validate_stream_count(streams, FILE_DETAILS) + assert "subtitles" in str(exc_info.value) + + def test_extra_stream_types_still_pass(self): + streams = { + "video": {"count": 1}, + "audio": {"count": 2}, + "subtitles": {"count": 3}, + "attachments": {"count": 1}, + } + validate_stream_count(streams, FILE_DETAILS) # should not raise diff --git a/tests/test_exception.py b/tests/test_exception.py new file mode 100644 index 0000000..1db7519 --- /dev/null +++ b/tests/test_exception.py @@ -0,0 +1,108 @@ +import pytest + +from ffconv.exception import ( + FFmpegError, + MKVmergeError, + ProcessError, + StreamOrderError, + StreamTypeMissingError, +) + +FILE_DETAILS = {"file_name": "test.mkv", "batch_name": "batch1"} + + +class TestMKVmergeError: + def test_message_contains_exit_code(self): + err = MKVmergeError("something went wrong", 1) + assert "1" in str(err) + + def test_message_contains_error_text(self): + err = MKVmergeError("something went wrong", 1) + assert "something went wrong" in str(err) + + def test_exit_code_stored(self): + err = MKVmergeError("error", 2) + assert err.exit_code == 2 + + def test_is_exception(self): + with pytest.raises(MKVmergeError): + raise MKVmergeError("error", 1) + + def test_is_subclass_of_exception(self): + assert issubclass(MKVmergeError, Exception) + + +class TestFFmpegError: + def test_message_contains_exit_code(self): + err = FFmpegError("encode failed", 1) + assert "1" in str(err) + + def test_message_contains_error_text(self): + err = FFmpegError("encode failed", 1) + assert "encode failed" in str(err) + + def test_exit_code_stored(self): + err = FFmpegError("error", 127) + assert err.exit_code == 127 + + def test_is_subclass_of_exception(self): + assert issubclass(FFmpegError, Exception) + + +class TestProcessError: + def test_message_contains_exit_code(self): + err = ProcessError("unknown failure", 255) + assert "255" in str(err) + + def test_message_contains_error_text(self): + err = ProcessError("unknown failure", 255) + assert "unknown failure" in str(err) + + def test_exit_code_stored(self): + err = ProcessError("error", 255) + assert err.exit_code == 255 + + def test_is_subclass_of_exception(self): + assert issubclass(ProcessError, Exception) + + +class TestStreamOrderError: + def test_message_contains_expected_stream_type(self): + err = StreamOrderError("video", 0, "audio", FILE_DETAILS) + assert "video" in str(err) + + def test_message_contains_actual_stream_type(self): + err = StreamOrderError("video", 0, "audio", FILE_DETAILS) + assert "audio" in str(err) + + def test_message_contains_index(self): + err = StreamOrderError("video", 0, "audio", FILE_DETAILS) + assert "0" in str(err) + + def test_message_contains_file_name(self): + err = StreamOrderError("video", 0, "audio", FILE_DETAILS) + assert "test.mkv" in str(err) + + def test_message_contains_batch_name(self): + err = StreamOrderError("video", 0, "audio", FILE_DETAILS) + assert "batch1" in str(err) + + def test_is_subclass_of_exception(self): + assert issubclass(StreamOrderError, Exception) + + +class TestStreamTypeMissingError: + def test_message_contains_stream_type(self): + err = StreamTypeMissingError("subtitles", FILE_DETAILS) + assert "subtitles" in str(err) + + def test_message_contains_file_name(self): + err = StreamTypeMissingError("audio", FILE_DETAILS) + assert "test.mkv" in str(err) + + def test_message_contains_batch_name(self): + err = StreamTypeMissingError("video", FILE_DETAILS) + assert "batch1" in str(err) + + def test_is_subclass_of_exception(self): + assert issubclass(StreamTypeMissingError, Exception) diff --git a/tests/test_helper.py b/tests/test_helper.py new file mode 100644 index 0000000..48fd702 --- /dev/null +++ b/tests/test_helper.py @@ -0,0 +1,211 @@ +import json + +from ffconv.helper import ( + combine_arguments_by_batch, + dict_to_list, + files_in_dir, + preprocess_streams, + read_json, + remove_empty_dict_values, + replace_conflicting_characters_in_filename, + split_list_of_dicts_by_key, +) + + +class TestRemoveEmptyDictValues: + def test_removes_none_values(self): + assert remove_empty_dict_values({"a": 1, "b": None}) == {"a": 1} + + def test_removes_empty_string(self): + assert remove_empty_dict_values({"a": "hello", "b": ""}) == {"a": "hello"} + + def test_removes_empty_list(self): + assert remove_empty_dict_values({"a": [1], "b": []}) == {"a": [1]} + + def test_empty_dict(self): + assert remove_empty_dict_values({}) == {} + + def test_all_values_present(self): + assert remove_empty_dict_values({"a": 1, "b": "x"}) == {"a": 1, "b": "x"} + + def test_removes_zero(self): + assert remove_empty_dict_values({"a": 1, "b": 0}) == {"a": 1} + + +class TestDictToList: + def test_single_pair(self): + assert dict_to_list({"-c:v": "libx264"}) == ["-c:v", "libx264"] + + def test_multiple_pairs(self): + result = dict_to_list({"-c:v": "libx264", "-crf": "18"}) + assert result == ["-c:v", "libx264", "-crf", "18"] + + def test_three_pairs(self): + result = dict_to_list({"-c:v": "libx265", "-crf": "22", "-preset": "slow"}) + assert result == ["-c:v", "libx265", "-crf", "22", "-preset", "slow"] + + +class TestSplitListOfDictsByKey: + def test_single_type(self): + tracks = [{"type": "video", "id": 0}, {"type": "video", "id": 1}] + result, keys = split_list_of_dicts_by_key(tracks, "type") + assert keys == ["video"] + assert result == [[{"type": "video", "id": 0}, {"type": "video", "id": 1}]] + + def test_multiple_types(self): + tracks = [ + {"type": "video", "id": 0}, + {"type": "audio", "id": 1}, + {"type": "subtitles", "id": 2}, + ] + result, keys = split_list_of_dicts_by_key(tracks, "type") + assert keys == ["video", "audio", "subtitles"] + assert len(result) == 3 + assert result[0] == [{"type": "video", "id": 0}] + assert result[1] == [{"type": "audio", "id": 1}] + assert result[2] == [{"type": "subtitles", "id": 2}] + + def test_preserves_first_seen_order(self): + tracks = [{"type": "audio", "id": 0}, {"type": "video", "id": 1}] + _, keys = split_list_of_dicts_by_key(tracks, "type") + assert keys == ["audio", "video"] + + def test_groups_same_type_together(self): + tracks = [ + {"type": "audio", "id": 0}, + {"type": "audio", "id": 1}, + {"type": "video", "id": 2}, + ] + result, keys = split_list_of_dicts_by_key(tracks, "type") + assert keys == ["audio", "video"] + assert len(result[0]) == 2 + + def test_default_key_is_codec_type(self): + tracks = [{"codec_type": "video"}, {"codec_type": "audio"}] + _result, keys = split_list_of_dicts_by_key(tracks) + assert keys == ["video", "audio"] + + +class TestCombineArgumentsByBatch: + def test_single_batch(self): + inputs = [{"batch": 1, "input": "a.mkv"}] + outputs = [{"batch": 1, "output": "a.mp4"}] + result = combine_arguments_by_batch(inputs, outputs) + assert result == [{"batch": 1, "input": "a.mkv", "output": "a.mp4"}] + + def test_multiple_batches(self): + inputs = [{"batch": 1, "input": "a.mkv"}, {"batch": 2, "input": "b.mkv"}] + outputs = [{"batch": 1, "output": "a.mp4"}, {"batch": 2, "output": "b.mp4"}] + result = combine_arguments_by_batch(inputs, outputs) + assert len(result) == 2 + assert result[0] == {"batch": 1, "input": "a.mkv", "output": "a.mp4"} + assert result[1] == {"batch": 2, "input": "b.mkv", "output": "b.mp4"} + + def test_later_list_overwrites_shared_key(self): + a = [{"batch": 1, "value": "first"}] + b = [{"batch": 1, "value": "second"}] + result = combine_arguments_by_batch(a, b) + assert result[0]["value"] == "second" + + +class TestPreprocessStreams: + def test_indexes_by_id(self): + streams = [ + {"id": 0, "properties": {"codec_id": "A_AAC"}}, + {"id": 1, "properties": {"codec_id": "A_AC3"}}, + ] + result = preprocess_streams(streams) + assert result[0] == streams[0] + assert result[1] == streams[1] + + def test_returns_dict(self): + streams = [{"id": 5, "properties": {}}] + result = preprocess_streams(streams) + assert isinstance(result, dict) + assert 5 in result + + def test_empty_list(self): + assert preprocess_streams([]) == {} + + +class TestFilesInDir: + def test_finds_mkv_files(self, tmp_path): + (tmp_path / "video.mkv").touch() + (tmp_path / "other.mp4").touch() + result = files_in_dir(tmp_path) + assert len(result) == 1 + assert result[0].name == "video.mkv" + + def test_custom_file_type(self, tmp_path): + (tmp_path / "video.mkv").touch() + (tmp_path / "clip.mp4").touch() + result = files_in_dir(tmp_path, ["*.mp4"]) + assert len(result) == 1 + assert result[0].name == "clip.mp4" + + def test_empty_directory(self, tmp_path): + result = files_in_dir(tmp_path) + assert result == [] + + def test_case_insensitive_match(self, tmp_path): + (tmp_path / "video.MKV").touch() + result = files_in_dir(tmp_path) + assert len(result) == 1 + + def test_finds_files_recursively(self, tmp_path): + sub = tmp_path / "sub" + sub.mkdir() + (sub / "deep.mkv").touch() + result = files_in_dir(tmp_path) + assert len(result) == 1 + assert result[0].name == "deep.mkv" + + def test_multiple_mkv_files(self, tmp_path): + (tmp_path / "a.mkv").touch() + (tmp_path / "b.mkv").touch() + result = files_in_dir(tmp_path) + assert len(result) == 2 + + +class TestReadJson: + def test_reads_valid_json(self, tmp_path): + data = {"key": "value", "number": 42} + json_file = tmp_path / "test.json" + json_file.write_text(json.dumps(data)) + assert read_json(json_file) == data + + def test_reads_nested_json(self, tmp_path): + data = {"video": {"-c:v": "libx264", "-crf": "18"}} + json_file = tmp_path / "preset.json" + json_file.write_text(json.dumps(data)) + assert read_json(json_file) == data + + +class TestReplaceConflictingCharactersInFilename: + def test_removes_single_quotes(self, tmp_path): + original = tmp_path / "video's.mkv" + original.touch() + result = replace_conflicting_characters_in_filename(original) + assert result.name == "videos.mkv" + assert result.exists() + assert not original.exists() + + def test_removes_double_quotes(self, tmp_path): + original = tmp_path / 'video"test".mkv' + original.touch() + result = replace_conflicting_characters_in_filename(original) + assert result.name == "videotest.mkv" + assert result.exists() + + def test_no_conflicting_characters(self, tmp_path): + original = tmp_path / "clean_video.mkv" + original.touch() + result = replace_conflicting_characters_in_filename(original) + assert result.name == "clean_video.mkv" + assert result.exists() + + def test_returns_path_object(self, tmp_path): + original = tmp_path / "video.mkv" + original.touch() + result = replace_conflicting_characters_in_filename(original) + assert isinstance(result, type(original)) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..d6bba86 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,259 @@ +import json +import re +import shutil +import subprocess + +import pytest + +from ffconv.cli import ffmpeg_convert_file, mkvmerge_identify_streams + +pytestmark = pytest.mark.skipif( + shutil.which("ffmpeg") is None + or shutil.which("ffprobe") is None + or shutil.which("mkvmerge") is None, + reason="ffmpeg, ffprobe and mkvmerge required", +) + +FAST_VIDEO_PRESET = { + "-c:v": "libx264", + "-crf": "51", + "-preset": "ultrafast", + "-pix_fmt": "yuv420p", +} +FAST_AUDIO_PRESET = {"-c:a": "aac", "-ab": "32k"} + + +def _ffprobe_tags(path): + result = subprocess.run( + ["ffprobe", "-v", "quiet", "-print_format", "json", "-show_format", str(path)], + capture_output=True, + check=True, + ) + return json.loads(result.stdout)["format"].get("tags", {}) + + +def _convert( + test_mkv, + stream_mapping, + output_path, + extension="mp4", + filter_preset=None, + auto_audio_preset=False, +): + ffmpeg_convert_file( + input_file=test_mkv, + output_path=output_path, + output_extension=extension, + stream_mapping=stream_mapping, + video_preset=FAST_VIDEO_PRESET, + audio_preset=FAST_AUDIO_PRESET, + filter_preset=filter_preset, + total_items=1, + item_index=0, + batch_index=1, + batch_name="test", + auto_audio_preset=auto_audio_preset, + ) + + +class TestMkvmergeIdentifyStreams: + def test_detects_all_three_stream_types(self, test_mkv): + streams, _ = mkvmerge_identify_streams( + test_mkv, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + assert "video" in streams + assert "audio" in streams + assert "subtitles" in streams + + def test_each_type_has_one_stream(self, test_mkv): + streams, _ = mkvmerge_identify_streams( + test_mkv, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + assert streams["video"]["count"] == 1 + assert streams["audio"]["count"] == 1 + assert streams["subtitles"]["count"] == 1 + + def test_mapping_returned_for_first_item(self, test_mkv): + _, mapping = mkvmerge_identify_streams( + test_mkv, total_items=2, item_index=0, batch_index=1, batch_name="test" + ) + assert mapping is not None + assert set(mapping.keys()) == {"video", "audio", "subtitles"} + + def test_no_mapping_for_subsequent_items(self, test_mkv): + _, mapping = mkvmerge_identify_streams( + test_mkv, total_items=2, item_index=1, batch_index=1, batch_name="test" + ) + assert mapping is None + + def test_subtitle_id_remapped_to_stream_index(self, stream_mapping): + # Subtitle id must be the subtitle-stream index (si=), not the global track id. + # With 1 video + 1 audio track before it, the global id is 2, remapped to 0. + assert stream_mapping["subtitles"]["id"] == 0 + + def test_audio_codec_id_present(self, stream_mapping): + assert "codec_id" in stream_mapping["audio"]["properties"] + assert stream_mapping["audio"]["properties"]["codec_id"] == "A_AAC" + + +class TestFfmpegConvertFile: + def test_output_file_created(self, test_mkv, stream_mapping, tmp_path): + _convert(test_mkv, stream_mapping, tmp_path) + assert (tmp_path / "test.mp4").exists() + + def test_output_file_is_not_empty(self, test_mkv, stream_mapping, tmp_path): + _convert(test_mkv, stream_mapping, tmp_path) + assert (tmp_path / "test.mp4").stat().st_size > 0 + + def test_encoded_on_comment_present(self, test_mkv, stream_mapping, tmp_path): + _convert(test_mkv, stream_mapping, tmp_path) + tags = _ffprobe_tags(tmp_path / "test.mp4") + assert tags.get("comment", "").startswith("Encoded on ") + + def test_encoded_on_date_format(self, test_mkv, stream_mapping, tmp_path): + _convert(test_mkv, stream_mapping, tmp_path) + tags = _ffprobe_tags(tmp_path / "test.mp4") + assert re.match( + r"Encoded on \d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", tags.get("comment", "") + ) + + def test_title_metadata_matches_filename_stem( + self, test_mkv, stream_mapping, tmp_path + ): + _convert(test_mkv, stream_mapping, tmp_path) + tags = _ffprobe_tags(tmp_path / "test.mp4") + assert tags.get("title") == test_mkv.stem + + def test_extension_with_leading_dot(self, test_mkv, stream_mapping, tmp_path): + _convert(test_mkv, stream_mapping, tmp_path, extension=".mp4") + assert (tmp_path / "test.mp4").exists() + + def test_explicit_output_file_path(self, test_mkv, stream_mapping, tmp_path): + output_file = tmp_path / "custom_name.mp4" + _convert(test_mkv, stream_mapping, output_file) + assert output_file.exists() + + def test_filter_preset_before(self, test_mkv, stream_mapping, tmp_path): + filter_preset = {"before": "scale=in_color_matrix=bt709:out_color_matrix=bt601"} + _convert(test_mkv, stream_mapping, tmp_path, filter_preset=filter_preset) + assert (tmp_path / "test.mp4").exists() + + def test_filter_preset_after(self, test_mkv, stream_mapping, tmp_path): + filter_preset = {"after": "scale=in_color_matrix=bt601:out_color_matrix=bt709"} + _convert(test_mkv, stream_mapping, tmp_path, filter_preset=filter_preset) + assert (tmp_path / "test.mp4").exists() + + def test_filter_preset_before_and_after(self, test_mkv, stream_mapping, tmp_path): + filter_preset = { + "before": "scale=in_color_matrix=bt709:out_color_matrix=bt601", + "after": "scale=in_color_matrix=bt601:out_color_matrix=bt709", + } + _convert(test_mkv, stream_mapping, tmp_path, filter_preset=filter_preset) + assert (tmp_path / "test.mp4").exists() + + def test_auto_audio_aac_selects_copy_preset( + self, test_mkv, stream_mapping, tmp_path + ): + auto_preset = { + "default": {"-c:a": "aac", "-ab": "32k"}, + "copy": {"-c:a": "copy"}, + } + _convert(test_mkv, stream_mapping, tmp_path, auto_audio_preset=auto_preset) + assert (tmp_path / "test.mp4").exists() + + +def _ffprobe_audio_codec(path): + result = subprocess.run( + [ + "ffprobe", + "-v", + "quiet", + "-print_format", + "json", + "-show_streams", + "-select_streams", + "a", + str(path), + ], + capture_output=True, + check=True, + ) + return json.loads(result.stdout)["streams"][0]["codec_name"] + + +class TestASSSubtitles: + def test_identifies_subtitle_stream(self, test_mkv_ass): + streams, _ = mkvmerge_identify_streams( + test_mkv_ass, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + assert "subtitles" in streams + assert streams["subtitles"]["count"] == 1 + + def test_ass_subtitle_codec_id(self, test_mkv_ass): + streams, _ = mkvmerge_identify_streams( + test_mkv_ass, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + codec_id = streams["subtitles"]["streams"][0]["properties"]["codec_id"] + assert codec_id == "S_TEXT/ASS" + + def test_converts_to_mp4(self, test_mkv_ass, tmp_path): + _, mapping = mkvmerge_identify_streams( + test_mkv_ass, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + _convert(test_mkv_ass, mapping, tmp_path) + assert (tmp_path / "test_ass.mp4").exists() + assert (tmp_path / "test_ass.mp4").stat().st_size > 0 + + def test_encoded_on_metadata(self, test_mkv_ass, tmp_path): + _, mapping = mkvmerge_identify_streams( + test_mkv_ass, total_items=1, item_index=0, batch_index=1, batch_name="test" + ) + _convert(test_mkv_ass, mapping, tmp_path) + tags = _ffprobe_tags(tmp_path / "test_ass.mp4") + assert tags.get("comment", "").startswith("Encoded on ") + + +class TestNonAACAudioAutoPreset: + def test_vorbis_codec_id(self, test_mkv_vorbis): + _, mapping = mkvmerge_identify_streams( + test_mkv_vorbis, + total_items=1, + item_index=0, + batch_index=1, + batch_name="test", + ) + assert mapping["audio"]["properties"]["codec_id"] == "A_VORBIS" + + def test_non_aac_selects_default_preset(self, test_mkv_vorbis, tmp_path): + _, mapping = mkvmerge_identify_streams( + test_mkv_vorbis, + total_items=1, + item_index=0, + batch_index=1, + batch_name="test", + ) + auto_preset = { + "default": {"-c:a": "aac", "-ab": "32k"}, + "copy": {"-c:a": "copy"}, + } + _convert(test_mkv_vorbis, mapping, tmp_path, auto_audio_preset=auto_preset) + # Default preset re-encodes to AAC; copy would have left it as Vorbis + assert _ffprobe_audio_codec(tmp_path / "test_vorbis.mp4") == "aac" + + def test_non_aac_copy_would_fail_so_default_is_used( + self, test_mkv_vorbis, tmp_path + ): + # Confirm the output file is produced successfully with the default preset + _, mapping = mkvmerge_identify_streams( + test_mkv_vorbis, + total_items=1, + item_index=0, + batch_index=1, + batch_name="test", + ) + auto_preset = { + "default": {"-c:a": "aac", "-ab": "32k"}, + "copy": {"-c:a": "copy"}, + } + _convert(test_mkv_vorbis, mapping, tmp_path, auto_audio_preset=auto_preset) + assert (tmp_path / "test_vorbis.mp4").exists() diff --git a/tests/test_process.py b/tests/test_process.py new file mode 100644 index 0000000..565568d --- /dev/null +++ b/tests/test_process.py @@ -0,0 +1,84 @@ +from unittest.mock import MagicMock, patch + +import pytest +from loguru import logger + +from ffconv.exception import FFmpegError, MKVmergeError, ProcessError +from ffconv.process import ProcessCommand + + +def make_completed_process(returncode, stdout=b"", stderr=b""): + result = MagicMock() + result.returncode = returncode + result.stdout = stdout + result.stderr = stderr + return result + + +class TestProcessCommand: + def setup_method(self): + self.process = ProcessCommand(logger) + + def test_success_returns_completed_process(self): + mock_result = make_completed_process(0, stdout=b'{"tracks": []}') + with patch("subprocess.run", return_value=mock_result): + result = self.process.run( + "MKVmerge identify", ["mkvmerge", "--identify", "file.mkv"] + ) + assert result is mock_result + + def test_calls_subprocess_with_correct_args(self): + mock_result = make_completed_process(0) + command = ["ffmpeg", "-version"] + with patch("subprocess.run", return_value=mock_result) as mock_sp: + self.process.run("Test", command) + mock_sp.assert_called_once_with(command, capture_output=True, check=False) + + def test_mkvmerge_failure_raises_mkvmerge_error(self): + mock_result = make_completed_process(1, stderr=b"file not found") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + MKVmergeError + ): + self.process.run( + "MKVmerge identify", ["mkvmerge", "--identify", "missing.mkv"] + ) + + def test_ffmpeg_failure_raises_ffmpeg_error(self): + mock_result = make_completed_process(1, stderr=b"Invalid codec") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + FFmpegError + ): + self.process.run( + "FFmpeg convert", ["ffmpeg", "-i", "input.mkv", "output.mp4"] + ) + + def test_unknown_command_failure_raises_process_error(self): + mock_result = make_completed_process(1, stderr=b"some error") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + ProcessError + ): + self.process.run("Custom", ["custom-tool", "arg"]) + + def test_mkvmerge_error_contains_stderr_message(self): + mock_result = make_completed_process(1, stderr=b"No such file") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + MKVmergeError + ) as exc_info: + self.process.run("MKVmerge identify", ["mkvmerge", "--identify", "x.mkv"]) + assert "No such file" in str(exc_info.value) + + def test_ffmpeg_error_stores_exit_code(self): + mock_result = make_completed_process(255, stderr=b"error") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + FFmpegError + ) as exc_info: + self.process.run("FFmpeg convert", ["ffmpeg", "-i", "x.mkv", "out.mp4"]) + assert exc_info.value.exit_code == 255 + + def test_nonzero_mkvmerge_exit_code_raises(self): + for code in [1, 2, 127]: + mock_result = make_completed_process(code, stderr=b"error") + with patch("subprocess.run", return_value=mock_result), pytest.raises( + MKVmergeError + ): + self.process.run("MKVmerge", ["mkvmerge", "file.mkv"])