mypy --strict now reports less than 100 errors (#1553)

Author: WyattBlue

av/error.pyi

@@ -11,6 +11,9 @@ def make_error(
     filename: str | None = None,
     log: tuple[int, tuple[int, str, str] | None] | None = None,
 ) -> None: ...
+def err_check(res: int, filename: str | None = None) -> int: ...
+
+BUFFER_TOO_SMALL: EnumItem
 
 class ErrorType(EnumItem):
     BSF_NOT_FOUND: int

av/frame.pyi

@@ -1,12 +1,18 @@
 from fractions import Fraction
+from typing import TypedDict
+
+from av.sidedata.motionvectors import MotionVectors
+
+class SideData(TypedDict, total=False):
+    MOTION_VECTORS: MotionVectors
 
 class Frame:
     dts: int | None
     pts: int | None
     time: float | None
     time_base: Fraction
     is_corrupt: bool
-    side_data: dict[str, str]
+    side_data: SideData
     opaque: object
 
     def make_writable(self) -> None: ...

av/video/frame.pyi

@@ -46,8 +46,8 @@ class VideoFrame(Frame):
         width: int | None = None,
         height: int | None = None,
         format: str | None = None,
-        src_colorspace: int | None = None,
-        dst_colorspace: int | None = None,
+        src_colorspace: str | int | None = None,
+        dst_colorspace: str | int | None = None,
         interpolation: int | str | None = None,
         src_color_range: int | str | None = None,
         dst_color_range: int | str | None = None,

tests/test_codec_context.py

@@ -169,7 +169,7 @@ def test_bits_per_coded_sample(self):
             stream.bits_per_coded_sample = 31
 
             with pytest.raises(av.error.InvalidDataError):
-                for frame in container.decode(stream):
+                for _ in container.decode(stream):
                     pass
 
         with av.open(self.sandboxed("output.mov"), "w") as output:
@@ -396,7 +396,7 @@ def video_encoding(
             assert i == gop_size
 
         final_gop_size = decoded_frame_count - max(keyframe_indices)
-        self.assertLessEqual(final_gop_size, gop_size)
+        assert final_gop_size < gop_size
 
     def test_encoding_pcm_s24le(self) -> None:
         self.audio_encoding("pcm_s24le")

tests/test_decode.py

@@ -41,7 +41,7 @@ def test_decode_audio_sample_count(self) -> None:
         audio_stream = next(s for s in container.streams if s.type == "audio")
 
         assert audio_stream is container.streams.audio[0]
-        assert isinstance(audio_stream, av.audio.AudioStream)
+        assert isinstance(audio_stream, av.AudioStream)
 
         sample_count = 0
 

tests/test_encode.py

@@ -1,9 +1,9 @@
 import io
 import math
 from fractions import Fraction
-from unittest import SkipTest
 
 import numpy as np
+import pytest
 
 import av
 from av import AudioFrame, VideoFrame
@@ -19,7 +19,7 @@
 
 def write_rgb_rotate(output: av.container.OutputContainer) -> None:
     if not has_pillow:
-        raise SkipTest("Don't have Pillow")
+        pytest.skip()
 
     import PIL.Image as Image
 
@@ -233,8 +233,8 @@ def test_stream_index(self) -> None:
 
             astream = output.add_stream("mp2", 48000)
             assert astream in output.streams.audio
-            astream.layout = "stereo"  # type: ignore
-            astream.format = "s16"  # type: ignore
+            astream.layout = "stereo"
+            astream.format = "s16"
 
             assert vstream.index == 0
             assert astream.index == 1
@@ -385,4 +385,4 @@ def test_max_b_frames(self) -> None:
         for max_b_frames in range(4):
             file = encode_file_with_max_b_frames(max_b_frames)
             actual_max_b_frames = max_b_frame_run_in_file(file)
-            self.assertTrue(actual_max_b_frames <= max_b_frames)
+            assert actual_max_b_frames <= max_b_frames

tests/test_errors.py

@@ -53,11 +53,11 @@ def test_filenotfound():
         assert False, "No exception raised!"
 
 
-def test_buffertoosmall():
+def test_buffertoosmall() -> None:
     """Throw an exception from an enum."""
     try:
         av.error.err_check(-av.error.BUFFER_TOO_SMALL.value)
-    except av.BufferTooSmallError as e:
+    except av.error.BufferTooSmallError as e:
         assert e.errno == av.error.BUFFER_TOO_SMALL.value
     else:
         assert False, "No exception raised!"

tests/test_file_probing.py

@@ -9,7 +9,7 @@ class TestAudioProbe(TestCase):
     def setUp(self):
         self.file = av.open(fate_suite("aac/latm_stereo_to_51.ts"))
 
-    def test_container_probing(self):
+    def test_container_probing(self) -> None:
         assert self.file.bit_rate == 269558
         assert self.file.duration == 6165333
         assert str(self.file.format) == "<av.ContainerFormat 'mpegts'>"
@@ -20,9 +20,10 @@ def test_container_probing(self):
         assert self.file.start_time == 1400000
         assert len(self.file.streams) == 1
 
-    def test_stream_probing(self):
+    def test_stream_probing(self) -> None:
         stream = self.file.streams[0]
 
+        assert isinstance(stream, av.AudioStream)
         assert str(stream).startswith(
             "<av.AudioStream #0 aac_latm at 48000Hz, stereo, fltp at "
         )
@@ -52,7 +53,7 @@ def test_stream_probing(self):
 
 
 class TestAudioProbeCorrupt(TestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         # write an empty file
         path = self.sandboxed("empty.flac")
         with open(path, "wb"):

tests/test_filters.py

@@ -126,7 +126,7 @@ def test_audio_buffer_sink(self):
             if e.errno != errno.EAGAIN:
                 raise
 
-    def test_audio_buffer_resample(self):
+    def test_audio_buffer_resample(self) -> None:
         graph = Graph()
         graph.link_nodes(
             graph.add_abuffer(
@@ -147,6 +147,7 @@ def test_audio_buffer_resample(self):
             )
         )
         out_frame = graph.pull()
+        assert isinstance(out_frame, av.AudioFrame)
         assert out_frame.format.name == "s16"
         assert out_frame.layout.name == "stereo"
         assert out_frame.sample_rate == 44100
@@ -202,9 +203,7 @@ def test_audio_buffer_volume_filter(self):
         input_data = input_frame.to_ndarray()
         output_data = out_frame.to_ndarray()
 
-        self.assertTrue(
-            np.allclose(input_data * 0.5, output_data), "Check that volume is reduced"
-        )
+        assert np.allclose(input_data * 0.5, output_data)
 
     def test_video_buffer(self):
         input_container = av.open(format="lavfi", file="color=c=pink:duration=1:r=30")

tests/test_logging.py

@@ -5,80 +5,81 @@
 import av.error
 import av.logging
 
-from .common import TestCase
-
 
 def do_log(message: str) -> None:
     av.logging.log(av.logging.INFO, "test", message)
 
 
-class TestLogging(TestCase):
-    def test_adapt_level(self):
-        assert av.logging.adapt_level(av.logging.ERROR) == logging.ERROR
-        assert av.logging.adapt_level(av.logging.WARNING) == logging.WARNING
-        assert (
-            av.logging.adapt_level((av.logging.WARNING + av.logging.ERROR) // 2)
-            == logging.WARNING
-        )
-
-    def test_threaded_captures(self):
-        av.logging.set_level(av.logging.VERBOSE)
-
-        with av.logging.Capture(local=True) as logs:
-            do_log("main")
-            thread = threading.Thread(target=do_log, args=("thread",))
-            thread.start()
-            thread.join()
-
-        assert (av.logging.INFO, "test", "main") in logs
-        av.logging.set_level(None)
-
-    def test_global_captures(self):
-        av.logging.set_level(av.logging.VERBOSE)
-
-        with av.logging.Capture(local=False) as logs:
-            do_log("main")
-            thread = threading.Thread(target=do_log, args=("thread",))
-            thread.start()
-            thread.join()
-
-        assert (av.logging.INFO, "test", "main") in logs
-        assert (av.logging.INFO, "test", "thread") in logs
-        av.logging.set_level(None)
-
-    def test_repeats(self):
-        av.logging.set_level(av.logging.VERBOSE)
-
-        with av.logging.Capture() as logs:
-            do_log("foo")
-            do_log("foo")
-            do_log("bar")
-            do_log("bar")
-            do_log("bar")
-            do_log("baz")
-
-        logs = [log for log in logs if log[1] == "test"]
-
-        assert logs == [
-            (av.logging.INFO, "test", "foo"),
-            (av.logging.INFO, "test", "foo"),
-            (av.logging.INFO, "test", "bar"),
-            (av.logging.INFO, "test", "bar (repeated 2 more times)"),
-            (av.logging.INFO, "test", "baz"),
-        ]
-
-        av.logging.set_level(None)
-
-    def test_error(self):
-        av.logging.set_level(av.logging.VERBOSE)
-
-        log = (av.logging.ERROR, "test", "This is a test.")
-        av.logging.log(*log)
-        try:
-            av.error.err_check(-errno.EPERM)
-        except OSError as e:
-            assert e.log == log
-        else:
-            self.fail()
-
-        av.logging.set_level(None)
+def test_adapt_level() -> None:
+    assert av.logging.adapt_level(av.logging.ERROR) == logging.ERROR
+    assert av.logging.adapt_level(av.logging.WARNING) == logging.WARNING
+    assert (
+        av.logging.adapt_level((av.logging.WARNING + av.logging.ERROR) // 2)
+        == logging.WARNING
+    )
+
+
+def test_threaded_captures() -> None:
+    av.logging.set_level(av.logging.VERBOSE)
+
+    with av.logging.Capture(local=True) as logs:
+        do_log("main")
+        thread = threading.Thread(target=do_log, args=("thread",))
+        thread.start()
+        thread.join()
+
+    assert (av.logging.INFO, "test", "main") in logs
+    av.logging.set_level(None)
+
+
+def test_global_captures() -> None:
+    av.logging.set_level(av.logging.VERBOSE)
+
+    with av.logging.Capture(local=False) as logs:
+        do_log("main")
+        thread = threading.Thread(target=do_log, args=("thread",))
+        thread.start()
+        thread.join()
+
+    assert (av.logging.INFO, "test", "main") in logs
+    assert (av.logging.INFO, "test", "thread") in logs
+    av.logging.set_level(None)
+
+
+def test_repeats() -> None:
+    av.logging.set_level(av.logging.VERBOSE)
+
+    with av.logging.Capture() as logs:
+        do_log("foo")
+        do_log("foo")
+        do_log("bar")
+        do_log("bar")
+        do_log("bar")
+        do_log("baz")
+
+    logs = [log for log in logs if log[1] == "test"]
+
+    assert logs == [
+        (av.logging.INFO, "test", "foo"),
+        (av.logging.INFO, "test", "foo"),
+        (av.logging.INFO, "test", "bar"),
+        (av.logging.INFO, "test", "bar (repeated 2 more times)"),
+        (av.logging.INFO, "test", "baz"),
+    ]
+
+    av.logging.set_level(None)
+
+
+def test_error() -> None:
+    av.logging.set_level(av.logging.VERBOSE)
+
+    log = (av.logging.ERROR, "test", "This is a test.")
+    av.logging.log(*log)
+    try:
+        av.error.err_check(-errno.EPERM)
+    except av.error.PermissionError as e:
+        assert e.log == log
+    else:
+        assert False
+
+    av.logging.set_level(None)