Fix process_video and add test case

supervision/utils/video.py

@@ -230,29 +230,24 @@ def callback(scene: np.ndarray, index: int) -> np.ndarray:
         ```
     """
     source_video_info = VideoInfo.from_video_path(video_path=source_path)
+    max_frames = max_frames or source_video_info.total_frames
+    if source_video_info.total_frames is not None and max_frames is not None:
+        max_frames = min(max_frames, source_video_info.total_frames)
+
     video_frames_generator = get_video_frames_generator(
         source_path=source_path, end=max_frames
     )
     with VideoSink(target_path=target_path, video_info=source_video_info) as sink:
-        total_frames = (
-            min(source_video_info.total_frames, max_frames)
-            if max_frames is not None
-            else source_video_info.total_frames
-        )
         for index, frame in enumerate(
             tqdm(
                 video_frames_generator,
-                total=total_frames,
+                total=max_frames,
                 disable=not show_progress,
                 desc=progress_message,
             )
         ):
             result_frame = callback(frame, index)
             sink.write_frame(frame=result_frame)
-        else:
-            for index, frame in enumerate(video_frames_generator):
-                result_frame = callback(frame, index)
-                sink.write_frame(frame=result_frame)
 
 
 class FPSMonitor:

test/utils/test_video.py

@@ -0,0 +1,32 @@
+import tempfile
+
+import cv2
+import numpy as np
+
+from supervision.utils.video import process_video
+
+
+def create_test_video(path, num_frames, width=20, height=10):
+    fourcc = cv2.VideoWriter_fourcc(*"mp4v")
+    out = cv2.VideoWriter(path, fourcc, 1.0, (width, height))
+
+    for _ in range(num_frames):
+        frame = np.zeros((height, width, 3), dtype=np.uint8)
+        out.write(frame)
+
+    out.release()
+
+
+def test_process_video_max_frames_exceeds_total_frames():
+    with (
+        tempfile.NamedTemporaryFile(suffix=".mp4") as source_file,
+        tempfile.NamedTemporaryFile(suffix=".mp4") as target_file,
+    ):
+        create_test_video(source_file.name, num_frames=5)
+
+        process_video(
+            source_path=source_file.name,
+            target_path=target_file.name,
+            callback=lambda frame, _: frame,
+            max_frames=10,
+        )