diff --git a/pyproject.toml b/pyproject.toml index 6130ffd..6362a34 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "pychdk" -version = "0.1.1" +version = "0.1.2" description = "Pure Python CHDK PTP camera control" requires-python = ">=3.11" dependencies = [ diff --git a/src/pychdk/__init__.py b/src/pychdk/__init__.py index 2e758c2..ad0d715 100644 --- a/src/pychdk/__init__.py +++ b/src/pychdk/__init__.py @@ -1,7 +1,7 @@ """Pure Python CHDK PTP camera control.""" import importlib -__version__ = "0.1.1" +__version__ = "0.1.2" __all__ = [ "ChdkDevice", "list_devices", "DeviceInfo", "install_signal_handlers", diff --git a/src/pychdk/chdk.py b/src/pychdk/chdk.py index a6ae810..9d46d45 100644 --- a/src/pychdk/chdk.py +++ b/src/pychdk/chdk.py @@ -156,6 +156,35 @@ class ChdkPTP: def __init__(self, session): self._session = session + self._last_capture_chunks = 0 + + @property + def last_capture_chunks(self): + """How many chunks the last remote capture arrived in. + + Zeroed when a capture is attempted and incremented as each + chunk lands, so it is readable — and still true — after a + capture that failed part way through, and reads zero after one + that never got a chunk at all. A still that arrives in one + chunk and one that arrives in forty say different things about + the wire, and there is one bench session to find out which of + them a real camera does. + + Read it from the thread that ran the capture, or after that + thread has finished: a reader watching from elsewhere while a + capture is in flight sees a partial count, since it rises as + the chunks arrive. + """ + return self._last_capture_chunks + + def reset_capture_chunks(self): + """Zero the chunk count at the start of a capture attempt. + + Called by the caller that begins a capture, because a capture + can fail before any download is attempted and must not go on + reporting the previous capture's chunks. + """ + self._last_capture_chunks = 0 def get_version(self): """Get CHDK PTP protocol version. @@ -404,6 +433,9 @@ def remote_capture_get_data(self, format_flag): Args: format_flag: Which format to download (JPEG=1, RAW=2, DNG_HDR=4). + The chunk count is left in last_capture_chunks rather than + returned, so the signature callers depend on is unchanged. + Returns: Image data as bytes. @@ -412,8 +444,10 @@ def remote_capture_get_data(self, format_flag): """ image = bytearray() cursor = 0 + self._last_capture_chunks = 0 for _ in range(MAX_CAPTURE_CHUNKS): chunk, more, position = self.remote_capture_get_chunk(format_flag) + self._last_capture_chunks += 1 if position >= 0: cursor = position end = cursor + len(chunk) diff --git a/src/pychdk/device.py b/src/pychdk/device.py index 314c2da..f74524e 100644 --- a/src/pychdk/device.py +++ b/src/pychdk/device.py @@ -157,18 +157,54 @@ def __init__(self, device_info, _usb_device=None): self._open() def _open(self): - self._transport.open() - self._session.open() - self._connected = True - _open_devices.add(self) - # Re-register so our cleanup runs before any pyusb finalizers - # that were registered during device creation (atexit is LIFO). - atexit.register(_cleanup_all) + """Claim the interface and open a session, or claim nothing. + + Until the device is tracked there is nothing for the caller to + close: a constructor that raised here left the interface + claimed with no object to release it, so a host retrying + enumeration piled up claims on a port until the camera was + unplugged. Anything that fails past the claim gives it back — + including a failure inside the transport's own open, which can + hold a claim and still raise. + """ + try: + self._transport.open() + self._session.open() + self._connected = True + _open_devices.add(self) + # Re-register so our cleanup runs before any pyusb finalizers + # that were registered during device creation (atexit is LIFO). + atexit.register(_cleanup_all) + except BaseException: + self._connected = False + _open_devices.discard(self) + try: + self._transport.close() + except Exception: + pass + raise @property def is_connected(self): return self._connected + @property + def last_capture_chunks(self): + """How many chunks the last streamed capture arrived in. + + Read after shoot(stream=True) rather than returned by it: the + return value is the picture, and MultiCam.shoot promises a list + of those, one per camera. The count lives per device, so after + a MultiCam shot each camera's own figure is on its entry in + MultiCam.cameras. + + Zero means no chunk arrived, not that no capture was tried. + Read it from the thread that took the shot, or once that thread + has finished: MultiCam shoots on a pool, and a reader looking + at another worker's device mid-capture sees a partial count. + """ + return self._chdk.last_capture_chunks + def switch_mode(self, mode): """Switch camera to 'record' or 'play' mode. @@ -239,6 +275,11 @@ def shoot(self, shutter_speed=None, market_iso=None, dng=False, def _shoot_streaming(self, setup_parts, dng): """Capture using remote capture (PTP commands 13/14). + The chunk count is zeroed here, at the attempt, rather than + where the download begins: a capture refused, or one that never + becomes ready, would otherwise keep reporting the chunks of the + capture before it. + Setup and shutter go out as one script, because a second script kills the first unless NOKILL is set ("if script is running return error instead of killing", core/ptp.h) — so a separate @@ -266,6 +307,8 @@ def _shoot_streaming(self, setup_parts, dng): file. This method downloads one format, so it cannot, and _shoot_standard does not request a DNG either. """ + self._chdk.reset_capture_chunks() + if dng: raise NotImplementedError( "DNG capture is not implemented. Streaming would need the " @@ -431,13 +474,26 @@ def reconnect(self, wait=2.0): except Exception: pass time.sleep(wait) - self._transport.open() - self._session.open() - self._connected = True - _open_devices.add(self) + # Same claim, same rollback: a reopen that fails mid-session + # leaks exactly as a failed construction did. + self._open() def close(self): - """Close the connection to the camera.""" + """Close the connection to the camera. + + Safe to call more than once in sequence. + + Concurrently it is safe in one half and not the other, and the + halves are worth keeping apart. Releasing the USB interface is + serialised by pyusb itself, so two closers cannot double-release + it — PTPDevice.close carries the citation. Closing the PTP + session is not serialised: this sends a close over the wire, and + two threads can both find the session open and both send one, + because nothing here guards that. So a host that shares one + device across threads has to serialise its own teardown. + Nothing in this library shares one: MultiCam gives each worker + its own device. + """ self._connected = False _open_devices.discard(self) try: diff --git a/src/pychdk/multicam.py b/src/pychdk/multicam.py index 2fd37e1..6ca1908 100644 --- a/src/pychdk/multicam.py +++ b/src/pychdk/multicam.py @@ -12,13 +12,28 @@ class MultiCam: """Manages multiple CHDK cameras for coordinated capture.""" def __init__(self): + """Open every camera found, or leave none of them open. + + A camera that fails to open partway down the list leaves the + ones before it open and claimed, and the half-built MultiCam is + discarded, so nothing is left holding them: not the caller, who + never got an object, and not the cleanup registry, which tracks + devices weakly. They stay claimed until the process ends. + + PTPDevice.open's guarantee does not reach this, because these + cameras opened successfully. They are orphans rather than + partial opens, so the rollback has to be here. + """ devices = list_devices() if not devices: raise RuntimeError("No CHDK cameras found") self.cameras = [] - for info in devices: - cam = ChdkDevice(info) - self.cameras.append(cam) + try: + for info in devices: + self.cameras.append(ChdkDevice(info)) + except BaseException: + self.close() + raise def shoot(self, **kwargs): """Capture from all cameras concurrently. diff --git a/src/pychdk/ptp.py b/src/pychdk/ptp.py index c0389fc..efdc469 100644 --- a/src/pychdk/ptp.py +++ b/src/pychdk/ptp.py @@ -213,6 +213,16 @@ def _receive_data(self, tx_id): return data[:total_length - CONTAINER_HEADER_SIZE] def __enter__(self): + # No rollback here, deliberately. The property: a session open + # that raises holds no operating-system resource, whatever + # raised it. Nothing is claimed, so there is nothing to give + # back, which is what makes this unlike PTPDevice.open. + # + # Not covered: the camera's own state. If OPEN_SESSION reached + # the camera and we failed before recording the session, the + # camera may hold one we will never close, since close() + # returns early on _is_open — and nothing on this side can tell + # that from a session that was never opened at all. self.open() return self diff --git a/src/pychdk/usb_transport.py b/src/pychdk/usb_transport.py index c91d76b..302bc36 100644 --- a/src/pychdk/usb_transport.py +++ b/src/pychdk/usb_transport.py @@ -90,6 +90,11 @@ def __init__(self, usb_device): self._ep_out = None self._ep_int = None self._intf_num = None + # The interface we actually hold, set the moment the claim + # succeeds rather than when opening finishes. Everything + # between the claim and the end of open() can fail, and what + # is held has to be releasable in between. + self._claimed_intf = None self._is_open = False @property @@ -116,7 +121,15 @@ def serial_number(self): return None def open(self): - """Open the PTP device — claim interface and find endpoints.""" + """Open the PTP device — claim interface and find endpoints. + + Either this returns with the interface claimed, or it raises + having claimed nothing. That guarantee lives here rather than + in each caller because it was a convention before, and three + call sites independently failed to honour it: a claim taken + and then lost to an exception is held until the camera is + unplugged, and every caller had to remember that separately. + """ if self._is_open: return @@ -154,40 +167,82 @@ def open(self): pass usb.util.claim_interface(self._dev, self._intf_num) + self._claimed_intf = self._intf_num - # Find endpoints - intf = cfg[(self._intf_num, 0)] - for ep in intf: - attr = ep.bmAttributes & 0x03 # transfer type mask - direction = ep.bEndpointAddress & 0x80 # direction mask - if attr == usb.util.ENDPOINT_TYPE_BULK: - if direction == EP_DIR_IN: - self._ep_in = ep - else: - self._ep_out = ep - elif attr == usb.util.ENDPOINT_TYPE_INTR: - if direction == EP_DIR_IN: - self._ep_int = ep - - if self._ep_in is None or self._ep_out is None: - raise RuntimeError("Could not find bulk endpoints on PTP device") - - self._is_open = True - - # Disable pyusb's weakref.finalize cleanup for this Device. - # During Python shutdown, pyusb's finalizer can call libusb_open - # after the libusb context has been freed, causing a SIGSEGV. - # We handle all USB cleanup ourselves in close(). - self._dev._finalize_called = True + # Everything past the claim runs under the guarantee: if it + # raises, the interface goes back before the exception does. + try: + # Find endpoints + intf = cfg[(self._intf_num, 0)] + for ep in intf: + attr = ep.bmAttributes & 0x03 # transfer type mask + direction = ep.bEndpointAddress & 0x80 # direction mask + if attr == usb.util.ENDPOINT_TYPE_BULK: + if direction == EP_DIR_IN: + self._ep_in = ep + else: + self._ep_out = ep + elif attr == usb.util.ENDPOINT_TYPE_INTR: + if direction == EP_DIR_IN: + self._ep_int = ep + + if self._ep_in is None or self._ep_out is None: + raise RuntimeError( + "Could not find bulk endpoints on PTP device" + ) + + self._is_open = True + + # Disable pyusb's weakref.finalize cleanup for this Device. + # During Python shutdown, pyusb's finalizer can call + # libusb_open after the libusb context has been freed, + # causing a SIGSEGV. We handle all USB cleanup ourselves + # in close(). + self._dev._finalize_called = True + except BaseException: + # Take ownership of cleanup before letting go. The + # finalizer disabled above is no less dangerous on a device + # we opened part way: releasing the claim and then leaving + # pyusb to reopen a freed context at shutdown would trade a + # leaked interface for a killed process. + self._dev._finalize_called = True + try: + self.close() + except Exception: + pass + raise def close(self): - """Release the USB interface and dispose of device resources.""" - if not self._is_open: + """Release whatever is held, however far open() got. + + Opening claims the interface and then goes looking for + endpoints, so a device can own a claim while still failing to + open. Keying this on _is_open made close() a no-op in exactly + that case, and the claim was then held until the camera was + unplugged. It keys on the claim instead. + + Repeated closes are safe in sequence, and two threads calling + close at once are safe as well, so there is no lock of our own + here. pyusb serialises claiming and releasing on a reentrant + lock it holds itself, and releases only an interface it still + records as claimed: in usb/core.py, _ResourceManager keeps a + threading.RLock, managed_claim_interface and + managed_release_interface are both decorated @synchronized + against it, and the release calls the backend only when the + interface is in its claimed set, removing it in a finally — so + a second release for the same interface does nothing, and a + repeated claim does not double-claim for the same reason. + Checked against the installed pyusb (1.3.1) rather than + assumed; worth a re-read if that version moves. + """ + if self._claimed_intf is None and not self._is_open: return - try: - usb.util.release_interface(self._dev, self._intf_num) - except usb.core.USBError: - pass + if self._claimed_intf is not None: + try: + usb.util.release_interface(self._dev, self._claimed_intf) + except usb.core.USBError: + pass + self._claimed_intf = None try: usb.util.dispose_resources(self._dev) except usb.core.USBError: @@ -205,7 +260,23 @@ def bulk_read(self, size=None, timeout=DEFAULT_TIMEOUT): return bytes(self._ep_in.read(size, timeout=timeout)) def __enter__(self): - self.open() + """Open on the way in, giving the claim back if opening fails. + + Python does not call __exit__ when __enter__ raises, so nothing + outside the with block can release the interface: the rollback + has to be here. This is the third place the same fault turned + up — construction, reconnect, and now here — because open() + can claim and then raise, and every caller is left to remember + that separately. + """ + try: + self.open() + except BaseException: + try: + self.close() + except Exception: + pass + raise return self def __exit__(self, *args): diff --git a/tests/test_chdk.py b/tests/test_chdk.py index b0346ce..f65ed52 100644 --- a/tests/test_chdk.py +++ b/tests/test_chdk.py @@ -379,6 +379,52 @@ def test_camera_that_never_clears_more_raises(self): chdk.remote_capture_get_data(1) +class TestCaptureChunkCount: + """How many chunks a still arrived in is a bench observation.""" + + def _make_chdk(self): + mock_session = MagicMock() + return ChdkPTP(mock_session), mock_session + + def test_it_starts_at_zero(self): + chdk, _ = self._make_chdk() + assert chdk.last_capture_chunks == 0 + + def test_it_counts_the_chunks_of_a_capture(self): + chdk, session = self._make_chdk() + session.transaction.side_effect = [ + ([4, 1, 0xFFFFFFFF], b"AAAA"), + ([4, 1, 0xFFFFFFFF], b"BBBB"), + ([4, 0, 0xFFFFFFFF], b"CCCC"), + ] + chdk.remote_capture_get_data(1) + assert chdk.last_capture_chunks == 3 + + def test_a_later_capture_does_not_inherit_the_count(self): + chdk, session = self._make_chdk() + session.transaction.side_effect = [ + ([4, 1, 0xFFFFFFFF], b"AAAA"), + ([4, 0, 0xFFFFFFFF], b"BBBB"), + ] + chdk.remote_capture_get_data(1) + assert chdk.last_capture_chunks == 2 + session.transaction.side_effect = [([4, 0, 0xFFFFFFFF], b"ZZZZ")] + chdk.remote_capture_get_data(1) + assert chdk.last_capture_chunks == 1 + + def test_the_count_survives_a_capture_that_failed_part_way(self): + chdk, session = self._make_chdk() + session.transaction.side_effect = [ + ([4, 1, 0xFFFFFFFF], b"AAAA"), + ([4, 1, 0xFFFFFFFF], b"BBBB"), + RuntimeError("cable"), + ] + with pytest.raises(RuntimeError, match="cable"): + chdk.remote_capture_get_data(1) + # Two arrived before it broke, which is worth knowing. + assert chdk.last_capture_chunks == 2 + + class TestRemoteCaptureGetChunk: def _make_chdk(self): mock_session = MagicMock() diff --git a/tests/test_device.py b/tests/test_device.py index a13053a..dd8e56d 100644 --- a/tests/test_device.py +++ b/tests/test_device.py @@ -8,13 +8,19 @@ import pychdk from pychdk import device from pychdk.chdk import ( + ChdkCommand, MessageType, REMOTE_CAP_NOTSET, ScriptDataType, ScriptErrorType, ScriptMessage, ) -from pychdk.device import ChdkDevice, list_devices, DeviceInfo +from pychdk.device import ( + ChdkDevice, + list_devices, + DeviceInfo, + _open_devices, +) class TestListDevices: @@ -39,6 +45,230 @@ def test_empty_when_no_cameras(self, mock_find): assert list_devices() == [] +class _FakeClock: + """Stands in for the time module inside pychdk.device. + + monotonic() walks a scripted sequence and then holds its last + value, so a thirty-second deadline can be reached in no time at + all; sleep() records what was asked for without waiting. + """ + + def __init__(self, readings): + self._readings = list(readings) + self.slept = 0.0 + + def monotonic(self): + if len(self._readings) > 1: + return self._readings.pop(0) + return self._readings[0] + + def sleep(self, seconds): + self.slept += seconds + + +class TestCaptureChunkCountThroughShoot: + """Callers use shoot(), so the count has to be reachable from there.""" + + def _device_with_a_real_protocol(self): + info = DeviceInfo( + vendor_id=0x04A9, product_id=0x1234, + bus_num=1, device_num=5, serial_num="ABC", + ) + mock_session = MagicMock() + with patch("pychdk.device.PTPDevice"), \ + patch("pychdk.device.PTPSession", return_value=mock_session): + dev = ChdkDevice(info, _usb_device=MagicMock()) + return dev, mock_session + + def test_shoot_reports_how_many_chunks_arrived(self): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), # execute_script, id 7 + ([0x01], b""), # ready, JPEG + ([4, 1, 0xFFFFFFFF], b"AAAA"), # chunk 1 + ([4, 0, 0xFFFFFFFF], b"BBBB"), # chunk 2, the last + ([0], b""), # drain: nothing waiting + ] + assert dev.shoot(stream=True) == b"AAAABBBB" + assert dev.last_capture_chunks == 2 + + def test_a_capture_that_never_downloads_reports_nothing(self): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), # first capture: two chunks + ([0x01], b""), + ([4, 1, 0xFFFFFFFF], b"AAAA"), + ([4, 0, 0xFFFFFFFF], b"BBBB"), + ([0], b""), + ] + assert dev.shoot(stream=True) == b"AAAABBBB" + assert dev.last_capture_chunks == 2 + + session.transaction.side_effect = [ + ([8, 0], b""), # second: script starts + ([0], b""), # nothing ready + ([0], b""), # and the script has ended + ] + with pytest.raises(RuntimeError, match="without producing a capture"): + dev.shoot(stream=True) + # No chunk arrived, so reporting two would be a lie a bench + # reader would believe. + assert dev.last_capture_chunks == 0 + + def test_a_refused_capture_reports_nothing(self): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), + ([0x01], b""), + ([4, 0, 0xFFFFFFFF], b"JPEG"), + ([0], b""), + ] + assert dev.shoot(stream=True) == b"JPEG" + assert dev.last_capture_chunks == 1 + + with pytest.raises(NotImplementedError): + dev.shoot(dng=True, stream=True) + assert dev.last_capture_chunks == 0 + + def test_a_second_capture_that_ends_without_data_reports_nothing( + self, monkeypatch, + ): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), + ([0x01], b""), + ([4, 0, 0xFFFFFFFF], b"JPEG"), + ([0], b""), + ] + assert dev.shoot(stream=True) == b"JPEG" + assert dev.last_capture_chunks == 1 + + # Nothing ready and the script already finished: this ends on + # the script-ended path, not at the deadline. See + # test_a_capture_that_runs_out_its_deadline_reports_nothing. + session.transaction.side_effect = None + session.transaction.return_value = ([0], b"") + monkeypatch.setattr("pychdk.device.CAPTURE_INIT_GRACE", 0.0) + with pytest.raises(RuntimeError, match="without producing a capture"): + dev.shoot(stream=True) + assert dev.last_capture_chunks == 0 + + def test_a_capture_that_runs_out_its_deadline_reports_nothing( + self, monkeypatch, + ): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), + ([0x01], b""), + ([4, 0, 0xFFFFFFFF], b"JPEG"), + ([0], b""), + ] + assert dev.shoot(stream=True) == b"JPEG" + assert dev.last_capture_chunks == 1 + + # A camera that stays busy and never becomes ready. The clock is + # driven rather than waited on: the deadline is thirty seconds + # and the suite must not spend them. + def respond(operation, params=None, **kwargs): + command = params[0] + if command == ChdkCommand.EXECUTE_SCRIPT: + return ([8, 0], b"") + if command == ChdkCommand.REMOTE_CAPTURE_IS_READY: + return ([0], b"") # never ready + if command == ChdkCommand.SCRIPT_STATUS: + return ([0b01], b"") # still running, nothing to say + return ([0], b"") + + session.transaction.side_effect = respond + clock = _FakeClock([0.0, 0.0, 0.0, 0.0, 999.0]) + monkeypatch.setattr("pychdk.device.time", clock) + + # TimeoutError, not RuntimeError: only the deadline raises this. + with pytest.raises(TimeoutError, match="did not complete"): + dev.shoot(stream=True) + assert dev.last_capture_chunks == 0 + # It really went round the loop rather than falling straight out. + assert clock.slept > 0 + + def test_a_one_chunk_still_reports_one(self): + dev, session = self._device_with_a_real_protocol() + session.transaction.side_effect = [ + ([7, 0], b""), + ([0x01], b""), + ([4, 0, 0xFFFFFFFF], b"JPEG"), + ([0], b""), + ] + assert dev.shoot(stream=True) == b"JPEG" + assert dev.last_capture_chunks == 1 + + +class TestConstructionIsExceptionSafe: + """A claim taken during construction must not outlive the failure.""" + + def _info(self): + return DeviceInfo( + vendor_id=0x04A9, product_id=0x1234, + bus_num=1, device_num=5, serial_num="ABC", + ) + + def test_a_failed_session_releases_the_transport(self): + tracked_before = len(_open_devices) + with patch("pychdk.device.PTPDevice") as MockTransport, \ + patch("pychdk.device.PTPSession") as MockSession, \ + patch("pychdk.device.ChdkPTP"): + MockSession.return_value.open.side_effect = RuntimeError( + "session refused", + ) + with pytest.raises(RuntimeError, match="session refused"): + ChdkDevice(self._info(), _usb_device=MagicMock()) + transport = MockTransport.return_value + transport.open.assert_called_once() + # One open, one close: the claim does not survive the raise. + transport.close.assert_called_once() + assert len(_open_devices) == tracked_before + + def test_a_failed_transport_open_is_also_released(self): + tracked_before = len(_open_devices) + with patch("pychdk.device.PTPDevice") as MockTransport, \ + patch("pychdk.device.PTPSession"), \ + patch("pychdk.device.ChdkPTP"): + # A transport that claims the interface and then fails + # finding endpoints raises out of open() itself. + MockTransport.return_value.open.side_effect = RuntimeError( + "Could not find bulk endpoints on PTP device", + ) + with pytest.raises(RuntimeError, match="bulk endpoints"): + ChdkDevice(self._info(), _usb_device=MagicMock()) + MockTransport.return_value.close.assert_called_once() + assert len(_open_devices) == tracked_before + + def test_a_failed_construction_tracks_nothing(self): + tracked_before = len(_open_devices) + with patch("pychdk.device.PTPDevice"), \ + patch("pychdk.device.PTPSession") as MockSession, \ + patch("pychdk.device.ChdkPTP"): + MockSession.return_value.open.side_effect = RuntimeError("nope") + with pytest.raises(RuntimeError): + ChdkDevice(self._info(), _usb_device=MagicMock()) + # Nothing for _cleanup_all to find, and no half-built device. + assert len(_open_devices) == tracked_before + + def test_a_failed_reconnect_also_releases_the_transport(self): + with patch("pychdk.device.PTPDevice") as MockTransport, \ + patch("pychdk.device.PTPSession") as MockSession, \ + patch("pychdk.device.ChdkPTP"): + dev = ChdkDevice(self._info(), _usb_device=MagicMock()) + transport = MockTransport.return_value + transport.close.reset_mock() + MockSession.return_value.open.side_effect = RuntimeError("gone") + with pytest.raises(RuntimeError, match="gone"): + dev.reconnect(wait=0) + # Closed once on the way down, once releasing the failed open. + assert transport.close.call_count == 2 + assert dev not in _open_devices + assert not dev.is_connected + + class TestChdkDevice: def _make_device(self): info = DeviceInfo( diff --git a/tests/test_multicam.py b/tests/test_multicam.py index 6d0bd20..1c81a7e 100644 --- a/tests/test_multicam.py +++ b/tests/test_multicam.py @@ -16,6 +16,39 @@ def test_discovers_cameras(self, MockDevice, mock_list): mc = MultiCam() assert len(mc.cameras) == 2 + @patch("pychdk.multicam.list_devices") + @patch("pychdk.multicam.ChdkDevice") + def test_a_second_camera_that_fails_closes_the_first( + self, MockDevice, mock_list, + ): + mock_list.return_value = [ + DeviceInfo(0x04A9, 0x1234, 1, 5, "AAA"), + DeviceInfo(0x04A9, 0x1234, 1, 6, "BBB"), + ] + first = MagicMock() + MockDevice.side_effect = [first, RuntimeError("camera 2 will not open")] + + with pytest.raises(RuntimeError, match="camera 2 will not open"): + MultiCam() + + # The half-built MultiCam is discarded, so nothing else can + # ever close camera one: it would be claimed by a process with + # no handle on it until the card was unplugged. + first.close.assert_called_once() + + @patch("pychdk.multicam.list_devices") + @patch("pychdk.multicam.ChdkDevice") + def test_a_first_camera_that_fails_closes_nothing( + self, MockDevice, mock_list, + ): + mock_list.return_value = [ + DeviceInfo(0x04A9, 0x1234, 1, 5, "AAA"), + DeviceInfo(0x04A9, 0x1234, 1, 6, "BBB"), + ] + MockDevice.side_effect = RuntimeError("camera 1 will not open") + with pytest.raises(RuntimeError, match="camera 1 will not open"): + MultiCam() + @patch("pychdk.multicam.list_devices") def test_no_cameras_raises(self, mock_list): mock_list.return_value = [] diff --git a/tests/test_usb_transport.py b/tests/test_usb_transport.py index 022e9eb..d275401 100644 --- a/tests/test_usb_transport.py +++ b/tests/test_usb_transport.py @@ -55,6 +55,28 @@ def _make_mock_usb_device(vendor_id=0x04A9, product_id=0xABCD, return dev +def _make_mock_usb_device_without_bulk_endpoints(): + """A PTP interface that claims fine and then has nothing to talk on.""" + ep_int = _make_mock_endpoint(0x83) + ep_int.bmAttributes = 0x03 # interrupt only + + interface = MagicMock() + interface.bInterfaceClass = 6 + interface.bInterfaceSubClass = 1 + interface.bInterfaceProtocol = 1 + interface.bInterfaceNumber = 0 + interface.__iter__ = lambda self: iter([ep_int]) + + config = MagicMock() + config.__iter__ = lambda self: iter([interface]) + config.__getitem__ = lambda self, key: interface + + dev = MagicMock() + dev.__iter__ = lambda self: iter([config]) + dev.__getitem__ = lambda self, i: config + return dev + + class TestFindPTPDevices: @patch("pychdk.usb_transport.usb.core.find") def test_finds_canon_ptp_devices(self, mock_find): @@ -84,6 +106,86 @@ def test_close_releases_interface(self): ptp.close() assert not ptp._is_open + @patch("pychdk.usb_transport.usb.util.release_interface") + @patch("pychdk.usb_transport.usb.util.claim_interface") + def test_a_claim_is_released_when_endpoint_discovery_fails( + self, mock_claim, mock_release, + ): + mock_dev = _make_mock_usb_device_without_bulk_endpoints() + ptp = PTPDevice(mock_dev) + with pytest.raises(RuntimeError, match="bulk endpoints"): + ptp.open() + # The interface was taken before discovery failed. + mock_claim.assert_called_once_with(mock_dev, 0) + ptp.close() + # So closing has to give it back, however far open() got. + mock_release.assert_called_once_with(mock_dev, 0) + + @patch("pychdk.usb_transport.usb.util.release_interface") + def test_closing_a_device_that_never_opened_releases_nothing( + self, mock_release, + ): + ptp = PTPDevice(_make_mock_usb_device()) + ptp.close() + mock_release.assert_not_called() + + @patch("pychdk.usb_transport.usb.util.release_interface") + @patch("pychdk.usb_transport.usb.util.claim_interface") + def test_a_claim_is_given_back_only_once(self, mock_claim, mock_release): + mock_dev = _make_mock_usb_device() + ptp = PTPDevice(mock_dev) + ptp.open() + ptp.close() + ptp.close() + mock_release.assert_called_once_with(mock_dev, 0) + + @patch("pychdk.usb_transport.usb.util.release_interface") + @patch("pychdk.usb_transport.usb.util.claim_interface") + def test_an_open_that_raises_holds_nothing(self, mock_claim, mock_release): + mock_dev = _make_mock_usb_device_without_bulk_endpoints() + ptp = PTPDevice(mock_dev) + with pytest.raises(RuntimeError, match="bulk endpoints"): + ptp.open() + # No caller did anything here: open() gave the claim back + # itself, so "open raised" means "nothing is held" without + # anyone having to remember a rollback. + mock_claim.assert_called_once_with(mock_dev, 0) + mock_release.assert_called_once_with(mock_dev, 0) + assert ptp._claimed_intf is None + assert not ptp._is_open + + @patch("pychdk.usb_transport.usb.util.release_interface") + @patch("pychdk.usb_transport.usb.util.claim_interface") + def test_a_failed_open_disables_the_pyusb_finalizer( + self, mock_claim, mock_release, + ): + mock_dev = _make_mock_usb_device_without_bulk_endpoints() + mock_dev._finalize_called = False + ptp = PTPDevice(mock_dev) + with pytest.raises(RuntimeError, match="bulk endpoints"): + ptp.open() + # Releasing the claim is not enough: a device we opened far + # enough to touch must not be left to pyusb's finalizer, which + # can reopen a freed context at shutdown and take the process + # with it. Trading a leak for a crash is the worse bargain. + assert mock_dev._finalize_called is True + + @patch("pychdk.usb_transport.usb.util.release_interface") + @patch("pychdk.usb_transport.usb.util.claim_interface") + def test_a_failed_open_in_a_with_block_claims_nothing( + self, mock_claim, mock_release, + ): + mock_dev = _make_mock_usb_device_without_bulk_endpoints() + ptp = PTPDevice(mock_dev) + # __exit__ never runs when __enter__ raises, so nothing outside + # the block can give the interface back. + with pytest.raises(RuntimeError, match="bulk endpoints"): + with ptp: + pass + mock_claim.assert_called_once_with(mock_dev, 0) + mock_release.assert_called_once_with(mock_dev, 0) + assert ptp._claimed_intf is None + def test_bulk_write(self): mock_dev = _make_mock_usb_device() ptp = PTPDevice(mock_dev)