from pathlib import Path import pytest from ffwrap.command import ( Options, build_command, match_audio_selection, plan_output_path, scale_filter, temp_output_path, ) from ffwrap.probe import MediaInfo, Stream def video(codec="h264", w=1920, h=1080, rotation=0, index=0): return Stream(index=index, kind="video", codec=codec, width=w, height=h, fps=23.976, rotation=rotation) def audio(index, codec="aac", lang="eng", ch=2, title=None, default=False, rate=48000): return Stream(index=index, kind="audio", codec=codec, language=lang, channels=ch, title=title, default=default, sample_rate=rate) def sub(index, codec="subrip", lang="eng"): return Stream(index=index, kind="subtitle", codec=codec, language=lang) def media(*streams, path="/videos/movie.mkv"): return MediaInfo(path=Path(path), duration=100.0, format_name="matroska", streams=list(streams)) def build(info, audio_idx=None, **opts): o = Options(**opts) idx = [s.index for s in info.audio] if audio_idx is None else audio_idx return build_command("ffmpeg", info, o, idx, Path("/out/movie.mp4")) def pairs(args): """Return the args as a set of 'flag value' strings for easy assertions.""" return {f"{a} {b}" for a, b in zip(args, args[1:])} # ---------------------------------------------------------------- remuxing def test_jellyfin_remux_mkv_to_mp4(): info = media(video("hevc"), audio(1, "ac3", ch=6), audio(2, "dts", ch=6), sub(3), sub(4, "hdmv_pgs_subtitle")) b = build(info, container="mp4", video_mode="copy", audio_mode="auto", subtitles=True) p = pairs(b.args) assert "-c:v copy" in p assert "-tag:v hvc1" in p # Apple-friendly HEVC tag assert "-c:a:0 copy" in p # AC3 is fine in MP4 assert "-c:a:1 aac" in p # DTS gets converted assert "-c:s:0 mov_text" in p # SRT -> mov_text assert "-map 0:4" not in p # PGS dropped for MP4 assert any("dropped" in n for n in b.notes) assert "-movflags +faststart" in p assert b.args[-1] == str(Path("/out/movie.mp4")) def test_mkv_keeps_everything(): info = media(video(), audio(1, "truehd", ch=8), sub(2, "hdmv_pgs_subtitle"), sub(3, "ass"), Stream(index=4, kind="attachment", codec="ttf")) b = build(info, container="mkv", video_mode="copy", audio_mode="auto") p = pairs(b.args) assert {"-c:v copy", "-c:a:0 copy", "-c:s:0 copy", "-c:s:1 copy", "-map 0:t?", "-c:t copy"} <= p assert "-movflags +faststart" not in p def test_copy_mode_warns_about_incompatible_codec(): info = media(video("vp9"), audio(1, "opus")) b = build(info, container="avi", video_mode="copy", audio_mode="copy") assert any("vp9" in n and "Warning" in n for n in b.notes) def test_auto_video_reencodes_only_when_needed(): info = media(video("vp9"), audio(1)) assert "-c:v copy" in pairs(build(info, container="mkv", video_mode="auto").args) b = build(info, container="mp4", video_mode="auto") # vp9 is OK in MP4 too assert "-c:v copy" in pairs(b.args) b = build(info, container="mov", video_mode="auto", video_encoder="libx264") assert "-c:v libx264" in pairs(b.args) b = build(info, container="mkv", video_mode="auto", resolution=720) assert "-c:v libx264" in pairs(b.args) def test_cover_art_is_not_mapped_as_video(): cover = Stream(index=0, kind="video", codec="mjpeg", width=500, height=500, attached_pic=True) info = media(cover, video("h264", index=1), audio(2)) b = build(info, container="mp4") assert "-map 0:1" in pairs(b.args) assert "-map 0:0" not in pairs(b.args) # --------------------------------------------------------------- encoding def test_reencode_h264_with_scaling_and_fps(): info = media(video("hevc", 3840, 2160), audio(1)) b = build(info, container="mp4", video_mode="encode", video_encoder="libx264", resolution=1080, fps="29.97", quality=21, speed=3) p = pairs(b.args) assert "-vf scale=-2:1080:flags=lanczos,fps=30000/1001" in p assert {"-c:v libx264", "-crf 21", "-preset slow", "-pix_fmt yuv420p"} <= p def test_quality_is_clamped_to_encoder_range(): info = media(video(), audio(1)) b = build(info, video_mode="encode", video_encoder="mpeg4", container="avi", quality=60) assert "-q:v 31" in pairs(b.args) def test_strip_audio(): info = media(video(), audio(1), audio(2)) b = build(info, audio_mode="strip") assert "-an" in b.args assert not any(a.startswith("-c:a") for a in b.args) assert "-map 0:1" not in pairs(b.args) def test_selected_audio_tracks_in_order_with_default_disposition(): info = media(video(), audio(1, lang="jpn"), audio(2, lang="eng"), audio(3, lang="eng", title="Commentary")) b = build(info, audio_idx=[2, 1], audio_mode="copy") maps = [b.args[i + 1] for i, a in enumerate(b.args) if a == "-map"] assert maps == ["0:0", "0:2", "0:1"] p = pairs(b.args) assert {"-disposition:a:0 default", "-disposition:a:1 0"} <= p def test_audio_downmix_resample_and_bitrate(): info = media(video(), audio(1, "dts", ch=6, rate=48000)) b = build(info, audio_mode="encode", audio_encoder="aac", audio_channels=2, audio_sample_rate=44100, audio_bitrate=160) assert {"-c:a:0 aac", "-ac:a:0 2", "-ar:a:0 44100", "-b:a:0 160k"} <= pairs(b.args) def test_auto_audio_reencodes_when_channels_change(): info = media(video(), audio(1, "aac", ch=6)) b = build(info, audio_mode="auto", audio_channels=2) assert "-c:a:0 aac" in pairs(b.args) b = build(info, audio_mode="auto") assert "-c:a:0 copy" in pairs(b.args) def test_audio_encoder_limits(): info = media(video(), audio(1, "truehd", ch=8)) b = build(info, container="mkv", audio_mode="encode", audio_encoder="libmp3lame", audio_sample_rate=96000) p = pairs(b.args) assert "-ac:a:0 2" in p # MP3 is stereo only assert "-ar:a:0 48000" in p # nearest supported rate b = build(info, container="mkv", audio_mode="encode", audio_encoder="libopus") assert "-mapping_family:a:0 1" in pairs(b.args) b = build(info, container="mkv", audio_mode="encode", audio_encoder="flac") assert not any(a.startswith("-b:a") for a in b.args) # ---------------------------------------------------------------- scaling @pytest.mark.parametrize("w,h,rot,res,expected", [ (1920, 1080, 0, 720, "scale=-2:720:flags=lanczos"), (1080, 1920, 0, 720, "scale=720:-2:flags=lanczos"), # portrait (1920, 1080, 90, 720, "scale=720:-2:flags=lanczos"), # rotated phone video (1920, 1080, 0, 1080, None), # already that size (1280, 720, 0, 2160, "scale=-2:2160:flags=lanczos"), # upscale allowed (1920, 1080, 0, 0, None), ]) def test_scale_filter(w, h, rot, res, expected): assert scale_filter(video(w=w, h=h, rotation=rot), Options(resolution=res)) == expected def test_scale_no_upscale_and_custom(): v = video(w=1280, h=720) assert scale_filter(v, Options(resolution=1080, no_upscale=True)) is None assert scale_filter(v, Options(resolution=480, no_upscale=True)) == "scale=-2:480:flags=lanczos" custom = scale_filter(v, Options(resolution=-1, custom_width=640, custom_height=640)) assert custom.startswith("scale=640:640:force_original_aspect_ratio=decrease") assert scale_filter(v, Options(resolution=-1, custom_width=853)) == "scale=852:-2:flags=lanczos" assert scale_filter(v, Options(resolution=-1)) is None # ------------------------------------------------------------ output paths def test_output_next_to_source(tmp_path): src = tmp_path / "a.mkv" assert plan_output_path(src, Options(container="mp4"), set()) == tmp_path / "a.mp4" def test_output_never_overwrites_source(tmp_path): src = tmp_path / "a.mp4" src.touch() out = plan_output_path(src, Options(container="mp4", if_exists="overwrite"), set()) assert out == tmp_path / "a_converted.mp4" def test_output_exists_policies(tmp_path): src = tmp_path / "a.mkv" (tmp_path / "a.mp4").touch() assert plan_output_path(src, Options(if_exists="skip"), set()) is None assert plan_output_path(src, Options(if_exists="overwrite"), set()) == tmp_path / "a.mp4" assert plan_output_path(src, Options(if_exists="rename"), set()) == tmp_path / "a (1).mp4" def test_output_custom_dir_and_batch_collisions(tmp_path): out_dir = tmp_path / "out" opts = Options(output_dir=str(out_dir), suffix="_web", if_exists="overwrite") first = plan_output_path(tmp_path / "s1" / "ep.mkv", opts, set()) assert first == out_dir / "ep_web.mp4" taken = {__import__("os").path.normcase(str(first))} # Same name from another folder in the same batch must not clobber the first one. assert plan_output_path(tmp_path / "s2" / "ep.mkv", opts, taken) == out_dir / "ep_web (1).mp4" def test_temp_output_path(): assert temp_output_path(Path("/x/movie.mp4")) == Path("/x/movie.part.mp4") # ----------------------------------------------------------- track matching def test_match_identical_layout_by_position(): ref = [audio(1, lang="jpn"), audio(2, lang="eng")] tgt = [audio(3, lang="jpn"), audio(4, lang="eng")] assert match_audio_selection(ref, [2], tgt) == [4] def test_match_by_language_when_layout_differs(): ref = [audio(1, "ac3", "jpn", 6), audio(2, "aac", "eng", 2)] tgt = [audio(1, "aac", "eng", 2), audio(2, "ac3", "jpn", 6), audio(3, "aac", "eng", 2, title="Commentary")] assert match_audio_selection(ref, [2, 1], tgt) == [1, 2] def test_match_falls_back_to_default_track(): ref = [audio(1, lang="eng")] tgt = [audio(1, lang="fre"), audio(2, lang="ger", default=True)] assert match_audio_selection(ref, [1], tgt) == [2] def test_match_empty_selection(): assert match_audio_selection([audio(1)], [], [audio(1)]) == [] assert match_audio_selection([audio(1)], [1], []) == [] def test_options_roundtrip(): o = Options(container="mkv", quality=18, no_upscale=True, output_dir="/tmp/x") assert Options.from_dict(o.to_dict()) == o assert Options.from_dict({"quality": "bad", "unknown": 1}) == Options()