fix: release camera capture handle on failed or repeated open
- Fix critical resource leak when cv2.VideoCapture.isOpened() returns False: create capture in local variable, verify it opened, call .release() before raising RuntimeError, only assign to self._capture after confirming success - Fix potential leak on repeated open(): release any existing self._capture before creating a new one - Add regression test: test_open_raises_when_capture_not_opened now verifies .release() is called on failed capture - Add new test: test_open_twice_releases_first_capture verifies first capture's .release() is called before second is assigned All 28 tests passing (5 camera + 23 existing).
This commit is contained in:
parent
ab7664cd51
commit
efe3f65cd5
2 changed files with 35 additions and 2 deletions
|
|
@ -16,9 +16,16 @@ class CameraSource:
|
||||||
self._capture: cv2.VideoCapture | None = None
|
self._capture: cv2.VideoCapture | None = None
|
||||||
|
|
||||||
def open(self) -> None:
|
def open(self) -> None:
|
||||||
self._capture = cv2.VideoCapture(self.device_index)
|
# Release any existing capture before creating a new one
|
||||||
if not self._capture.isOpened():
|
if self._capture is not None:
|
||||||
|
self._capture.release()
|
||||||
|
|
||||||
|
# Create capture in local variable and verify it opened before assignment
|
||||||
|
capture = cv2.VideoCapture(self.device_index)
|
||||||
|
if not capture.isOpened():
|
||||||
|
capture.release()
|
||||||
raise RuntimeError(f"Could not open camera at index {self.device_index}")
|
raise RuntimeError(f"Could not open camera at index {self.device_index}")
|
||||||
|
self._capture = capture
|
||||||
|
|
||||||
def read_frame(self) -> np.ndarray:
|
def read_frame(self) -> np.ndarray:
|
||||||
if self._capture is None:
|
if self._capture is None:
|
||||||
|
|
|
||||||
|
|
@ -16,6 +16,9 @@ def test_open_raises_when_capture_not_opened():
|
||||||
with pytest.raises(RuntimeError, match="Could not open camera"):
|
with pytest.raises(RuntimeError, match="Could not open camera"):
|
||||||
source.open()
|
source.open()
|
||||||
|
|
||||||
|
# Verify that release() was called on the failed capture
|
||||||
|
mock_instance.release.assert_called_once()
|
||||||
|
|
||||||
|
|
||||||
def test_read_frame_returns_frame_from_capture():
|
def test_read_frame_returns_frame_from_capture():
|
||||||
with patch("bookmark.camera.cv2.VideoCapture") as mock_video_capture:
|
with patch("bookmark.camera.cv2.VideoCapture") as mock_video_capture:
|
||||||
|
|
@ -49,3 +52,26 @@ def test_read_frame_raises_if_not_opened():
|
||||||
source = CameraSource(device_index=0)
|
source = CameraSource(device_index=0)
|
||||||
with pytest.raises(RuntimeError, match="not opened"):
|
with pytest.raises(RuntimeError, match="not opened"):
|
||||||
source.read_frame()
|
source.read_frame()
|
||||||
|
|
||||||
|
|
||||||
|
def test_open_twice_releases_first_capture():
|
||||||
|
with patch("bookmark.camera.cv2.VideoCapture") as mock_video_capture:
|
||||||
|
# Setup two distinct mock instances
|
||||||
|
first_mock = MagicMock()
|
||||||
|
first_mock.isOpened.return_value = True
|
||||||
|
second_mock = MagicMock()
|
||||||
|
second_mock.isOpened.return_value = True
|
||||||
|
mock_video_capture.side_effect = [first_mock, second_mock]
|
||||||
|
|
||||||
|
source = CameraSource(device_index=0)
|
||||||
|
|
||||||
|
# First open
|
||||||
|
source.open()
|
||||||
|
assert source._capture is first_mock
|
||||||
|
|
||||||
|
# Second open should release the first capture
|
||||||
|
source.open()
|
||||||
|
assert source._capture is second_mock
|
||||||
|
|
||||||
|
# Verify the first mock's release() was called before the second was assigned
|
||||||
|
first_mock.release.assert_called_once()
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue