Skip to content

Commit

Permalink
mypy --strict no reports less than 100 errors
Browse files Browse the repository at this point in the history
  • Loading branch information
WyattBlue committed Sep 25, 2024
1 parent b558b06 commit 3a364ac
Show file tree
Hide file tree
Showing 12 changed files with 128 additions and 118 deletions.
3 changes: 3 additions & 0 deletions av/error.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
8 changes: 7 additions & 1 deletion av/frame.pyi
Original file line number Diff line number Diff line change
@@ -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: ...
4 changes: 2 additions & 2 deletions av/video/frame.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
4 changes: 2 additions & 2 deletions tests/test_codec_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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")
Expand Down
2 changes: 1 addition & 1 deletion tests/test_decode.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
10 changes: 5 additions & 5 deletions tests/test_encode.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
4 changes: 2 additions & 2 deletions tests/test_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -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!"
7 changes: 4 additions & 3 deletions tests/test_file_probing.py
Original file line number Diff line number Diff line change
Expand Up @@ -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'>"
Expand All @@ -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 "
)
Expand Down Expand Up @@ -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"):
Expand Down
7 changes: 3 additions & 4 deletions tests/test_filters.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Expand Down Expand Up @@ -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")
Expand Down
145 changes: 73 additions & 72 deletions tests/test_logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
4 changes: 2 additions & 2 deletions tests/test_python_io.py
Original file line number Diff line number Diff line change
Expand Up @@ -355,13 +355,13 @@ def test_writing_to_pipe_readonly(self) -> None:
) as cm:
write(buf)

def test_writing_to_pipe_writeonly(self):
def test_writing_to_pipe_writeonly(self) -> None:
av.logging.set_level(av.logging.VERBOSE)

buf = WriteOnlyPipe()
with pytest.raises(
ValueError, match=escape("[mp4] muxer does not support non seekable output")
) as cm:
):
write(buf)

av.logging.set_level(None)
Loading

0 comments on commit 3a364ac

Please sign in to comment.