[build] Deflake VideoStream tests

Author: Breakthrough

tests/conftest.py

@@ -161,14 +161,41 @@ def delayed_start_video() -> str:
     return check_exists("tests/resources/delayed_start.mp4")
 
 
+@pytest.fixture
+def auto_close():
+    """Registers VideoStreams (or anything closeable) for deterministic cleanup at test end.
+
+    Usage: ``video = auto_close(open_video(path))``. Returns its argument unchanged.
+    Closing test-owned streams while the interpreter is healthy avoids ResourceWarnings
+    (unclosed PyAV containers / file handles) finalizing during interpreter shutdown,
+    where native teardown can crash the process exit code (windows-latest CI flake).
+    """
+    from tests.helpers import close_video_stream
+
+    streams = []
+
+    def _register(stream):
+        streams.append(stream)
+        return stream
+
+    yield _register
+    for stream in streams:
+        close_video_stream(stream)
+
+
 def pytest_unconfigure(config):
     """Diagnostic for a windows-latest CI flake (silent exit 1 after a green run):
     report any non-main threads still alive at session end. Leaked threads keep
     VideoStreams alive into interpreter shutdown, where native teardown can crash
     the process exit code. tqdm's global monitor singleton is expected and ignored."""
+    import gc
     import sys
     import threading
 
+    # Finalize any lingering test-owned objects (e.g. av containers kept alive by reference
+    # cycles) while the interpreter is still healthy, instead of at interpreter shutdown.
+    gc.collect()
+
     leftover = [
         t
         for t in threading.enumerate()

tests/helpers.py

@@ -11,13 +11,45 @@
 #
 """Shared test helpers."""
 
+import contextlib
+import typing as ty
+
 from click.testing import CliRunner
 
 from scenedetect._cli import scenedetect as _scenedetect_cli
 from scenedetect._cli.context import CliContext
 from scenedetect._cli.controller import run_scenedetect
 
 
+def close_video_stream(stream: ty.Any) -> None:
+    """Deterministically release a VideoStream's native resources.
+
+    `VideoStream` has no public close()/context-manager API, so tests release the
+    backend-specific handles directly. Closing while the interpreter is healthy avoids
+    ResourceWarnings (and native teardown work) at interpreter shutdown. Safe to call
+    multiple times; never raises.
+    """
+    backend = getattr(stream, "BACKEND_NAME", None)
+    if backend == "pyav":
+        # Close the decode generator first to break its cycle with the container. `_io` is
+        # the file handle backing the container (opened by the stream when given a path).
+        for attr in ("_decoder", "_container", "_io"):
+            handle = getattr(stream, attr, None)
+            if handle is not None:
+                with contextlib.suppress(Exception):
+                    handle.close()
+    elif backend == "opencv":
+        cap = getattr(stream, "_cap", None)
+        if cap is not None:
+            with contextlib.suppress(Exception):
+                cap.release()
+    elif backend == "moviepy":
+        reader = getattr(stream, "_reader", None)
+        if reader is not None:
+            with contextlib.suppress(Exception):
+                reader.close()
+
+
 def invoke_cli(args: list[str], catch_exceptions: bool = False) -> tuple[int, str]:
     """Invoke the scenedetect CLI in-process using Click's CliRunner.
 
@@ -30,7 +62,15 @@ def invoke_cli(args: list[str], catch_exceptions: bool = False) -> tuple[int, st
     """
     context = CliContext()
     runner = CliRunner()
-    result = runner.invoke(_scenedetect_cli, args, obj=context, catch_exceptions=catch_exceptions)
-    if result.exit_code == 0:
-        run_scenedetect(context)
-    return result.exit_code, result.output
+    try:
+        result = runner.invoke(
+            _scenedetect_cli, args, obj=context, catch_exceptions=catch_exceptions
+        )
+        if result.exit_code == 0:
+            run_scenedetect(context)
+        return result.exit_code, result.output
+    finally:
+        # The CLI opens a VideoStream on `context` and has no teardown path; close it here so
+        # its native handles are released deterministically instead of at interpreter shutdown.
+        if context.video_stream is not None:
+            close_video_stream(context.video_stream)

tests/test_backend_pyav.py

@@ -22,11 +22,11 @@
 from scenedetect.backends.pyav import MAX_CONSECUTIVE_DECODE_FAILURES, VideoStreamAv
 
 
-def test_video_stream_pyav_bytesio(test_video_file: str):
+def test_video_stream_pyav_bytesio(test_video_file: str, auto_close):
     """Test that VideoStreamAv works with a BytesIO input in addition to a path."""
     # Mode must be binary!
     with open(test_video_file, mode="rb") as video_file:
-        stream = VideoStreamAv(path_or_io=video_file, threading_mode=None)
+        stream = auto_close(VideoStreamAv(path_or_io=video_file, threading_mode=None))
         assert stream.is_seekable
         stream.seek(50)
         for _ in range(10):
@@ -56,9 +56,9 @@ def __getattr__(self, name):
         return getattr(self._container, name)
 
 
-def test_read_tolerates_corrupt_frame(test_video_file: str):
+def test_read_tolerates_corrupt_frame(test_video_file: str, auto_close):
     """A decode error partway through the stream must be skipped, not stop decoding."""
-    stream = VideoStreamAv(test_video_file)
+    stream = auto_close(VideoStreamAv(test_video_file))
     injected = False
 
     def fault_injecting_decode(container, *args, **kwargs):
@@ -76,9 +76,9 @@ def fault_injecting_decode(container, *args, **kwargs):
     assert stream.decode_failures == 1
 
 
-def test_read_gives_up_after_consecutive_failures(test_video_file: str, caplog):
+def test_read_gives_up_after_consecutive_failures(test_video_file: str, caplog, auto_close):
     """After too many consecutive decode failures, read() must return False, not hang."""
-    stream = VideoStreamAv(test_video_file)
+    stream = auto_close(VideoStreamAv(test_video_file))
 
     def always_failing_decode(container, *args, **kwargs):
         raise _make_invalid_data_error()

tests/test_vfr.py

@@ -52,16 +52,16 @@ def _tc_to_secs(tc: str) -> float:
     return int(h) * 3600 + int(m) * 60 + int(s) + int(ms) / 1000
 
 
-def test_vfr_position_is_timecode(test_vfr_video: str):
+def test_vfr_position_is_timecode(test_vfr_video: str, auto_close):
     """Position should be a Timecode-backed FrameTimecode."""
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     assert video.read() is not False
     assert isinstance(video.position._time, Timecode)
 
 
-def test_vfr_position_monotonic_pyav(test_vfr_video: str):
+def test_vfr_position_monotonic_pyav(test_vfr_video: str, auto_close):
     """PTS-based position should be monotonically non-decreasing (PyAV)."""
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     last_seconds = -1.0
     frame_count = 0
     while True:
@@ -77,9 +77,9 @@ def test_vfr_position_monotonic_pyav(test_vfr_video: str):
     assert frame_count > 0
 
 
-def test_vfr_position_monotonic_opencv(test_vfr_video: str):
+def test_vfr_position_monotonic_opencv(test_vfr_video: str, auto_close):
     """PTS-based position should be monotonically non-decreasing (OpenCV)."""
-    video = open_video(test_vfr_video, backend="opencv")
+    video = auto_close(open_video(test_vfr_video, backend="opencv"))
     last_seconds = -1.0
     frame_count = 0
     while True:
@@ -96,13 +96,13 @@ def test_vfr_position_monotonic_opencv(test_vfr_video: str):
 
 
 @pytest.mark.parametrize("backend", ["pyav", "opencv"])
-def test_vfr_scene_detection(test_vfr_video: str, backend: str):
+def test_vfr_scene_detection(test_vfr_video: str, backend: str, auto_close):
     """Scene detection on VFR video should produce timestamps matching known ground truth.
 
     Both PyAV (native PTS) and OpenCV (CAP_PROP_POS_MSEC) should agree on scene cuts since
     both expose accurate PTS-derived timestamps.
     """
-    video = open_video(test_vfr_video, backend=backend)
+    video = auto_close(open_video(test_vfr_video, backend=backend))
     sm = SceneManager()
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video, end_time=10.0)
@@ -123,9 +123,9 @@ def test_vfr_scene_detection(test_vfr_video: str, backend: str):
         )
 
 
