wehub-resource-sync
9194ef5abd
Docs/Test Workflow / Test docs build (push) Failing after 0s
Check links & references / links-check (push) Failing after 1s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.10) (push) Failing after 0s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.11) (push) Failing after 0s
PR Conflict Labeler / main (push) Failing after 2s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.12) (push) Failing after 2s
Pytest/Test Workflow / Import Test and Pytest Run (ubuntu-latest, 3.13) (push) Failing after 0s
Pytest/Test Workflow / Build this Package (push) Failing after 5s
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.10) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.11) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.12) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (macos-latest, 3.13) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.10) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.11) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.12) (push) Has been cancelled
Pytest/Test Workflow / Import Test and Pytest Run (windows-latest, 3.13) (push) Has been cancelled
Pytest/Test Workflow / testing-guardian (push) Has been cancelled
692 lines
24 KiB
Python
692 lines
24 KiB
Python
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from queue import Empty, Full
|
|
from queue import Queue as StdQueue
|
|
from types import SimpleNamespace
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from supervision.utils.video import (
|
|
FPSMonitor,
|
|
VideoInfo,
|
|
_mux_audio,
|
|
get_video_frames_generator,
|
|
process_video,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def dummy_video_path(tmp_path):
|
|
path = str(tmp_path / "dummy_video.mp4")
|
|
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
|
|
out = cv2.VideoWriter(path, fourcc, 25, (640, 480))
|
|
for _ in range(10):
|
|
frame = np.zeros((480, 640, 3), dtype=np.uint8)
|
|
out.write(frame)
|
|
out.release()
|
|
return path
|
|
|
|
|
|
def test_process_video_exception_handling(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video correctly propagates exceptions from the callback.
|
|
|
|
Scenario: Processing a video where the callback raises an exception.
|
|
Expected: `process_video` should propagate the exception, allowing users to
|
|
handle errors during video processing.
|
|
"""
|
|
target_path = str(tmp_path / "target.mp4")
|
|
|
|
def callback_with_exception(frame, index):
|
|
if index == 5:
|
|
raise ValueError("Test exception at frame 5")
|
|
return frame
|
|
|
|
with pytest.raises(ValueError, match="Test exception at frame 5"):
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback_with_exception,
|
|
)
|
|
|
|
|
|
def test_process_video_success(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify successful video processing with a pass-through callback.
|
|
|
|
Scenario: Successfully processing a video with a simple pass-through callback.
|
|
Expected: The video is processed without error and the target file is created,
|
|
verifying the core functionality of `process_video`.
|
|
"""
|
|
target_path = str(tmp_path / "target_success.mp4")
|
|
|
|
def callback_success(frame, index):
|
|
return frame
|
|
|
|
# This should complete without exception
|
|
process_video(
|
|
source_path=dummy_video_path, target_path=target_path, callback=callback_success
|
|
)
|
|
|
|
assert os.path.exists(target_path)
|
|
|
|
|
|
def test_process_video_exception_with_small_buffer(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video handles exceptions correctly even with small buffers.
|
|
|
|
Scenario: Processing a video with minimal buffering where an exception occurs.
|
|
Expected: The exception is still correctly propagated even with low memory settings.
|
|
"""
|
|
target_path = str(tmp_path / "target_exception_small_buffer.mp4")
|
|
|
|
def callback_with_exception(frame, index):
|
|
if index == 5:
|
|
raise ValueError("Test exception at frame 5")
|
|
return frame
|
|
|
|
with pytest.raises(ValueError, match="Test exception at frame 5"):
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback_with_exception,
|
|
prefetch=1,
|
|
writer_buffer=1,
|
|
)
|
|
|
|
|
|
def test_process_video_enqueues_writer_sentinel_with_timeout(
|
|
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""process_video enqueues the writer sentinel and bounded worker joins."""
|
|
read_queue = StdQueue()
|
|
read_queue.put((0, np.zeros((2, 2, 3), dtype=np.uint8)))
|
|
read_queue.put((1, np.zeros((2, 2, 3), dtype=np.uint8)))
|
|
read_queue.put(None)
|
|
|
|
class RecordingWriteQueue:
|
|
"""Record writer queue puts so the shutdown path can be asserted."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the queue call log."""
|
|
self.put_calls: list[tuple[object, object]] = []
|
|
|
|
def put(self, item: object, timeout: object | None = None) -> None:
|
|
"""Record each put call and its timeout."""
|
|
self.put_calls.append((item, timeout))
|
|
|
|
def get(self, timeout: object | None = None) -> object:
|
|
"""The writer thread is disabled, so reads are not expected."""
|
|
raise AssertionError("writer queue should not be read in this test")
|
|
|
|
join_calls: list[object | None] = []
|
|
|
|
class FakeThread:
|
|
"""Thread stand-in that keeps the test single-threaded."""
|
|
|
|
def __init__(
|
|
self,
|
|
target: object,
|
|
args: tuple[object, ...] = (),
|
|
daemon: bool = False,
|
|
) -> None:
|
|
"""Store the thread target without starting it."""
|
|
self.target = target
|
|
self.args = args
|
|
self.daemon = daemon
|
|
|
|
def start(self) -> None:
|
|
"""Do nothing; the test preloads the queues instead."""
|
|
|
|
def join(self, timeout=None) -> None:
|
|
"""Do nothing; the worker targets are intentionally never started."""
|
|
join_calls.append(timeout)
|
|
|
|
class FakeVideoSink:
|
|
"""Minimal sink context manager used to verify shutdown ordering."""
|
|
|
|
def __init__(self, target_path: str, video_info: object) -> None:
|
|
"""Store constructor arguments for completeness."""
|
|
self.target_path = target_path
|
|
self.video_info = video_info
|
|
|
|
def __enter__(self) -> "FakeVideoSink":
|
|
"""Return the sink context manager."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
"""Propagate any exception without side effects."""
|
|
return None
|
|
|
|
def write_frame(self, frame: object) -> None:
|
|
"""The writer thread is disabled in this test."""
|
|
|
|
write_queue = RecordingWriteQueue()
|
|
queue_factory_calls = iter([read_queue, write_queue])
|
|
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.Queue",
|
|
lambda *args, **kwargs: next(queue_factory_calls),
|
|
)
|
|
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
|
|
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.VideoInfo.from_video_path",
|
|
lambda video_path: SimpleNamespace(total_frames=2),
|
|
)
|
|
|
|
target_path = str(tmp_path / "target_sentinel.mp4")
|
|
|
|
def callback(frame, index):
|
|
if index == 1:
|
|
raise ValueError("Test exception at frame 1")
|
|
return frame
|
|
|
|
with pytest.raises(ValueError, match="Test exception at frame 1"):
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback,
|
|
show_progress=False,
|
|
)
|
|
|
|
assert write_queue.put_calls[-1] == (None, 1)
|
|
assert all(timeout is None for _item, timeout in write_queue.put_calls[:-1])
|
|
assert join_calls == [10, 10]
|
|
|
|
|
|
def test_process_video_best_effort_sentinel_handles_full_queue(
|
|
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""process_video should not hang if the writer queue is already full."""
|
|
read_queue = StdQueue()
|
|
read_queue.put((0, np.zeros((2, 2, 3), dtype=np.uint8)))
|
|
read_queue.put(None)
|
|
|
|
class FullWriteQueue:
|
|
"""Record writer queue puts and fail the shutdown sentinel."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the queue call log."""
|
|
self.put_calls: list[tuple[object, object | None]] = []
|
|
|
|
def put(self, item: object, timeout: object | None = None) -> None:
|
|
"""Record the put and raise Full for the shutdown sentinel."""
|
|
self.put_calls.append((item, timeout))
|
|
if item is None:
|
|
raise Full
|
|
|
|
def get(self, timeout: object | None = None) -> object:
|
|
"""The writer thread is disabled, so reads are not expected."""
|
|
raise AssertionError("writer queue should not be read in this test")
|
|
|
|
join_calls: list[object | None] = []
|
|
|
|
class FakeThread:
|
|
"""Thread stand-in that keeps the test single-threaded."""
|
|
|
|
def __init__(
|
|
self,
|
|
target: object,
|
|
args: tuple[object, ...] = (),
|
|
daemon: bool = False,
|
|
) -> None:
|
|
"""Store the thread target without starting it."""
|
|
self.target = target
|
|
self.args = args
|
|
self.daemon = daemon
|
|
|
|
def start(self) -> None:
|
|
"""Do nothing; the test preloads the queues instead."""
|
|
|
|
def join(self, timeout=None) -> None:
|
|
"""Record join timeouts for shutdown verification."""
|
|
join_calls.append(timeout)
|
|
|
|
class FakeVideoSink:
|
|
"""Minimal sink context manager used to verify shutdown ordering."""
|
|
|
|
def __init__(self, target_path: str, video_info: object) -> None:
|
|
"""Store constructor arguments for completeness."""
|
|
self.target_path = target_path
|
|
self.video_info = video_info
|
|
|
|
def __enter__(self) -> "FakeVideoSink":
|
|
"""Return the sink context manager."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
"""Propagate any exception without side effects."""
|
|
return None
|
|
|
|
def write_frame(self, frame: object) -> None:
|
|
"""The writer thread is disabled in this test."""
|
|
|
|
write_queue = FullWriteQueue()
|
|
queue_factory_calls = iter([read_queue, write_queue])
|
|
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.Queue",
|
|
lambda *args, **kwargs: next(queue_factory_calls),
|
|
)
|
|
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
|
|
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.VideoInfo.from_video_path",
|
|
lambda video_path: SimpleNamespace(total_frames=1),
|
|
)
|
|
|
|
target_path = str(tmp_path / "target_full_queue.mp4")
|
|
|
|
def callback(frame, index):
|
|
raise ValueError("Test exception at frame 0")
|
|
|
|
with pytest.raises(ValueError, match="Test exception at frame 0"):
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback,
|
|
show_progress=False,
|
|
)
|
|
|
|
assert write_queue.put_calls[-1] == (None, 1)
|
|
assert join_calls == [10, 10]
|
|
|
|
|
|
def test_process_video_waits_for_reader_timeout_when_queue_is_empty(
|
|
dummy_video_path: str, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""process_video should keep waiting briefly when the reader queue times out."""
|
|
|
|
class TimeoutReadQueue:
|
|
"""Record the first frame read and then time out in shutdown."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the queue call log."""
|
|
self.get_calls: list[object | None] = []
|
|
|
|
def put(self, item: object, timeout: object | None = None) -> None:
|
|
"""The reader thread is disabled, so writes are not expected."""
|
|
raise AssertionError("reader queue should not be written in this test")
|
|
|
|
def get(self, timeout: object | None = None) -> object:
|
|
"""Yield one frame during processing, then time out during shutdown."""
|
|
self.get_calls.append(timeout)
|
|
if timeout is None:
|
|
return (0, np.zeros((2, 2, 3), dtype=np.uint8))
|
|
raise Empty
|
|
|
|
class RecordingWriteQueue:
|
|
"""Record writer queue puts so the shutdown path can be asserted."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Initialize the queue call log."""
|
|
self.put_calls: list[tuple[object, object | None]] = []
|
|
|
|
def put(self, item: object, timeout: object | None = None) -> None:
|
|
"""Record each put call and its timeout."""
|
|
self.put_calls.append((item, timeout))
|
|
|
|
def get(self, timeout: object | None = None) -> object:
|
|
"""The writer thread is disabled, so reads are not expected."""
|
|
raise AssertionError("writer queue should not be read in this test")
|
|
|
|
join_calls: list[object | None] = []
|
|
reader_alive_states = iter([True, False])
|
|
|
|
class FakeThread:
|
|
"""Thread stand-in that keeps the test single-threaded."""
|
|
|
|
def __init__(
|
|
self,
|
|
target: object,
|
|
args: tuple[object, ...] = (),
|
|
daemon: bool = False,
|
|
) -> None:
|
|
"""Store the thread target without starting it."""
|
|
self.target = target
|
|
self.args = args
|
|
self.daemon = daemon
|
|
|
|
def start(self) -> None:
|
|
"""Do nothing; the test preloads the queues instead."""
|
|
|
|
def join(self, timeout=None) -> None:
|
|
"""Record join timeouts for shutdown verification."""
|
|
join_calls.append(timeout)
|
|
|
|
def is_alive(self) -> bool:
|
|
"""Return a short-lived alive state so the timeout branch is hit."""
|
|
return next(reader_alive_states, False)
|
|
|
|
class FakeVideoSink:
|
|
"""Minimal sink context manager used to verify shutdown ordering."""
|
|
|
|
def __init__(self, target_path: str, video_info: object) -> None:
|
|
"""Store constructor arguments for completeness."""
|
|
self.target_path = target_path
|
|
self.video_info = video_info
|
|
|
|
def __enter__(self) -> "FakeVideoSink":
|
|
"""Return the sink context manager."""
|
|
return self
|
|
|
|
def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
|
|
"""Propagate any exception without side effects."""
|
|
return None
|
|
|
|
def write_frame(self, frame: object) -> None:
|
|
"""The writer thread is disabled in this test."""
|
|
|
|
read_queue = TimeoutReadQueue()
|
|
write_queue = RecordingWriteQueue()
|
|
queue_factory_calls = iter([read_queue, write_queue])
|
|
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.Queue",
|
|
lambda *args, **kwargs: next(queue_factory_calls),
|
|
)
|
|
monkeypatch.setattr("supervision.utils.video.threading.Thread", FakeThread)
|
|
monkeypatch.setattr("supervision.utils.video.VideoSink", FakeVideoSink)
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.VideoInfo.from_video_path",
|
|
lambda video_path: SimpleNamespace(total_frames=1),
|
|
)
|
|
|
|
target_path = str(tmp_path / "target_timeout.mp4")
|
|
|
|
def callback(frame, index):
|
|
raise ValueError("Test exception at frame 0")
|
|
|
|
with pytest.raises(ValueError, match="Test exception at frame 0"):
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback,
|
|
show_progress=False,
|
|
)
|
|
|
|
assert read_queue.get_calls == [None, 1, 1]
|
|
assert join_calls == [10, 10]
|
|
|
|
|
|
def test_process_video_max_frames(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video respects the max_frames parameter.
|
|
|
|
Scenario: Processing only a limited number of frames using `max_frames`.
|
|
Expected: Only the specified number of frames are processed, which is useful for
|
|
quick testing or sampling.
|
|
"""
|
|
target_path = str(tmp_path / "target_max_frames.mp4")
|
|
processed_indices = []
|
|
|
|
def callback(frame, index):
|
|
processed_indices.append(index)
|
|
return frame
|
|
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback,
|
|
max_frames=5,
|
|
)
|
|
|
|
assert len(processed_indices) == 5
|
|
assert processed_indices == [0, 1, 2, 3, 4]
|
|
|
|
|
|
def test_process_video_custom_params(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video works correctly with custom performance parameters.
|
|
|
|
Scenario: Processing video with custom prefetch and buffer parameters.
|
|
Expected: Video is processed successfully, showing that these performance-tuning
|
|
parameters are correctly handled.
|
|
"""
|
|
target_path = str(tmp_path / "target_custom_params.mp4")
|
|
|
|
def callback(frame, index):
|
|
return frame
|
|
|
|
# Test with very small prefetch and writer_buffer
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=callback,
|
|
prefetch=1,
|
|
writer_buffer=1,
|
|
)
|
|
|
|
assert os.path.exists(target_path)
|
|
|
|
|
|
def test_video_info(dummy_video_path) -> None:
|
|
"""
|
|
Verify that VideoInfo correctly retrieves metadata from a video file.
|
|
|
|
Scenario: Retrieving metadata from a video file using `VideoInfo`.
|
|
Expected: Correct width, height, fps, and frame count are returned, which is
|
|
essential for initializing annotators or calculating statistics.
|
|
"""
|
|
video_info = VideoInfo.from_video_path(dummy_video_path)
|
|
assert video_info.width == 640
|
|
assert video_info.height == 480
|
|
assert video_info.fps == pytest.approx(25.0)
|
|
assert isinstance(video_info.fps, float)
|
|
assert video_info.total_frames == 10
|
|
assert video_info.resolution_wh == (640, 480)
|
|
|
|
|
|
def test_video_info_float_fps(dummy_video_path, monkeypatch) -> None:
|
|
"""
|
|
Verify that VideoInfo preserves non-integer FPS values as floats.
|
|
|
|
Scenario: Retrieving metadata from a video while OpenCV reports 23.976 fps.
|
|
Expected: fps is returned as the original float value, not truncated to an
|
|
integer. This prevents frame-timing drift in long videos.
|
|
"""
|
|
original_get = cv2.VideoCapture.get
|
|
|
|
def mocked_get(self, prop_id):
|
|
if prop_id == cv2.CAP_PROP_FPS:
|
|
return 23.976
|
|
return original_get(self, prop_id)
|
|
|
|
monkeypatch.setattr(cv2.VideoCapture, "get", mocked_get)
|
|
|
|
video_info = VideoInfo.from_video_path(dummy_video_path)
|
|
assert isinstance(video_info.fps, float)
|
|
assert video_info.fps == pytest.approx(23.976)
|
|
assert video_info.fps != int(video_info.fps)
|
|
|
|
|
|
def test_get_video_frames_generator(dummy_video_path) -> None:
|
|
"""
|
|
Verify that get_video_frames_generator yields frames with correct shapes.
|
|
|
|
Scenario: Iterating over video frames using a generator.
|
|
Expected: All frames are yielded in order as NumPy arrays with correct shapes,
|
|
enabling frame-by-frame processing loops.
|
|
"""
|
|
generator = get_video_frames_generator(dummy_video_path)
|
|
frames = list(generator)
|
|
assert len(frames) == 10
|
|
assert all(isinstance(frame, np.ndarray) for frame in frames)
|
|
assert all(frame.shape == (480, 640, 3) for frame in frames)
|
|
|
|
|
|
def test_get_video_frames_generator_releases_on_early_break(monkeypatch) -> None:
|
|
"""
|
|
Verify that the capture is released when a consumer breaks out early.
|
|
|
|
Scenario: A consumer iterates one frame then abandons the generator, raising
|
|
GeneratorExit at the yield point.
|
|
Expected: The `try/finally` guard still calls `release()`, avoiding a decoder
|
|
leak.
|
|
"""
|
|
|
|
class FakeCapture:
|
|
def __init__(self) -> None:
|
|
self.released = False
|
|
|
|
def read(self):
|
|
return True, np.zeros((2, 2, 3), dtype=np.uint8)
|
|
|
|
def grab(self):
|
|
return True
|
|
|
|
def release(self) -> None:
|
|
self.released = True
|
|
|
|
fake_capture = FakeCapture()
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video._validate_and_setup_video",
|
|
lambda *args, **kwargs: (fake_capture, 0, 100),
|
|
)
|
|
|
|
generator = get_video_frames_generator("dummy")
|
|
next(generator)
|
|
generator.close()
|
|
|
|
assert fake_capture.released
|
|
|
|
|
|
def test_get_video_frames_generator_with_stride(dummy_video_path) -> None:
|
|
"""
|
|
Verify that get_video_frames_generator correctly handles the stride parameter.
|
|
|
|
Scenario: Iterating over video frames with specified stride (e.g., every 2nd frame).
|
|
Expected: The generator correctly skips frames according to the stride, allowing
|
|
for faster processing of high-FPS videos.
|
|
"""
|
|
generator = get_video_frames_generator(dummy_video_path, stride=2)
|
|
frames = list(generator)
|
|
assert len(frames) == 5
|
|
|
|
|
|
def test_fps_monitor_uses_frame_intervals(monkeypatch) -> None:
|
|
"""FPSMonitor must divide elapsed time by intervals, not sample count."""
|
|
timestamps = iter([0.0, 0.5, 1.0])
|
|
monkeypatch.setattr(
|
|
"supervision.utils.video.time.monotonic", lambda: next(timestamps)
|
|
)
|
|
|
|
fps_monitor = FPSMonitor()
|
|
fps_monitor.tick()
|
|
fps_monitor.tick()
|
|
fps_monitor.tick()
|
|
|
|
assert fps_monitor.fps == pytest.approx(2.0)
|
|
|
|
|
|
def test_process_video_preserve_audio_calls_mux(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video calls _mux_audio when preserve_audio=True.
|
|
|
|
Scenario: Processing a video with preserve_audio=True and ffmpeg available.
|
|
Expected: _mux_audio is called exactly once with the correct source and target
|
|
paths, confirming the audio muxing step is triggered after frame writing completes.
|
|
"""
|
|
target_path = str(tmp_path / "target_audio.mp4")
|
|
|
|
with patch("supervision.utils.video._mux_audio") as mock_mux:
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=lambda frame, idx: frame,
|
|
preserve_audio=True,
|
|
)
|
|
mock_mux.assert_called_once_with(
|
|
source_path=dummy_video_path, video_path=target_path
|
|
)
|
|
|
|
|
|
def test_process_video_no_audio_by_default(dummy_video_path, tmp_path) -> None:
|
|
"""
|
|
Verify that process_video does not call _mux_audio when preserve_audio=False.
|
|
|
|
Scenario: Default process_video call without setting preserve_audio.
|
|
Expected: _mux_audio is never called, preserving existing behavior for callers
|
|
that do not need audio.
|
|
"""
|
|
target_path = str(tmp_path / "target_no_audio.mp4")
|
|
|
|
with patch("supervision.utils.video._mux_audio") as mock_mux:
|
|
process_video(
|
|
source_path=dummy_video_path,
|
|
target_path=target_path,
|
|
callback=lambda frame, idx: frame,
|
|
)
|
|
mock_mux.assert_not_called()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("which_rv", "run_kwargs"),
|
|
[
|
|
pytest.param(None, {}, id="ffmpeg_missing"),
|
|
pytest.param(
|
|
"/usr/bin/ffmpeg",
|
|
{"return_value": MagicMock(returncode=1, stderr=b"")},
|
|
id="ffmpeg_fails",
|
|
),
|
|
pytest.param(
|
|
"/usr/bin/ffmpeg",
|
|
{"side_effect": OSError("mux failed")},
|
|
id="subprocess_raises",
|
|
),
|
|
],
|
|
)
|
|
def test_mux_audio_file_unchanged_on_failure(
|
|
dummy_video_path, tmp_path, which_rv, run_kwargs
|
|
) -> None:
|
|
"""_mux_audio leaves the output file unchanged when muxing cannot complete."""
|
|
target_path = str(tmp_path / "video.mp4")
|
|
shutil.copy(dummy_video_path, target_path)
|
|
original_size = os.path.getsize(target_path)
|
|
|
|
with (
|
|
patch("supervision.utils.video.shutil.which", return_value=which_rv),
|
|
patch("supervision.utils.video.subprocess.run", **run_kwargs),
|
|
):
|
|
_mux_audio(source_path=dummy_video_path, video_path=target_path)
|
|
|
|
assert os.path.getsize(target_path) == original_size
|
|
|
|
|
|
def test_mux_audio_replaces_file_on_success(dummy_video_path, tmp_path) -> None:
|
|
"""_mux_audio calls os.replace with video_path as destination on success."""
|
|
target_path = str(tmp_path / "video.mp4")
|
|
shutil.copy(dummy_video_path, target_path)
|
|
|
|
success_result = MagicMock()
|
|
success_result.returncode = 0
|
|
success_result.stderr = b""
|
|
|
|
with (
|
|
patch("supervision.utils.video.shutil.which", return_value="/usr/bin/ffmpeg"),
|
|
patch("supervision.utils.video.subprocess.run", return_value=success_result),
|
|
patch("supervision.utils.video.os.replace") as mock_replace,
|
|
):
|
|
_mux_audio(source_path=dummy_video_path, video_path=target_path)
|
|
|
|
mock_replace.assert_called_once()
|
|
assert mock_replace.call_args[0][1] == target_path
|
|
|
|
|
|
def test_get_video_frames_generator_with_start_end(dummy_video_path) -> None:
|
|
"""
|
|
Verify that get_video_frames_generator respects start and end frame indices.
|
|
|
|
Scenario: Iterating over a specific range of video frames using `start` and `end`.
|
|
Expected: Only frames within the specified range are yielded, enabling targeted
|
|
analysis of video segments.
|
|
"""
|
|
generator = get_video_frames_generator(dummy_video_path, start=2, end=5)
|
|
frames = list(generator)
|
|
assert len(frames) == 3
|