Improve test coverage from 79% to 80% (#811)

Author: Copilot

coverage.json

@@ -168,7 +168,7 @@ ignore = [
 ]
 
 [tool.ruff.lint.per-file-ignores]
-"**/tests/**.py" = ["D103", "D104", "D100"]
+"**/tests/**.py" = ["D103", "D104", "D100", "DOC501"]
 
 [tool.ruff.format]
 quote-style = "double"

pyproject.toml

@@ -0,0 +1,164 @@
+"""Tests for FFmpeg filter schema definitions."""
+
+from ...common.schema import (
+    FFMpegFilter,
+    FFMpegIOType,
+    StreamType,
+)
+
+
+def test_filter_pre_dict_property() -> None:
+    """Test that pre_dict property converts pre list to dict."""
+    filter_def = FFMpegFilter(
+        name="scale",
+        description="Scale video",
+        pre=(("width", "1280"), ("height", "720")),
+        stream_typings_input=(),
+        stream_typings_output=(),
+    )
+
+    pre_dict = filter_def.pre_dict
+    assert isinstance(pre_dict, dict)
+    assert pre_dict["width"] == "1280"
+    assert pre_dict["height"] == "720"
+
+
+def test_filter_to_def_property() -> None:
+    """Test that to_def property creates FFMpegFilterDef."""
+    from ...common.schema import FFMpegFilterDef
+
+    filter_def = FFMpegFilter(
+        name="scale",
+        description="Scale video",
+        stream_typings_input=(FFMpegIOType(name="default", type=StreamType.video),),
+        stream_typings_output=(FFMpegIOType(name="default", type=StreamType.video),),
+    )
+
+    simple_def = filter_def.to_def
+    assert isinstance(simple_def, FFMpegFilterDef)
+    assert simple_def.name == "scale"
+    assert simple_def.typings_input == ("video",)
+    assert simple_def.typings_output == ("video",)
+
+
+def test_filter_input_typings_source() -> None:
+    """Test input_typings for filter source (no inputs)."""
+    filter_def = FFMpegFilter(
+        name="color",
+        description="Generate color",
+        stream_typings_input=(),
+        stream_typings_output=(),
+        is_filter_source=True,
+    )
+
+    typings = filter_def.input_typings
+    assert isinstance(typings, set)
+    assert len(typings) == 0
+
+
+def test_filter_input_typings_static() -> None:
+    """Test input_typings for static input filter."""
+    filter_def = FFMpegFilter(
+        name="scale",
+        description="Scale video",
+        stream_typings_input=(FFMpegIOType(name="default", type=StreamType.video),),
+        stream_typings_output=(),
+    )
+
+    typings = filter_def.input_typings
+    assert StreamType.video in typings
+    assert len(typings) == 1
+
+
+def test_filter_input_typings_dynamic_video() -> None:
+    """Test input_typings for dynamic input filter with video only."""
+    filter_def = FFMpegFilter(
+        name="concat",
+        description="Concatenate videos",
+        stream_typings_input=(),
+        stream_typings_output=(),
+        is_dynamic_input=True,
+        formula_typings_input="video",
+    )
+
+    typings = filter_def.input_typings
+    assert StreamType.video in typings
+    assert StreamType.audio not in typings
+
+
+def test_filter_input_typings_dynamic_audio() -> None:
+    """Test input_typings for dynamic input filter with audio only."""
+    filter_def = FFMpegFilter(
+        name="amix",
+        description="Mix audio",
+        stream_typings_input=(),
+        stream_typings_output=(),
+        is_dynamic_input=True,
+        formula_typings_input="audio",
+    )
+
+    typings = filter_def.input_typings
+    assert StreamType.audio in typings
+    assert StreamType.video not in typings
+
+
+def test_filter_input_typings_dynamic_both() -> None:
+    """Test input_typings for dynamic input filter with both video and audio."""
+    filter_def = FFMpegFilter(
+        name="amerge",
+        description="Merge audio and video",
+        stream_typings_input=(),
+        stream_typings_output=(),
+        is_dynamic_input=True,
+        formula_typings_input="videoaudio",
+    )
+
+    typings = filter_def.input_typings
+    assert StreamType.video in typings
+    assert StreamType.audio in typings
+
+
+def test_filter_output_typings_static() -> None:
+    """Test output_typings for static output filter."""
+    filter_def = FFMpegFilter(
+        name="scale",
+        description="Scale video",
+        stream_typings_input=(),
+        stream_typings_output=(FFMpegIOType(name="default", type=StreamType.video),),
+    )
+
+    typings = filter_def.output_typings
+    assert StreamType.video in typings
+    assert len(typings) == 1
+
+
+def test_filter_output_typings_multiple() -> None:
+    """Test output_typings for filter with multiple outputs."""
+    filter_def = FFMpegFilter(
+        name="split",
+        description="Split video",
+        stream_typings_input=(),
+        stream_typings_output=(
+            FFMpegIOType(name="output1", type=StreamType.video),
+            FFMpegIOType(name="output2", type=StreamType.video),
+        ),
+    )
+
+    typings = filter_def.output_typings
+    assert StreamType.video in typings
+    assert len(typings) == 1  # Both outputs are video, so set has 1 element
+
+
+def test_filter_output_typings_dynamic() -> None:
+    """Test output_typings for dynamic output filter."""
+    filter_def = FFMpegFilter(
+        name="split",
+        description="Split stream",
+        stream_typings_input=(),
+        stream_typings_output=(),
+        is_dynamic_output=True,
+        formula_typings_output="video",
+    )
+
+    typings = filter_def.output_typings
+    assert StreamType.video in typings

src/ffmpeg/common/tests/test_schema.py

@@ -343,3 +343,69 @@ async def test_run_async_awaitable_is_awaitable(self) -> None:
         process = await result
         assert isinstance(process, asyncio.subprocess.Process)
         await process.wait()
+
+
+class TestMergeOutputs:
+    """Tests for the merge_outputs method."""
+
+    def test_merge_outputs_basic(self) -> None:
+        """Test that merge_outputs combines multiple output streams."""
+        video = input("input.mp4").video
+        output1 = video.output(filename="output1.mp4")
+        output2 = video.output(filename="output2.webm")
+
+        merged = output1.merge_outputs(output2)
+
+        # Verify that the merged output is a GlobalStream
+        from ...nodes import GlobalStream
+
+        assert isinstance(merged, GlobalStream)
+
+        # Verify compilation includes both outputs
+        args = merged.compile()
+        assert "output1.mp4" in args
+        assert "output2.webm" in args
+
+    def test_merge_outputs_multiple(self) -> None:
+        """Test that merge_outputs can handle multiple streams."""
+        video = input("input.mp4").video
+        output1 = video.output(filename="out1.mp4")
+        output2 = video.output(filename="out2.mp4")
+        output3 = video.output(filename="out3.mp4")
+
+        merged = output1.merge_outputs(output2, output3)
+
+        # Verify all three outputs are in the compiled command
+        args = merged.compile()
+        assert "out1.mp4" in args
+        assert "out2.mp4" in args
+        assert "out3.mp4" in args
+
+
+class TestOverwriteOutput:
+    """Tests for the overwrite_output method."""
+
+    def test_overwrite_output_adds_y_flag(self) -> None:
+        """Test that overwrite_output adds the -y flag."""
+        stream = input("input.mp4").output(filename="output.mp4")
+        overwrite_stream = stream.overwrite_output()
+
+        # Verify the compiled command includes -y
+        args = overwrite_stream.compile()
+        assert "-y" in args
+
+
+class TestRunWithInput:
+    """Tests for run method with input parameter."""
+
+    def test_run_with_stdin_input(self) -> None:
+        """Test that run method accepts input parameter."""
+        import inspect
+
+        stream = input("test.mp4").output(filename="output.mp4")
+        sig = inspect.signature(stream.run)
+
+        assert "input" in sig.parameters
+        # Check the type annotation
+        param = sig.parameters["input"]
+        assert param.default is None

src/ffmpeg/dag/global_runnable/tests/test_runnable.py

@@ -35,6 +35,17 @@ def test_get_actual_type_nested_union() -> None:
     )  # Should return the first non-None type
 
 