-def test_vfr_seek_pyav(test_vfr_video: str):
+def test_vfr_seek_pyav(test_vfr_video: str, auto_close):
     """Seeking should work with VFR video."""
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     target_time = 2.0  # seconds
     video.seek(target_time)
     frame = video.read()
@@ -134,21 +134,21 @@ def test_vfr_seek_pyav(test_vfr_video: str):
     assert abs(video.position.seconds - target_time) < 1.0
 
 
-def test_vfr_stats_manager(test_vfr_video: str):
+def test_vfr_stats_manager(test_vfr_video: str, auto_close):
     """StatsManager should work correctly with VFR video."""
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     stats = StatsManager()
     sm = SceneManager(stats_manager=stats)
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video)
     assert len(sm.get_scene_list()) > 0
 
 
-def test_vfr_csv_output(test_vfr_video: str, tmp_path):
+def test_vfr_csv_output(test_vfr_video: str, tmp_path, auto_close):
     """CSV export should work correctly with VFR video."""
     from scenedetect.output import write_scene_list
 
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     sm = SceneManager()
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video)
@@ -167,10 +167,10 @@ def test_vfr_csv_output(test_vfr_video: str, tmp_path):
 
 
 @pytest.mark.parametrize("backend", ["pyav", "opencv"])
-def test_vfr_drop3_scene_detection(test_vfr_drop3_video: str, backend: str):
+def test_vfr_drop3_scene_detection(test_vfr_drop3_video: str, backend: str, auto_close):
     """Synthetic VFR video (drop every 3rd frame, alternating 1x/2x durations) should produce
     timecodes matching known ground truth with both backends."""
