From 88d61cff86732c7e3cc12cd641faf6f31b1683c8 Mon Sep 17 00:00:00 2001 From: Johannes Schrimpf Date: Tue, 9 Jun 2026 13:16:18 +0200 Subject: [PATCH 1/5] Add VGA/2K/4K resolution support Add 480 (VGA), 1440 (2K) and 2160 (4K) to the camera resolution property, using the enum members that actually exist in the protocol (RESOLUTION_VGA_480P, RESOLUTION_QHD_2K, RESOLUTION_UHD_4K). Add tests for the resolution mapping in both directions and for the existing streaming_protocol property, which was previously untested. Co-Authored-By: Claude Opus 4.8 (1M context) --- blueye/sdk/camera.py | 26 +++++++++++++---- tests/test_camera.py | 68 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 6 deletions(-) diff --git a/blueye/sdk/camera.py b/blueye/sdk/camera.py index 71ce2e3c..c8bf39d1 100644 --- a/blueye/sdk/camera.py +++ b/blueye/sdk/camera.py @@ -1068,32 +1068,46 @@ def get_resolution(self) -> int: The camera resolution. """ self._update_camera_parameters() - if self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_HD_720P: + if self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_VGA_480P: + return 480 + elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_HD_720P: return 720 elif ( self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P ): return 1080 + elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_QHD_2K: + return 1440 + elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_UHD_4K: + return 2160 def set_resolution(self, resolution: int): """Set the camera resolution. Args: - resolution (int): Set the camera in vertical pixels. Valid values are 720 or 1080. + resolution (int): Set the camera in vertical pixels. Valid values are 480, 720, 1080, + 1440 or 2160. Raises: - ValueError: If the resolution is not 720 or 1080. + ValueError: If the resolution is not one of the valid values. """ - if resolution not in (720, 1080): + if resolution not in (480, 720, 1080, 1440, 2160): raise ValueError( - f"{resolution} is not a valid resolution. Valid values are 720 or 1080" + f"{resolution} is not a valid resolution. " + "Valid values are 480, 720, 1080, 1440 or 2160" ) if self._camera_parameters is None: self._update_camera_parameters() - if resolution == 720: + if resolution == 480: + self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_VGA_480P + elif resolution == 720: self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_HD_720P elif resolution == 1080: self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P + elif resolution == 1440: + self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_QHD_2K + elif resolution == 2160: + self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_UHD_4K self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) diff --git a/tests/test_camera.py b/tests/test_camera.py index bf2a07d4..501213fe 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -84,6 +84,74 @@ def test_recording_resolution_invalid_type(mocked_camera): mocked_camera.set_recording_resolution("invalid_resolution") +@pytest.mark.parametrize( + "enum_value, expected", + [ + (bp.Resolution.RESOLUTION_VGA_480P, 480), + (bp.Resolution.RESOLUTION_HD_720P, 720), + (bp.Resolution.RESOLUTION_FULLHD_1080P, 1080), + (bp.Resolution.RESOLUTION_QHD_2K, 1440), + (bp.Resolution.RESOLUTION_UHD_4K, 2160), + ], +) +def test_resolution_getter(mocked_camera, enum_value, expected): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=enum_value) + ) + assert mocked_camera.get_resolution() == expected + + +@pytest.mark.parametrize( + "value, expected_enum", + [ + (480, bp.Resolution.RESOLUTION_VGA_480P), + (720, bp.Resolution.RESOLUTION_HD_720P), + (1080, bp.Resolution.RESOLUTION_FULLHD_1080P), + (1440, bp.Resolution.RESOLUTION_QHD_2K), + (2160, bp.Resolution.RESOLUTION_UHD_4K), + ], +) +def test_resolution_setter(mocked_camera, value, expected_enum): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_resolution(value) + assert mocked_camera._camera_parameters.resolution == expected_enum + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_resolution_setter_invalid_value(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_resolution(600) + + +def test_streaming_protocol_getter(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(streaming_protocol=bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264) + ) + assert ( + mocked_camera.get_streaming_protocol() + == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264 + ) + + +def test_streaming_protocol_setter(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_streaming_protocol(bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_MJPEG) + assert ( + mocked_camera._camera_parameters.streaming_protocol + == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_MJPEG + ) + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_streaming_protocol_invalid_type(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_streaming_protocol("invalid_protocol") + + def test_old_drones_use_resolution_field(mocked_camera): # Set the version to a value that does not support separate recording resolution mocked_camera._parent_drone.software_version_short = "4.3" From fe2c53c72d43b105eec3b844415daf73e7f3be96 Mon Sep 17 00:00:00 2001 From: Johannes Schrimpf Date: Wed, 19 Aug 2026 15:17:25 +0200 Subject: [PATCH 2/5] Complete the camera parameter coverage against the protocol blueye.protocol 3.3.0 exposes more of CameraParameters than the SDK did: * Add 60 fps to the framerate property. The Ultra supports it at 1440p and below, p2_drone and gst_rtsp_record already map and cap it. * Add the Ultra image parameters (brightness, contrast, saturation, gamma, sharpness, backlight_compensation, denoise and the eHDR options), the ISO gain, and the streaming MTU size. All of them were already handled by the drone but had no SDK property. * Deprecate the resolution property. Drones running Blunux 4.4 or newer take the resolution from stream_resolution/recording_resolution and ignore this field when setting parameters, so assigning to it silently does nothing. The integration test xfails on those drones, and a stream_resolution round trip takes its place. * Raise instead of returning None when the drone reports a resolution or frame rate the SDK does not recognize, and map both through lookup tables rather than if/elif chains. Co-Authored-By: Claude Opus 5 (1M context) --- blueye/sdk/camera.py | 451 ++++++++++++++++++++++++++++++++++---- tests/test_camera.py | 196 ++++++++++++++++- tests/test_integration.py | 18 ++ 3 files changed, 626 insertions(+), 39 deletions(-) diff --git a/blueye/sdk/camera.py b/blueye/sdk/camera.py index c8bf39d1..bec66861 100644 --- a/blueye/sdk/camera.py +++ b/blueye/sdk/camera.py @@ -18,6 +18,22 @@ logger = logging.getLogger(__name__) +_RESOLUTION_TO_HEIGHT = { + blueye.protocol.Resolution.RESOLUTION_VGA_480P: 480, + blueye.protocol.Resolution.RESOLUTION_HD_720P: 720, + blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P: 1080, + blueye.protocol.Resolution.RESOLUTION_QHD_2K: 1440, + blueye.protocol.Resolution.RESOLUTION_UHD_4K: 2160, +} +_HEIGHT_TO_RESOLUTION = {height: res for res, height in _RESOLUTION_TO_HEIGHT.items()} + +_FRAMERATE_TO_FPS = { + blueye.protocol.Framerate.FRAMERATE_FPS_25: 25, + blueye.protocol.Framerate.FRAMERATE_FPS_30: 30, + blueye.protocol.Framerate.FRAMERATE_FPS_60: 60, +} +_FPS_TO_FRAMERATE = {fps: framerate for framerate, fps in _FRAMERATE_TO_FPS.items()} + class Tilt: """Handles the camera tilt functionality for the Blueye drone.""" @@ -831,6 +847,17 @@ def __getattr__(self, name): _VERSION_GATED_PARAMS = { "recording_bitrate": "5.0.0", "recording_codec": "5.0.0", + "brightness": "5.0.0", + "contrast": "5.0.0", + "saturation": "5.0.0", + "gamma": "5.0.0", + "sharpness": "5.0.0", + "backlight_compensation": "5.0.0", + "denoise": "5.0.0", + "ehdr_enabled": "5.0.0", + "ehdr_exposure_min_number": "5.0.0", + "ehdr_exposure_max_number": "5.0.0", + "mtu_size": "5.1.0", } def __setattr__(self, name, value): @@ -1061,29 +1088,352 @@ def set_hue(self, hue: int): hue = deprecated_property("get_hue", "set_hue") + def get_gain(self) -> float: + """Get the camera gain. + + Only available on Pioneer/Pro/X1/X3. + + Returns: + The camera ISO gain. + """ + self._update_camera_parameters() + return self._camera_parameters.gain + + def set_gain(self, gain: float): + """Set the camera gain. + + Only available on Pioneer/Pro/X1/X3. + + Args: + gain (float): Set the ISO gain (0..1). + """ + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.gain = gain + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_brightness(self) -> int: + """Get the camera brightness. + + Only available on Ultra. + + Returns: + The camera brightness. + """ + self._update_camera_parameters() + return self._camera_parameters.brightness + + def set_brightness(self, brightness: int): + """Set the camera brightness. + + Only available on Ultra. + + Args: + brightness (int): Set the brightness (-10..10), 0 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.brightness = brightness + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_contrast(self) -> int: + """Get the camera contrast. + + Only available on Ultra. + + Returns: + The camera contrast. + """ + self._update_camera_parameters() + return self._camera_parameters.contrast + + def set_contrast(self, contrast: int): + """Set the camera contrast. + + Only available on Ultra. + + Args: + contrast (int): Set the contrast (-50..50), 0 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.contrast = contrast + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_saturation(self) -> int: + """Get the camera saturation. + + Only available on Ultra. + + Returns: + The camera saturation. + """ + self._update_camera_parameters() + return self._camera_parameters.saturation + + def set_saturation(self, saturation: int): + """Set the camera saturation. + + Only available on Ultra. + + Args: + saturation (int): Set the saturation (0..50), 8 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.saturation = saturation + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_gamma(self) -> int: + """Get the camera gamma. + + Only available on Ultra. + + Returns: + The camera gamma. + """ + self._update_camera_parameters() + return self._camera_parameters.gamma + + def set_gamma(self, gamma: int): + """Set the camera gamma. + + Only available on Ultra. + + Args: + gamma (int): Set the gamma (4..79), 22 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.gamma = gamma + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_sharpness(self) -> int: + """Get the camera sharpness. + + Only available on Ultra. + + Returns: + The camera sharpness. + """ + self._update_camera_parameters() + return self._camera_parameters.sharpness + + def set_sharpness(self, sharpness: int): + """Set the camera sharpness. + + Only available on Ultra. + + Args: + sharpness (int): Set the sharpness (-20..20), -20 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.sharpness = sharpness + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_backlight_compensation(self) -> int: + """Get the camera backlight compensation. + + Only available on Ultra. + + Returns: + The camera backlight compensation. + """ + self._update_camera_parameters() + return self._camera_parameters.backlight_compensation + + def set_backlight_compensation(self, compensation: int): + """Set the camera backlight compensation. + + Only available on Ultra. + + Args: + compensation (int): Set the backlight compensation (-150..150), 10 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.backlight_compensation = compensation + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_denoise(self) -> int: + """Get the camera noise reduction. + + Only available on Ultra. + + Returns: + The camera noise reduction. + """ + self._update_camera_parameters() + return self._camera_parameters.denoise + + def set_denoise(self, denoise: int): + """Set the camera noise reduction. + + Only available on Ultra. + + Args: + denoise (int): Set the noise reduction (-20..20), -20 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.denoise = denoise + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def is_ehdr_enabled(self) -> bool: + """Get the state of the eHDR mode. + + Only available on Ultra. + + Returns: + The current state of the eHDR mode. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_enabled + + def enable_ehdr(self, enable_ehdr: bool): + """Enable or disable the eHDR mode. + + Only available on Ultra. + + Args: + enable_ehdr (bool): True to enable eHDR mode, False to disable it. Enabled by default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_enabled = enable_ehdr + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_ehdr_exposure_min_number(self) -> int: + """Get the minimum number of eHDR frames. + + Only available on Ultra. + + Returns: + The minimum number of eHDR frames. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_exposure_min_number + + def set_ehdr_exposure_min_number(self, number: int): + """Set the minimum number of eHDR frames. + + Only available on Ultra. + + Args: + number (int): Set the minimum number of eHDR frames (1..4), 1 is the default. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_exposure_min_number = number + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + + def get_ehdr_exposure_max_number(self) -> int: + """Get the maximum number of eHDR frames. + + Only available on Ultra. + + Returns: + The maximum number of eHDR frames. + """ + self._update_camera_parameters() + return self._camera_parameters.ehdr_exposure_max_number + + def set_ehdr_exposure_max_number(self, number: int): + """Set the maximum number of eHDR frames. + + Only available on Ultra. + + Args: + number (int): Set the maximum number of eHDR frames (1..4), 2 is the default. Setting + it higher than 2 can reduce the frame rate. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.0.0. + """ + self._parent_drone._verify_required_blunux_version("5.0.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.ehdr_exposure_max_number = number + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + def get_resolution(self) -> int: """Get the camera resolution. + Deprecated: + Drones running Blunux 4.4 or newer take the resolution from the stream and recording + resolution fields. Use + [`get_stream_resolution`][blueye.sdk.camera.Camera.get_stream_resolution] and + [`get_recording_resolution`][blueye.sdk.camera.Camera.get_recording_resolution] + instead. + + Raises: + RuntimeError: If the drone reports a resolution the SDK does not recognize. + Returns: The camera resolution. """ + warnings.warn( + "`Camera.get_resolution` is deprecated, use `Camera.get_stream_resolution` or " + "`Camera.get_recording_resolution` instead", + DeprecationWarning, + stacklevel=2, + ) self._update_camera_parameters() - if self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_VGA_480P: - return 480 - elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_HD_720P: - return 720 - elif ( - self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P - ): - return 1080 - elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_QHD_2K: - return 1440 - elif self._camera_parameters.resolution == blueye.protocol.Resolution.RESOLUTION_UHD_4K: - return 2160 + try: + return _RESOLUTION_TO_HEIGHT[self._camera_parameters.resolution] + except KeyError: + raise RuntimeError( + "Drone reported an unsupported resolution: " + f"{self._camera_parameters.resolution.name}" + ) from None def set_resolution(self, resolution: int): """Set the camera resolution. + Deprecated: + Drones running Blunux 4.4 or newer ignore this field when camera parameters are set, + so calling this method has no effect on them. Use + [`set_stream_resolution`][blueye.sdk.camera.Camera.set_stream_resolution] and + [`set_recording_resolution`][blueye.sdk.camera.Camera.set_recording_resolution] + instead. + Args: resolution (int): Set the camera in vertical pixels. Valid values are 480, 720, 1080, 1440 or 2160. @@ -1091,24 +1441,21 @@ def set_resolution(self, resolution: int): Raises: ValueError: If the resolution is not one of the valid values. """ - if resolution not in (480, 720, 1080, 1440, 2160): + warnings.warn( + "`Camera.set_resolution` is deprecated and is ignored by drones running Blunux 4.4 or " + "newer, use `Camera.set_stream_resolution` or `Camera.set_recording_resolution` " + "instead", + DeprecationWarning, + stacklevel=2, + ) + if resolution not in _HEIGHT_TO_RESOLUTION: raise ValueError( f"{resolution} is not a valid resolution. " "Valid values are 480, 720, 1080, 1440 or 2160" ) if self._camera_parameters is None: self._update_camera_parameters() - if resolution == 480: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_VGA_480P - elif resolution == 720: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_HD_720P - elif resolution == 1080: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_FULLHD_1080P - elif resolution == 1440: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_QHD_2K - elif resolution == 2160: - self._camera_parameters.resolution = blueye.protocol.Resolution.RESOLUTION_UHD_4K - + self._camera_parameters.resolution = _HEIGHT_TO_RESOLUTION[resolution] self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) resolution = deprecated_property("get_resolution", "set_resolution") @@ -1200,33 +1547,38 @@ def set_recording_resolution(self, resolution: blueye.protocol.Resolution): def get_framerate(self) -> int: """Get the camera frame rate. + Raises: + RuntimeError: If the drone reports a frame rate the SDK does not recognize. + Returns: The camera frame rate. """ self._update_camera_parameters() - if self._camera_parameters.framerate == blueye.protocol.Framerate.FRAMERATE_FPS_25: - return 25 - elif self._camera_parameters.framerate == blueye.protocol.Framerate.FRAMERATE_FPS_30: - return 30 + try: + return _FRAMERATE_TO_FPS[self._camera_parameters.framerate] + except KeyError: + raise RuntimeError( + "Drone reported an unsupported framerate: " + f"{self._camera_parameters.framerate.name}" + ) from None def set_framerate(self, framerate: int): """Set the camera frame rate. Args: - framerate (int): Set the camera frame rate in frames per second. - Valid values are 25 or 30. + framerate (int): Set the camera frame rate in frames per second. Valid values are 25, + 30 or 60. 25 fps is only supported on Pioneer/Pro/X1/X3, and 60 fps only on the + Ultra at 1440p or lower. If the requested frame rate is not supported at the + current resolution the drone reduces it while respecting the resolution. Raises: - ValueError: If the framerate is not 25 or 30. + ValueError: If the frame rate is not 25, 30 or 60. """ - if framerate not in (25, 30): - raise ValueError(f"{framerate} is not a valid framerate. Valid values are 25 or 30") + if framerate not in _FPS_TO_FRAMERATE: + raise ValueError(f"{framerate} is not a valid framerate. Valid values are 25, 30 or 60") if self._camera_parameters is None: self._update_camera_parameters() - if framerate == 25: - self._camera_parameters.framerate = blueye.protocol.Framerate.FRAMERATE_FPS_25 - elif framerate == 30: - self._camera_parameters.framerate = blueye.protocol.Framerate.FRAMERATE_FPS_30 + self._camera_parameters.framerate = _FPS_TO_FRAMERATE[framerate] self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) framerate = deprecated_property("get_framerate", "set_framerate") @@ -1321,6 +1673,31 @@ def set_streaming_protocol(self, protocol: blueye.protocol.StreamingProtocol): streaming_protocol = deprecated_property("get_streaming_protocol", "set_streaming_protocol") + def get_mtu_size(self) -> int: + """Get the network MTU size used for video streaming. + + Returns: + The MTU size in bytes (0 if the drone default is used). + """ + self._update_camera_parameters() + return self._camera_parameters.mtu_size + + def set_mtu_size(self, mtu_size: int): + """Set the network MTU size used for video streaming. + + Args: + mtu_size (int): Set the MTU size in bytes (68..65535). The Blueye app allows values + between 500 and 1460. A value of 0 means the drone uses its default of 1400. + + Raises: + RuntimeError: If the connected drone is running a Blunux version older than 5.1.0. + """ + self._parent_drone._verify_required_blunux_version("5.1.0") + if self._camera_parameters is None: + self._update_camera_parameters() + self._camera_parameters.mtu_size = mtu_size + self._parent_drone._req_rep_client.set_camera_parameters(self._camera_parameters) + def get_record_time(self) -> Optional[int]: """Get the duration of the current camera recording. diff --git a/tests/test_camera.py b/tests/test_camera.py index 501213fe..5fc46735 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -94,6 +94,7 @@ def test_recording_resolution_invalid_type(mocked_camera): (bp.Resolution.RESOLUTION_UHD_4K, 2160), ], ) +@pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_resolution_getter(mocked_camera, enum_value, expected): mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( bp.CameraParameters(resolution=enum_value) @@ -111,6 +112,7 @@ def test_resolution_getter(mocked_camera, enum_value, expected): (2160, bp.Resolution.RESOLUTION_UHD_4K), ], ) +@pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_resolution_setter(mocked_camera, value, expected_enum): mocked_camera._camera_parameters = bp.CameraParameters() mocked_camera.set_resolution(value) @@ -120,6 +122,7 @@ def test_resolution_setter(mocked_camera, value, expected_enum): ) +@pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_resolution_setter_invalid_value(mocked_camera): with pytest.raises(ValueError): mocked_camera.set_resolution(600) @@ -130,8 +133,7 @@ def test_streaming_protocol_getter(mocked_camera): bp.CameraParameters(streaming_protocol=bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264) ) assert ( - mocked_camera.get_streaming_protocol() - == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264 + mocked_camera.get_streaming_protocol() == bp.StreamingProtocol.STREAMING_PROTOCOL_RTSP_H264 ) @@ -218,3 +220,193 @@ def test_configure_passes_timeout(mocked_camera): mocked_camera._parent_drone._req_rep_client.set_camera_parameters.call_args[1]["timeout"] == 5.0 ) + + +@pytest.fixture +def mocked_ultra_camera(mocked_drone: Drone): + """A camera on a drone running a Blunux version that supports all camera parameters.""" + from blueye.sdk.camera import Camera + + mocked_drone.software_version_short = "5.1.0" + return Camera(mocked_drone) + + +def test_get_resolution_warns_deprecation(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=bp.Resolution.RESOLUTION_HD_720P) + ) + with pytest.warns(DeprecationWarning, match="get_stream_resolution"): + assert mocked_camera.get_resolution() == 720 + + +def test_set_resolution_warns_deprecation(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + with pytest.warns(DeprecationWarning, match="set_stream_resolution"): + mocked_camera.set_resolution(720) + + +@pytest.mark.filterwarnings("ignore::DeprecationWarning") +def test_resolution_getter_raises_on_unknown_resolution(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(resolution=bp.Resolution.RESOLUTION_UNSPECIFIED) + ) + with pytest.raises(RuntimeError): + mocked_camera.get_resolution() + + +@pytest.mark.parametrize( + "enum_value, expected", + [ + (bp.Framerate.FRAMERATE_FPS_25, 25), + (bp.Framerate.FRAMERATE_FPS_30, 30), + (bp.Framerate.FRAMERATE_FPS_60, 60), + ], +) +def test_framerate_getter(mocked_camera, enum_value, expected): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(framerate=enum_value) + ) + assert mocked_camera.get_framerate() == expected + + +@pytest.mark.parametrize( + "value, expected_enum", + [ + (25, bp.Framerate.FRAMERATE_FPS_25), + (30, bp.Framerate.FRAMERATE_FPS_30), + (60, bp.Framerate.FRAMERATE_FPS_60), + ], +) +def test_framerate_setter(mocked_camera, value, expected_enum): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_framerate(value) + assert mocked_camera._camera_parameters.framerate == expected_enum + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_framerate_setter_invalid_value(mocked_camera): + with pytest.raises(ValueError): + mocked_camera.set_framerate(24) + + +def test_framerate_getter_raises_on_unknown_framerate(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(framerate=bp.Framerate.FRAMERATE_UNSPECIFIED) + ) + with pytest.raises(RuntimeError): + mocked_camera.get_framerate() + + +# (getter, setter, CameraParameters field name, test value) for the Ultra-only image parameters +ULTRA_IMAGE_PARAMETERS = [ + ("get_brightness", "set_brightness", "brightness", 5), + ("get_contrast", "set_contrast", "contrast", 20), + ("get_saturation", "set_saturation", "saturation", 10), + ("get_gamma", "set_gamma", "gamma", 30), + ("get_sharpness", "set_sharpness", "sharpness", -10), + ("get_backlight_compensation", "set_backlight_compensation", "backlight_compensation", 100), + ("get_denoise", "set_denoise", "denoise", -5), + ("is_ehdr_enabled", "enable_ehdr", "ehdr_enabled", True), + ( + "get_ehdr_exposure_min_number", + "set_ehdr_exposure_min_number", + "ehdr_exposure_min_number", + 2, + ), + ( + "get_ehdr_exposure_max_number", + "set_ehdr_exposure_max_number", + "ehdr_exposure_max_number", + 3, + ), +] + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_getter(mocked_ultra_camera, getter, setter, field_name, value): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(**{field_name: value}) + ) + assert getattr(mocked_ultra_camera, getter)() == value + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_setter(mocked_ultra_camera, getter, setter, field_name, value): + mocked_ultra_camera._camera_parameters = bp.CameraParameters() + getattr(mocked_ultra_camera, setter)(value) + assert getattr(mocked_ultra_camera._camera_parameters, field_name) == value + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_ultra_camera._camera_parameters + ) + + +@pytest.mark.parametrize("getter, setter, field_name, value", ULTRA_IMAGE_PARAMETERS) +def test_ultra_image_parameter_setter_requires_blunux_5( + mocked_camera, getter, setter, field_name, value +): + # The mocked_camera fixture runs Blunux 4.4.1 + with pytest.raises(RuntimeError): + getattr(mocked_camera, setter)(value) + + +def test_gain_getter(mocked_camera): + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(gain=0.5) + ) + assert mocked_camera.get_gain() == pytest.approx(0.5) + + +def test_gain_setter(mocked_camera): + mocked_camera._camera_parameters = bp.CameraParameters() + mocked_camera.set_gain(0.5) + assert mocked_camera._camera_parameters.gain == pytest.approx(0.5) + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_camera._camera_parameters + ) + + +def test_mtu_size_getter(mocked_ultra_camera): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(mtu_size=1400) + ) + assert mocked_ultra_camera.get_mtu_size() == 1400 + + +def test_mtu_size_setter(mocked_ultra_camera): + mocked_ultra_camera._camera_parameters = bp.CameraParameters() + mocked_ultra_camera.set_mtu_size(1200) + assert mocked_ultra_camera._camera_parameters.mtu_size == 1200 + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + mocked_ultra_camera._camera_parameters + ) + + +def test_mtu_size_setter_requires_blunux_5_1(mocked_camera): + # The mocked_camera fixture runs Blunux 4.4.1 + with pytest.raises(RuntimeError): + mocked_camera.set_mtu_size(1200) + + +def test_configure_rejects_ultra_image_parameters_on_old_drone(mocked_camera): + """Batched changes should honour the same Blunux version requirements as the setters.""" + mocked_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters() + ) + with pytest.raises(RuntimeError): + with mocked_camera.configure() as params: + params.brightness = 5 + mocked_camera._parent_drone._req_rep_client.set_camera_parameters.assert_not_called() + + +def test_configure_rejects_mtu_size_on_blunux_5_0(mocked_drone: Drone): + from blueye.sdk.camera import Camera + + mocked_drone.software_version_short = "5.0.0" + camera = Camera(mocked_drone) + camera._parent_drone._req_rep_client.get_camera_parameters.return_value = bp.CameraParameters() + with pytest.raises(RuntimeError): + with camera.configure() as params: + params.mtu_size = 1200 + camera._parent_drone._req_rep_client.set_camera_parameters.assert_not_called() diff --git a/tests/test_integration.py b/tests/test_integration.py index a340d600..e3d0dd37 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -1,6 +1,8 @@ from time import time +import blueye.protocol as bp import pytest +from packaging import version def polling_assert_with_timeout(getter, value_to_wait_for, timeout): @@ -74,13 +76,29 @@ def test_camera_hue(self, real_drone): real_drone.camera.set_hue(30) polling_assert_with_timeout(real_drone.camera.get_hue, 30, 1) + @pytest.mark.filterwarnings("ignore::DeprecationWarning") def test_camera_resolution(self, real_drone): + if version.parse(real_drone.software_version_short) >= version.parse("4.4"): + pytest.xfail( + "Drones running Blunux 4.4 or newer ignore the deprecated resolution field when " + "camera parameters are set, use the stream/recording resolution methods instead" + ) _ = real_drone.camera.get_resolution() real_drone.camera.set_resolution(720) polling_assert_with_timeout(real_drone.camera.get_resolution, 720, 1) real_drone.camera.set_resolution(1080) polling_assert_with_timeout(real_drone.camera.get_resolution, 1080, 1) + def test_camera_stream_resolution(self, real_drone): + original_resolution = real_drone.camera.get_stream_resolution() + try: + real_drone.camera.set_stream_resolution(bp.Resolution.RESOLUTION_HD_720P) + polling_assert_with_timeout( + real_drone.camera.get_stream_resolution, bp.Resolution.RESOLUTION_HD_720P, 3 + ) + finally: + real_drone.camera.set_stream_resolution(original_resolution) + def test_camera_framerate(self, real_drone): _ = real_drone.camera.get_framerate() real_drone.camera.set_framerate(25) From 9c76f7bc8a21d2a903c733406773cbdb9bc7194d Mon Sep 17 00:00:00 2001 From: Johannes Schrimpf Date: Thu, 20 Aug 2026 14:01:13 +0200 Subject: [PATCH 3/5] Cover the camera parameter cache contracts and recording_bitrate recording_bitrate had no test coverage at all - no getter, no setter and no version gate - despite being the property our bench integration suite leans on hardest. Add those three, following the shapes of the existing parameter tests. Add the two cache contracts callers actually depend on: * Getters refresh from the drone rather than answering from the cache, so a read-back is trustworthy after the drone resolves an automatic request to a concrete value. * A setter on a camera that has not talked to the drone yet fetches the current parameters first, so the first assignment after construction cannot send a default-constructed struct and zero resolution, framerate and codec. Deliberately not covered: that a setter transmits the whole cached struct and carries unrelated fields along with it. That behaviour is what #221 is for, so pinning it here would prejudge the discussion. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_camera.py | 66 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 66 insertions(+) diff --git a/tests/test_camera.py b/tests/test_camera.py index 5fc46735..73b640f1 100644 --- a/tests/test_camera.py +++ b/tests/test_camera.py @@ -410,3 +410,69 @@ def test_configure_rejects_mtu_size_on_blunux_5_0(mocked_drone: Drone): with camera.configure() as params: params.mtu_size = 1200 camera._parent_drone._req_rep_client.set_camera_parameters.assert_not_called() + + +def test_recording_bitrate_getter(mocked_ultra_camera): + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(recording_bitrate=24_000_000) + ) + assert mocked_ultra_camera.get_recording_bitrate() == 24_000_000 + + +def test_recording_bitrate_setter(mocked_ultra_camera): + mocked_ultra_camera._camera_parameters = bp.CameraParameters(recording_bitrate=0) + + mocked_ultra_camera.set_recording_bitrate(24_000_000) + + mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.assert_called_once_with( + bp.CameraParameters(recording_bitrate=24_000_000) + ) + assert mocked_ultra_camera._camera_parameters.recording_bitrate == 24_000_000 + + +def test_recording_bitrate_setter_requires_blunux_5(mocked_camera): + with pytest.raises(RuntimeError): + mocked_camera.set_recording_bitrate(24_000_000) + + +def test_parameter_getter_refreshes_from_the_drone(mocked_ultra_camera): + """A getter must ask the drone, never answer from the cache. + + Every getter method calls _update_camera_parameters() first, so a value + that changed since the last request - because another client set it, or + because the drone resolved an "automatic" request to a concrete number - is + reported as it is now and not as it was last written. + """ + mocked_ultra_camera._camera_parameters = bp.CameraParameters(recording_bitrate=0) + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters(recording_bitrate=14_000_000) + ) + + assert mocked_ultra_camera.get_recording_bitrate() == 14_000_000 + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.assert_called_once() + + +def test_setter_on_a_cold_camera_fetches_current_parameters_first(mocked_ultra_camera): + """A setter on a camera that has not talked to the drone yet must fetch the + current parameters before it sends. + + set_camera_parameters carries the whole CameraParameters struct, so without + the lazy fetch the first setter after construction would send a + default-constructed one and zero every field the caller never touched - + resolution, framerate and codec among them. + """ + assert mocked_ultra_camera._camera_parameters is None + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.return_value = ( + bp.CameraParameters( + stream_resolution=bp.Resolution.RESOLUTION_FULLHD_1080P, + framerate=bp.Framerate.FRAMERATE_FPS_30, + ) + ) + + mocked_ultra_camera.set_recording_bitrate(24_000_000) + + mocked_ultra_camera._parent_drone._req_rep_client.get_camera_parameters.assert_called_once() + sent = mocked_ultra_camera._parent_drone._req_rep_client.set_camera_parameters.call_args[0][0] + assert sent.recording_bitrate == 24_000_000 + assert sent.stream_resolution == bp.Resolution.RESOLUTION_FULLHD_1080P + assert sent.framerate == bp.Framerate.FRAMERATE_FPS_30 From 3e056229dceac4031c3536212ea1ffb3664247a8 Mon Sep 17 00:00:00 2001 From: Johannes Schrimpf Date: Thu, 20 Aug 2026 14:45:28 +0200 Subject: [PATCH 4/5] Guard the drone-connected camera tests by drone model Running the integration tests against an X3 Ultra on Blunux 5.1.0 left three camera tests failing for reasons that have nothing to do with the SDK: - bitrate: the Ultra applies the stream bitrate but always reports it as 0. Measured on the wire, requesting 2 Mbit/s gives 1.98 Mbit/s and 12 Mbit/s gives 11.88 Mbit/s, so only the readback is missing. The drone fills h264_bitrate from the camera control node, and the Ultra encodes H264 in the RTSP server rather than in the camera. - hue: only available on Pioneer/Pro/X1/X3, the Ultra camera has no hue control. - framerate: the Ultra only applies 25 fps to the recording pipeline, and the reported frame rate comes from the stream pipeline, which stays at 30. Add a drone_model fixture and use it to xfail the bitrate readback and skip the two unsupported parameters, so a drone-connected run is green on both families. Cover what the Ultra does support instead: 60 fps at 1080p, which the drone caps to 30 above 1440p, and sweep every resolution this branch adds across both stream_resolution and recording_resolution. Co-Authored-By: Claude Opus 5 (1M context) --- tests/conftest.py | 19 +++++++++++ tests/test_integration.py | 69 ++++++++++++++++++++++++++++++++++----- 2 files changed, 80 insertions(+), 8 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 7cbb3329..4d47c7ce 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,3 +1,5 @@ +import time + import blueye.protocol as bp import pytest @@ -13,6 +15,23 @@ def real_drone(): return blueye.sdk.Drone() +@pytest.fixture(scope="class") +def drone_model(real_drone) -> bp.Model: + """Fixture that reports the model of the connected drone + + Used to skip integration tests for camera parameters the connected hardware does not + support. Returns MODEL_UNSPECIFIED if no drone info is received, which is also what + drones older than the introduction of the model field report. + """ + deadline = time.monotonic() + 3 + while time.monotonic() < deadline: + drone_info = real_drone.telemetry.get(bp.DroneInfoTel) + if drone_info is not None: + return drone_info.drone_info.model + time.sleep(0.1) + return bp.Model.MODEL_UNSPECIFIED + + @pytest.fixture def mocked_requests(requests_mock): import json diff --git a/tests/test_integration.py b/tests/test_integration.py index e3d0dd37..3445a982 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -4,6 +4,14 @@ import pytest from packaging import version +ALL_RESOLUTIONS = [ + bp.Resolution.RESOLUTION_VGA_480P, + bp.Resolution.RESOLUTION_HD_720P, + bp.Resolution.RESOLUTION_FULLHD_1080P, + bp.Resolution.RESOLUTION_QHD_2K, + bp.Resolution.RESOLUTION_UHD_4K, +] + def polling_assert_with_timeout(getter, value_to_wait_for, timeout): """Waits for a getter to return the value we are waiting for""" @@ -48,7 +56,13 @@ def test_camera_record_time(self, real_drone): real_drone.camera.set_recording(True) polling_assert_with_timeout(real_drone.camera.get_record_time, 1, 3) - def test_camera_bitrate(self, real_drone): + def test_camera_bitrate(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.xfail( + "The Ultra applies the stream bitrate but always reports it as 0. The drone " + "fills h264_bitrate from the camera control node, and the Ultra camera does not " + "encode H264 - the RTSP server does, and its bitrate never reaches the reply" + ) _ = real_drone.camera.get_bitrate() real_drone.camera.set_bitrate(2000000) polling_assert_with_timeout(real_drone.camera.get_bitrate, 2000000, 1) @@ -69,7 +83,11 @@ def test_camera_whitebalance(self, real_drone): real_drone.camera.set_whitebalance(3400) polling_assert_with_timeout(real_drone.camera.get_whitebalance, 3400, 1) - def test_camera_hue(self, real_drone): + def test_camera_hue(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.skip( + "Hue is only available on Pioneer/Pro/X1/X3, the Ultra camera has no hue control" + ) _ = real_drone.camera.get_hue() real_drone.camera.set_hue(20) polling_assert_with_timeout(real_drone.camera.get_hue, 20, 1) @@ -89,19 +107,54 @@ def test_camera_resolution(self, real_drone): real_drone.camera.set_resolution(1080) polling_assert_with_timeout(real_drone.camera.get_resolution, 1080, 1) - def test_camera_stream_resolution(self, real_drone): + @pytest.mark.parametrize("resolution", ALL_RESOLUTIONS, ids=lambda r: r.name) + def test_camera_stream_resolution(self, real_drone, resolution): original_resolution = real_drone.camera.get_stream_resolution() try: - real_drone.camera.set_stream_resolution(bp.Resolution.RESOLUTION_HD_720P) - polling_assert_with_timeout( - real_drone.camera.get_stream_resolution, bp.Resolution.RESOLUTION_HD_720P, 3 - ) + real_drone.camera.set_stream_resolution(resolution) + polling_assert_with_timeout(real_drone.camera.get_stream_resolution, resolution, 3) finally: real_drone.camera.set_stream_resolution(original_resolution) - def test_camera_framerate(self, real_drone): + @pytest.mark.parametrize("resolution", ALL_RESOLUTIONS, ids=lambda r: r.name) + def test_camera_recording_resolution(self, real_drone, resolution): + original_resolution = real_drone.camera.get_recording_resolution() + try: + real_drone.camera.set_recording_resolution(resolution) + polling_assert_with_timeout(real_drone.camera.get_recording_resolution, resolution, 3) + finally: + real_drone.camera.set_recording_resolution(original_resolution) + + def test_camera_framerate(self, real_drone, drone_model): + if drone_model == bp.Model.MODEL_X3_ULTRA: + pytest.skip( + "The Ultra only applies 25 fps to the recording pipeline, and the reported frame " + "rate comes from the stream pipeline, which stays at 30. See " + "test_camera_framerate_60_fps for the frame rate the Ultra does support" + ) _ = real_drone.camera.get_framerate() real_drone.camera.set_framerate(25) polling_assert_with_timeout(real_drone.camera.get_framerate, 25, 1) real_drone.camera.set_framerate(30) polling_assert_with_timeout(real_drone.camera.get_framerate, 30, 1) + + def test_camera_framerate_60_fps(self, real_drone, drone_model): + """60 fps is only supported on the Ultra, and only at 1440p or lower. + + The drone caps the frame rate against the highest of the stream and recording + resolution, so both must be lowered before requesting 60 fps. + """ + if drone_model != bp.Model.MODEL_X3_ULTRA: + pytest.skip("60 fps is only supported on the Ultra") + original_stream_resolution = real_drone.camera.get_stream_resolution() + original_recording_resolution = real_drone.camera.get_recording_resolution() + original_framerate = real_drone.camera.get_framerate() + try: + real_drone.camera.set_stream_resolution(bp.Resolution.RESOLUTION_FULLHD_1080P) + real_drone.camera.set_recording_resolution(bp.Resolution.RESOLUTION_FULLHD_1080P) + real_drone.camera.set_framerate(60) + polling_assert_with_timeout(real_drone.camera.get_framerate, 60, 3) + finally: + real_drone.camera.set_framerate(original_framerate) + real_drone.camera.set_stream_resolution(original_stream_resolution) + real_drone.camera.set_recording_resolution(original_recording_resolution) From 01f66554e775e27fa680702d957e32b1caf65676 Mon Sep 17 00:00:00 2001 From: Johannes Schrimpf Date: Thu, 20 Aug 2026 16:07:54 +0200 Subject: [PATCH 5/5] Isolate the unit tests from live drone telemetry conftest has had a mocked_telemetry_client fixture since "Fix mocks for new clients", but mocked_drone never requested it, so every unit test built a real TelemetryClient. That opens a ZMQ subscriber against the drone IP, and with a drone on the same network it fills the state within a few hundred milliseconds: t=0.00s msgs_in_state= 0 ControlModeTel=False is_weather_vaning_active()=None t=0.25s msgs_in_state= 8 ControlModeTel=True is_weather_vaning_active()=False The tests asserting that a getter returns None when no telemetry has been received then fail, and because it is a race between the subscriber thread and the assertion, a different one fails on each run. CI never saw it - there is no drone there - but a full run at a desk with a drone on it failed every time. Wire the fixture into mocked_drone, and give the mock a real dict for _state plus a get() that looks up in it, so the 28 tests that seed telemetry with _telemetry_watcher._state[SomeTel] keep working. get() reads the attribute on every call rather than closing over the dict, because several tests rebind _state to a fresh dict to simulate telemetry going away. Also drops the full suite from 43s to 18s, since it no longer opens a socket and starts a subscriber thread per test. Co-Authored-By: Claude Opus 5 (1M context) --- tests/conftest.py | 20 +++++++++++++++++++- tests/test_sdk.py | 16 +++++++++++++++- 2 files changed, 34 insertions(+), 2 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 4d47c7ce..53909166 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -79,7 +79,24 @@ def mocked_ctrl_client(mocker): @pytest.fixture def mocked_telemetry_client(mocker): - return mocker.patch("blueye.sdk.drone.TelemetryClient", autospec=True) + """Patch the telemetry subscriber so unit tests never read telemetry off the network. + + Without this the drone built by `mocked_drone` opens a real ZMQ subscriber against the + drone IP, and any drone on the same network fills its state within a few hundred + milliseconds. That makes every "returns None when no telemetry has been received" test + fail at random, depending on which one happens to run late enough to receive something. + + The mock keeps a real dict for `_state` and looks messages up in it, raising KeyError for + the ones that are missing, so tests can keep seeding telemetry by assigning to + `drone._telemetry_watcher._state[SomeTel]`. + """ + patched = mocker.patch("blueye.sdk.drone.TelemetryClient", autospec=True) + instance = patched.return_value + instance._state = {} + # Read the attribute on every call rather than closing over the dict, because the real + # get() looks up self._state and several tests rebind _state to a fresh dict. + instance.get.side_effect = lambda key: instance._state[key] + return patched @pytest.fixture @@ -98,6 +115,7 @@ def mocked_drone( mocker, mocked_requests, mocked_ctrl_client, + mocked_telemetry_client, mocked_watchdog_publisher, mocked_req_rep_client, ): diff --git a/tests/test_sdk.py b/tests/test_sdk.py index 0ad9ebf8..5cb2f1ba 100644 --- a/tests/test_sdk.py +++ b/tests/test_sdk.py @@ -1,6 +1,6 @@ import json from time import time -from unittest.mock import Mock, PropertyMock +from unittest.mock import Mock, NonCallableMock, PropertyMock import blueye.protocol as bp import pytest @@ -33,6 +33,20 @@ def test_angle_conversion(self, mocked_drone, old_angle, new_angle): assert pose["yaw"] == new_angle +def test_mocked_drone_is_isolated_from_live_telemetry(mocked_drone): + """Unit tests must never read telemetry off the network. + + `mocked_drone` used to build a real TelemetryClient, which subscribes to the drone IP. With + a drone on the same network that fills the state within a few hundred milliseconds, and the + tests asserting that a getter returns None when no telemetry has been received start failing + at random. + """ + # A spec'd mock passes isinstance() against the class it specs, so check for the mock + # itself rather than asserting the watcher is not a TelemetryClient. + assert isinstance(mocked_drone._telemetry_watcher, NonCallableMock) + assert mocked_drone.telemetry.get(bp.ControlModeTel) is None + + def test_zmq_connection_error(mocked_drone): mocked_drone._req_rep_client.ping.side_effect = bp.exceptions.ResponseTimeout with pytest.raises(ConnectionError):