+def test_get_actual_type_types_union() -> None:
+    """Test _get_actual_type with types.UnionType (Python 3.10+)."""
+    import types
+
+    # Test that it can handle types.UnionType properly
+    if hasattr(types, "UnionType"):
+        # Create a union type using the | operator (Python 3.10+)
+        union_type = int | str | None
+        assert _get_actual_type(union_type) is int
+
+
 def test_get_actual_type_string_annotation() -> None:
     # Simulate string annotation as might be found in dataclasses
     assert _get_actual_type("int") is int

src/ffmpeg/ffprobe/tests/test_parse.py

@@ -0,0 +1,43 @@
+"""Tests for audio channel layout definitions."""
+
+from ..channel_layout import CHANNEL_LAYOUT
+
+
+def test_channel_layout_basic() -> None:
+    """Test basic channel layout definitions."""
+    assert CHANNEL_LAYOUT["mono"] == 1
+    assert CHANNEL_LAYOUT["stereo"] == 2
+
+
+def test_channel_layout_surround() -> None:
+    """Test surround sound channel layouts."""
+    assert CHANNEL_LAYOUT["5.1"] == 6
+    assert CHANNEL_LAYOUT["7.1"] == 8
+
+
+def test_channel_layout_complex() -> None:
+    """Test complex channel layouts."""
+    assert CHANNEL_LAYOUT["5.1.2"] == 8
+    assert CHANNEL_LAYOUT["7.1.4"] == 12
+    assert CHANNEL_LAYOUT["22.2"] == 24
+
+
+def test_channel_layout_special() -> None:
+    """Test special channel layouts."""
+    assert CHANNEL_LAYOUT["quad"] == 4
+    assert CHANNEL_LAYOUT["hexagonal"] == 6
+    assert CHANNEL_LAYOUT["octagonal"] == 8
+    assert CHANNEL_LAYOUT["hexadecagonal"] == 16
+
+
+def test_channel_layout_downmix() -> None:
+    """Test downmix channel layout."""
+    assert CHANNEL_LAYOUT["downmix"] == 2
+
+
+def test_channel_layout_all_defined() -> None:
+    """Test that all channel layouts are defined with valid channel counts."""
+    for layout, channels in CHANNEL_LAYOUT.items():
+        assert isinstance(layout, str)
+        assert isinstance(channels, int)
+        assert channels > 0