-    video = open_video(test_vfr_drop3_video, backend=backend)
+    video = auto_close(open_video(test_vfr_drop3_video, backend=backend))
     sm = SceneManager()
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video, show_progress=False)
@@ -191,9 +191,9 @@ def test_vfr_drop3_scene_detection(test_vfr_drop3_video: str, backend: str):
 
 
 @pytest.mark.parametrize("backend", ["pyav", "opencv"])
-def test_vfr_drop3_position_monotonic(test_vfr_drop3_video: str, backend: str):
+def test_vfr_drop3_position_monotonic(test_vfr_drop3_video: str, backend: str, auto_close):
     """PTS-based position should be monotonically non-decreasing on synthetic VFR video."""
-    video = open_video(test_vfr_drop3_video, backend=backend)
+    video = auto_close(open_video(test_vfr_drop3_video, backend=backend))
     last_seconds = -1.0
     frame_count = 0
     while True:
@@ -208,22 +208,22 @@ def test_vfr_drop3_position_monotonic(test_vfr_drop3_video: str, backend: str):
     assert frame_count == 160  # 2/3 of original 240 frames in 10s at 24000/1001
 
 
-def test_cfr_position_is_timecode(test_movie_clip: str):
+def test_cfr_position_is_timecode(test_movie_clip: str, auto_close):
     """CFR video positions should also be Timecode-backed with PTS support."""
-    video = open_video(test_movie_clip, backend="pyav")
+    video = auto_close(open_video(test_movie_clip, backend="pyav"))
     assert video.read() is not False
     assert isinstance(video.position._time, Timecode)
 
 
-def test_cfr_frame_num_exact(test_movie_clip: str):
+def test_cfr_frame_num_exact(test_movie_clip: str, auto_close):
     """For CFR video, frame_num should be exact (not approximate)."""
-    video = open_video(test_movie_clip, backend="pyav")
+    video = auto_close(open_video(test_movie_clip, backend="pyav"))
     for expected_frame in range(1, 11):
         assert video.read() is not False
         assert video.position.frame_num == expected_frame - 1
 
 
-def test_vfr_save_images_opencv_matches_pyav(test_vfr_video: str, tmp_path):
+def test_vfr_save_images_opencv_matches_pyav(test_vfr_video: str, tmp_path, auto_close):
     """OpenCV save-images thumbnails should match PyAV thumbnails for all scenes.
 
     If the OpenCV seek off-by-one bug is present, scene thumbnails will show content from the
@@ -233,7 +233,7 @@ def test_vfr_save_images_opencv_matches_pyav(test_vfr_video: str, tmp_path):
     # must not run per-backend: the cut at 00:01:39.474 scores content_val=27.08 against the
     # default threshold of 27.0, so decoder/colorspace differences between backends (or FFmpeg
     # builds - e.g. av 17.1.0 on macOS arm64) can flip it, changing the scene count.
-    video = open_video(test_vfr_video, backend="pyav")
+    video = auto_close(open_video(test_vfr_video, backend="pyav"))
     sm = SceneManager()
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video)
@@ -246,7 +246,7 @@ def test_vfr_save_images_opencv_matches_pyav(test_vfr_video: str, tmp_path):
     for backend in ("pyav", "opencv"):
         out_dir = tmp_path / backend
         out_dir.mkdir()
-        video = open_video(test_vfr_video, backend=backend)
+        video = auto_close(open_video(test_vfr_video, backend=backend))
         rebased = [
             (FrameTimecode(start, fps=video.frame_rate), FrameTimecode(end, fps=video.frame_rate))
             for start, end in scene_list
@@ -282,9 +282,9 @@ def test_vfr_save_images_opencv_matches_pyav(test_vfr_video: str, tmp_path):
 
 
 @pytest.mark.parametrize("backend", ["pyav", "opencv"])
-def test_vfr_csv_accuracy(test_vfr_video: str, backend: str, tmp_path):
+def test_vfr_csv_accuracy(test_vfr_video: str, backend: str, tmp_path, auto_close):
     """CSV timecodes for VFR video should match known ground truth for both backends."""
-    video = open_video(test_vfr_video, backend=backend)
+    video = auto_close(open_video(test_vfr_video, backend=backend))
     sm = SceneManager()
     sm.add_detector(ContentDetector())
     sm.detect_scenes(video=video, end_time=10.0)
@@ -417,15 +417,15 @@ def test_vfr_fcp_export(test_vfr_video: str, fcp_format: str, tmp_path):
     assert root.tag == ("fcpxml" if fcp_format == "fcpx" else "xmeml")
 
 
-def test_vfr_csv_backend_conformance(test_vfr_video: str):
+def test_vfr_csv_backend_conformance(test_vfr_video: str, auto_close):
     """PyAV and OpenCV should produce identical scene timecodes for VFR video.
 
     Only the known interior scenes are compared; the last scene's end time may vary slightly
     between backends since it reflects the clip boundary rather than a detected cut.
     """
     timecodes: dict[str, list[tuple[str, str]]] = {}
     for backend in ("pyav", "opencv"):
-        video = open_video(test_vfr_video, backend=backend)
+        video = auto_close(open_video(test_vfr_video, backend=backend))
         sm = SceneManager()
         sm.add_detector(ContentDetector())
         sm.detect_scenes(video=video, end_time=10.0)