From 8d962329ac353418309e8f7e0bcb2665f186bfb5 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:16:49 -0500 Subject: [PATCH 1/7] Fix ERSTE argument order and TOCTOU race in EM power supply Fix set_operational_error_enable_mask to place the operational mask in the second position of the ERSTE command (ERSTE ,), matching the protocol format. Previously it was incorrectly placed in the first (hardware) position. Also fix the docstring parameter type from EMPowerSupplyHardwareErrorsRegister to EMPowerSupplyOperationalErrorsRegister. Add comments to both set_hardware_error_enable_mask and set_operational_error_enable_mask noting the non-atomic read-modify-write TOCTOU limitation. Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 10 ++++++++-- tests/test_em_power_supply.py | 2 +- 2 files changed, 9 insertions(+), 3 deletions(-) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index 56e167f..1b4a9f3 100644 --- a/lakeshore/em_power_supply.py +++ b/lakeshore/em_power_supply.py @@ -662,6 +662,9 @@ def set_hardware_error_enable_mask(self, register_mask): register_mask (ElectromagnetPowerSupply.EMPowerSupplyHardwareErrorsRegister): Register mask configuration object. """ + # Note: read-modify-write is not atomic. The lock is held per-call, so another thread could + # modify the register between the query and command. This is acceptable for typical use since + # concurrent register modification is unusual. operational_mask = self.query("ERSTE?").split(',')[1] self.command(f"ERSTE {register_mask.to_integer()},{operational_mask}") @@ -710,11 +713,14 @@ def set_operational_error_enable_mask(self, register_mask): Status Byte Register. Args: - register_mask (ElectromagnetPowerSupply.EMPowerSupplyHardwareErrorsRegister): Register mask configuration + register_mask (ElectromagnetPowerSupply.EMPowerSupplyOperationalErrorsRegister): Register mask configuration object. """ + # Note: read-modify-write is not atomic. The lock is held per-call, so another thread could + # modify the register between the query and command. This is acceptable for typical use since + # concurrent register modification is unusual. hardware_mask = self.query("ERSTE?").split(',')[0] - self.command(f"ERSTE {register_mask.to_integer()},{hardware_mask}") + self.command(f"ERSTE {hardware_mask},{register_mask.to_integer()}") def get_operational_error_enable_mask(self): """Returns which operational error bits will set the summary bit in the Status Byte Register. diff --git a/tests/test_em_power_supply.py b/tests/test_em_power_supply.py index 5509e3b..8b55d72 100644 --- a/tests/test_em_power_supply.py +++ b/tests/test_em_power_supply.py @@ -283,7 +283,7 @@ def test_set_operational_error_enable_mask(self): operational_error_enable_mask = self.dut.EMPowerSupplyOperationalErrorsRegister.from_integer(4) self.dut.set_operational_error_enable_mask(operational_error_enable_mask) self.fake_connection.get_outgoing_message() - self.assertIn("ERSTE 4", self.fake_connection.get_outgoing_message()) + self.assertIn("ERSTE 1,4", self.fake_connection.get_outgoing_message()) def test_get_operational_error_enable_mask(self): self.fake_connection.setup_response("0,4; 0") From 93433532892ba4628f03a9cb3996acb6711d44da Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:17:39 -0500 Subject: [PATCH 2/7] Add bounds validation to EM power supply safety-critical methods Add input validation to set_current, set_ramp_rate, set_limits, and set_ramp_segment to prevent dangerous out-of-range values from being sent to the electromagnet power supply (up to 135A output). Validates type checking and range bounds per the instrument specifications. Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 45 ++++++++++++++++++++++++++++++++++-- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index 1b4a9f3..b90e674 100644 --- a/lakeshore/em_power_supply.py +++ b/lakeshore/em_power_supply.py @@ -251,7 +251,20 @@ def set_limits(self, max_current, max_ramp_rate): A. The Model 648 bounds are 0.0000 - 135.1000 A. max_ramp_rate (float): The maximum output current ramp rate setting allowed (0.0001 - 50.000 A/s). - """ + + Raises: + ValueError: If max_current or max_ramp_rate is not a number or out of range. + """ + if not isinstance(max_current, (int, float)): + raise ValueError(f"Max current must be a number, got {type(max_current).__name__}") + if not isinstance(max_ramp_rate, (int, float)): + raise ValueError(f"Max ramp rate must be a number, got {type(max_ramp_rate).__name__}") + if max_current < 0 or max_current > 135.1: + raise ValueError( + f"Max current {max_current} A is outside valid range (0 - 135.1 A)") + if max_ramp_rate < 0.0001 or max_ramp_rate > 50.0: + raise ValueError( + f"Max ramp rate {max_ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") self.command(f"LIMIT {max_current}, {max_ramp_rate}") def get_limits(self): @@ -273,7 +286,15 @@ def set_ramp_rate(self, ramp_rate): Args: ramp_rate (float): The rate at which the current will ramp when a new output current setting is entered (0.0001 - 50.000 A/s). + + Raises: + ValueError: If ramp_rate is not a number or out of range. """ + if not isinstance(ramp_rate, (int, float)): + raise ValueError(f"Ramp rate must be a number, got {type(ramp_rate).__name__}") + if ramp_rate < 0.0001 or ramp_rate > 50.0: + raise ValueError( + f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") self.command(f"RATE {ramp_rate}") def get_ramp_rate(self): @@ -294,7 +315,18 @@ def set_ramp_segment(self, segment, current, ramp_rate): current (float): Specifies the upper output current setting that will use this segment. ramp_rate (float): Specifies the rate at which the current will ramp. (0.0001 - 50.000 A/s). - """ + Raises: + ValueError: If any parameter is invalid or out of range. + """ + if segment not in range(1, 6): + raise ValueError(f"Segment must be 1-5, got {segment}") + if not isinstance(current, (int, float)): + raise ValueError(f"Current must be a number, got {type(current).__name__}") + if not isinstance(ramp_rate, (int, float)): + raise ValueError(f"Ramp rate must be a number, got {type(ramp_rate).__name__}") + if ramp_rate < 0.0001 or ramp_rate > 50.0: + raise ValueError( + f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") self.command(f"RSEGS {segment}, {current}, {ramp_rate}") def get_ramp_segment(self, segment): @@ -336,7 +368,16 @@ def set_current(self, current): Args: current (float): The output current value that the output will ramp to at the present ramp rate. The Model 643 bounds are 0.0000 - +/-70.1000 A. The Model 648 bounds are 0.0000 - +/-135.1000 A. + + Raises: + ValueError: If current is not a number or exceeds safe bounds. """ + if not isinstance(current, (int, float)): + raise ValueError(f"Current must be a number, got {type(current).__name__}") + max_current = 135.1 # Model 648 maximum; Model 643 is 70.1 A + if abs(current) > max_current: + raise ValueError( + f"Current magnitude {abs(current)} A exceeds maximum safe value of {max_current} A") self.command(f"SETI {current}") def get_current(self): From 12a91a46c81f50d33fc4fb239d49c578d5964926 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:23:33 -0500 Subject: [PATCH 3/7] Harden EM power supply validation against NaN/Inf and add missing bounds Add _validate_number helper that rejects bool, non-numeric types, NaN, and Inf values. Previously float('nan') and float('inf') passed all validation checks due to NaN comparison semantics and would send invalid commands to the instrument. Also add missing current magnitude bounds check to set_ramp_segment, and fix pre-existing docstring bug in get_operational_error_enable_mask (return type was incorrectly documented as EMPowerSupplyHardwareErrorsRegister). Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 56 +++++++++++++++++++++++++----------- 1 file changed, 39 insertions(+), 17 deletions(-) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index b90e674..0e10a27 100644 --- a/lakeshore/em_power_supply.py +++ b/lakeshore/em_power_supply.py @@ -1,9 +1,33 @@ """Implements functionality unique to the Model 643 and 648 electromagnet power supplies.""" +import math + import serial from .generic_instrument import GenericInstrument, RegisterBase, _parse_response, InstrumentException +def _validate_number(value, name): + """Validate that a value is a finite number (rejects bool, NaN, and inf). + + Args: + value: The value to validate. + name (str): Parameter name for error messages. + + Returns: + The validated value (unchanged). + + Raises: + ValueError: If the value is not a finite number. + """ + if isinstance(value, bool): + raise ValueError(f"{name} must be a number, got bool") + if not isinstance(value, (int, float)): + raise ValueError(f"{name} must be a number, got {type(value).__name__}") + if not math.isfinite(value): + raise ValueError(f"{name} must be finite, got {value}") + return value + + class ElectromagnetPowerSupply(GenericInstrument): """Class object representing a Lake Shore Model 643 or 648 electromagnet power supply.""" vid_pid = [(0x1FB9, 0x0601), (0x1FB9, 0x0602)] # 643, 648 @@ -253,12 +277,10 @@ def set_limits(self, max_current, max_ramp_rate): max_ramp_rate (float): The maximum output current ramp rate setting allowed (0.0001 - 50.000 A/s). Raises: - ValueError: If max_current or max_ramp_rate is not a number or out of range. + ValueError: If max_current or max_ramp_rate is not a finite number or out of range. """ - if not isinstance(max_current, (int, float)): - raise ValueError(f"Max current must be a number, got {type(max_current).__name__}") - if not isinstance(max_ramp_rate, (int, float)): - raise ValueError(f"Max ramp rate must be a number, got {type(max_ramp_rate).__name__}") + _validate_number(max_current, "Max current") + _validate_number(max_ramp_rate, "Max ramp rate") if max_current < 0 or max_current > 135.1: raise ValueError( f"Max current {max_current} A is outside valid range (0 - 135.1 A)") @@ -288,10 +310,9 @@ def set_ramp_rate(self, ramp_rate): (0.0001 - 50.000 A/s). Raises: - ValueError: If ramp_rate is not a number or out of range. + ValueError: If ramp_rate is not a finite number or out of range. """ - if not isinstance(ramp_rate, (int, float)): - raise ValueError(f"Ramp rate must be a number, got {type(ramp_rate).__name__}") + _validate_number(ramp_rate, "Ramp rate") if ramp_rate < 0.0001 or ramp_rate > 50.0: raise ValueError( f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") @@ -316,14 +337,16 @@ def set_ramp_segment(self, segment, current, ramp_rate): ramp_rate (float): Specifies the rate at which the current will ramp. (0.0001 - 50.000 A/s). Raises: - ValueError: If any parameter is invalid or out of range. + ValueError: If any parameter is not a finite number or out of range. """ if segment not in range(1, 6): raise ValueError(f"Segment must be 1-5, got {segment}") - if not isinstance(current, (int, float)): - raise ValueError(f"Current must be a number, got {type(current).__name__}") - if not isinstance(ramp_rate, (int, float)): - raise ValueError(f"Ramp rate must be a number, got {type(ramp_rate).__name__}") + _validate_number(current, "Current") + _validate_number(ramp_rate, "Ramp rate") + max_current = 135.1 # Model 648 maximum; Model 643 is 70.1 A + if abs(current) > max_current: + raise ValueError( + f"Current magnitude {abs(current)} A exceeds maximum safe value of {max_current} A") if ramp_rate < 0.0001 or ramp_rate > 50.0: raise ValueError( f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") @@ -370,10 +393,9 @@ def set_current(self, current): 643 bounds are 0.0000 - +/-70.1000 A. The Model 648 bounds are 0.0000 - +/-135.1000 A. Raises: - ValueError: If current is not a number or exceeds safe bounds. + ValueError: If current is not a finite number or exceeds safe bounds. """ - if not isinstance(current, (int, float)): - raise ValueError(f"Current must be a number, got {type(current).__name__}") + _validate_number(current, "Current") max_current = 135.1 # Model 648 maximum; Model 643 is 70.1 A if abs(current) > max_current: raise ValueError( @@ -771,7 +793,7 @@ def get_operational_error_enable_mask(self): Status Byte Register. Returns: - ElectromagnetPowerSupply.EMPowerSupplyHardwareErrorsRegister: Register mask configuration object. + ElectromagnetPowerSupply.EMPowerSupplyOperationalErrorsRegister: Register mask configuration object. """ operational_mask = int(self.query("ERSTE?").split(',')[1]) return self.EMPowerSupplyOperationalErrorsRegister.from_integer(operational_mask) From b186f0a9b88f383461ae716bcce1b3fb88938360 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:24:19 -0500 Subject: [PATCH 4/7] Add validation to remaining safety-critical EM power supply methods Add input validation to set_internal_water, set_magnet_water, set_display_brightness, set_front_panel_lock, and set_programming_mode. These methods control physical systems (especially cooling water for a 135A electromagnet) where invalid values could cause equipment damage. Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index 0e10a27..81fedfa 100644 --- a/lakeshore/em_power_supply.py +++ b/lakeshore/em_power_supply.py @@ -439,7 +439,12 @@ def set_internal_water(self, mode): Args: mode (int): Internal water mode (0, 1, 2, or 3). 0 = Manual-Off, 1 = Manual-On, 2 = Auto, 3 = Disabled. + + Raises: + ValueError: If mode is not 0, 1, 2, or 3. """ + if mode not in (0, 1, 2, 3): + raise ValueError(f"Internal water mode must be 0, 1, 2, or 3, got {mode}") self.command(f"INTWTR {mode}") def get_internal_water(self) : @@ -456,7 +461,12 @@ def set_magnet_water(self, mode): Args: mode (int): Magnet water mode. (0, 1, 2, or 3). 0 = Manual-Off, 1 = Manual-On, 2 = Auto, 3 = Disabled. + + Raises: + ValueError: If mode is not 0, 1, 2, or 3. """ + if mode not in (0, 1, 2, 3): + raise ValueError(f"Magnet water mode must be 0, 1, 2, or 3, got {mode}") self.command(f"MAGWTR {mode}") def get_magnet_water(self) : @@ -472,7 +482,12 @@ def set_display_brightness(self, brightness_level): Args: brightness_level (int): The display brightness. 0=25%, 1=50%, 2=75%, 3=100%. + + Raises: + ValueError: If brightness_level is not 0, 1, 2, or 3. """ + if brightness_level not in (0, 1, 2, 3): + raise ValueError(f"Brightness level must be 0, 1, 2, or 3, got {brightness_level}") self.command(f"DISP {brightness_level}") def get_display_brightness(self) : @@ -489,7 +504,12 @@ def set_front_panel_lock(self, lock_state, code): Args: lock_state (int): The lock state to be set (0, 1, or 2). 0=unlock, 1=lock, and 2=lock limits. code (int): Keypad lock code required to make changes to the lock state of the front panel. + + Raises: + ValueError: If lock_state is not 0, 1, or 2. """ + if lock_state not in (0, 1, 2): + raise ValueError(f"Lock state must be 0, 1, or 2, got {lock_state}") self.command(f"LOCK {lock_state},{code}") def get_front_panel_status(self): @@ -513,7 +533,12 @@ def set_programming_mode(self, mode): Args: mode (int): Programming mode (0, 1, or 2). 0=Internal, 1=External, 2=Sum. + + Raises: + ValueError: If mode is not 0, 1, or 2. """ + if mode not in (0, 1, 2): + raise ValueError(f"Programming mode must be 0, 1, or 2, got {mode}") self.command(f"XPGM {mode}") def get_programming_mode(self): From 2399fa73d3e41c0dbc43cc3f79953528726edf33 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:25:21 -0500 Subject: [PATCH 5/7] Add comprehensive validation tests and strengthen ERSTE mask test Add 43 new test cases covering all validation logic: - Type rejection (string, None, bool, NaN, Inf) for set_current, set_ramp_rate, set_limits, and set_ramp_segment - Range boundary tests (both acceptance and rejection) - Discrete mode validation for set_internal_water, set_magnet_water, set_display_brightness, set_front_panel_lock, set_programming_mode Strengthen test_set_hardware_error_enable_mask to use distinguishable non-zero values (hw=4, op=7) so a swapped-argument regression would be caught by the assertion. Co-Authored-By: Claude Opus 4.6 --- tests/test_em_power_supply.py | 201 +++++++++++++++++++++++++++++++++- 1 file changed, 199 insertions(+), 2 deletions(-) diff --git a/tests/test_em_power_supply.py b/tests/test_em_power_supply.py index 8b55d72..eede1d2 100644 --- a/tests/test_em_power_supply.py +++ b/tests/test_em_power_supply.py @@ -1,3 +1,4 @@ +import math from tests.utils import TestWithFakeEMPowerSupply @@ -252,12 +253,12 @@ def test_get_operation_event(self): self.assertIn("OPST?", self.fake_connection.get_outgoing_message()) def test_set_hardware_error_enable_mask(self): - self.fake_connection.setup_response("0,0; 0") + self.fake_connection.setup_response("0,7; 0") self.fake_connection.setup_response("0") hardware_error_enable_mask = self.dut.EMPowerSupplyHardwareErrorsRegister.from_integer(4) self.dut.set_hardware_error_enable_mask(hardware_error_enable_mask) self.fake_connection.get_outgoing_message() - self.assertIn("ERSTE 4", self.fake_connection.get_outgoing_message()) + self.assertIn("ERSTE 4,7", self.fake_connection.get_outgoing_message()) def test_get_hardware_error_enable_mask(self): self.fake_connection.setup_response("4,0; 0") @@ -302,3 +303,199 @@ def test_get_operational_error_event(self): response = self.dut.get_operational_error_event() self.assertEqual(response.to_integer(), 1) self.assertIn("ERSTR?", self.fake_connection.get_outgoing_message()) + + +class TestCurrentValidation(TestWithFakeEMPowerSupply): + + def test_set_current_rejects_string(self): + with self.assertRaises(ValueError): + self.dut.set_current("fifty") + + def test_set_current_rejects_none(self): + with self.assertRaises(ValueError): + self.dut.set_current(None) + + def test_set_current_rejects_bool(self): + with self.assertRaises(ValueError): + self.dut.set_current(True) + + def test_set_current_rejects_nan(self): + with self.assertRaises(ValueError): + self.dut.set_current(float('nan')) + + def test_set_current_rejects_inf(self): + with self.assertRaises(ValueError): + self.dut.set_current(float('inf')) + + def test_set_current_rejects_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_current(135.2) + + def test_set_current_rejects_below_negative_max(self): + with self.assertRaises(ValueError): + self.dut.set_current(-135.2) + + def test_set_current_accepts_boundary_positive(self): + self.fake_connection.setup_response("0") + self.dut.set_current(135.1) + self.assertIn("SETI 135.1", self.fake_connection.get_outgoing_message()) + + def test_set_current_accepts_boundary_negative(self): + self.fake_connection.setup_response("0") + self.dut.set_current(-135.1) + self.assertIn("SETI -135.1", self.fake_connection.get_outgoing_message()) + + def test_set_current_accepts_zero(self): + self.fake_connection.setup_response("0") + self.dut.set_current(0) + self.assertIn("SETI 0", self.fake_connection.get_outgoing_message()) + + +class TestRampRateValidation(TestWithFakeEMPowerSupply): + + def test_set_ramp_rate_rejects_string(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate("fast") + + def test_set_ramp_rate_rejects_nan(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(float('nan')) + + def test_set_ramp_rate_rejects_zero(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(0) + + def test_set_ramp_rate_rejects_negative(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(-1.0) + + def test_set_ramp_rate_rejects_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(50.1) + + def test_set_ramp_rate_accepts_lower_boundary(self): + self.fake_connection.setup_response("0") + self.dut.set_ramp_rate(0.0001) + self.assertIn("RATE 0.0001", self.fake_connection.get_outgoing_message()) + + def test_set_ramp_rate_accepts_upper_boundary(self): + self.fake_connection.setup_response("0") + self.dut.set_ramp_rate(50.0) + self.assertIn("RATE 50.0", self.fake_connection.get_outgoing_message()) + + +class TestLimitsValidation(TestWithFakeEMPowerSupply): + + def test_set_limits_rejects_string_current(self): + with self.assertRaises(ValueError): + self.dut.set_limits("high", 3.0) + + def test_set_limits_rejects_string_ramp_rate(self): + with self.assertRaises(ValueError): + self.dut.set_limits(50.0, "fast") + + def test_set_limits_rejects_negative_current(self): + with self.assertRaises(ValueError): + self.dut.set_limits(-1.0, 3.0) + + def test_set_limits_rejects_current_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_limits(135.2, 3.0) + + def test_set_limits_rejects_ramp_rate_below_min(self): + with self.assertRaises(ValueError): + self.dut.set_limits(50.0, 0.0) + + def test_set_limits_rejects_ramp_rate_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_limits(50.0, 50.1) + + def test_set_limits_rejects_nan_current(self): + with self.assertRaises(ValueError): + self.dut.set_limits(float('nan'), 3.0) + + def test_set_limits_rejects_inf_ramp_rate(self): + with self.assertRaises(ValueError): + self.dut.set_limits(50.0, float('inf')) + + def test_set_limits_accepts_boundary_values(self): + self.fake_connection.setup_response("0") + self.dut.set_limits(135.1, 50.0) + self.assertIn("LIMIT 135.1, 50.0", self.fake_connection.get_outgoing_message()) + + def test_set_limits_accepts_lower_boundary_values(self): + self.fake_connection.setup_response("0") + self.dut.set_limits(0, 0.0001) + self.assertIn("LIMIT 0, 0.0001", self.fake_connection.get_outgoing_message()) + + +class TestRampSegmentValidation(TestWithFakeEMPowerSupply): + + def test_set_ramp_segment_rejects_segment_zero(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(0, 50, 2.5) + + def test_set_ramp_segment_rejects_segment_six(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(6, 50, 2.5) + + def test_set_ramp_segment_rejects_string_current(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, "fifty", 2.5) + + def test_set_ramp_segment_rejects_string_ramp_rate(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, 50, "fast") + + def test_set_ramp_segment_rejects_ramp_rate_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, 50, 50.1) + + def test_set_ramp_segment_rejects_current_above_max(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, 200, 2.5) + + def test_set_ramp_segment_rejects_nan_current(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, float('nan'), 2.5) + + def test_set_ramp_segment_accepts_segment_one(self): + self.fake_connection.setup_response("0") + self.dut.set_ramp_segment(1, 50, 2.5) + self.assertIn("RSEGS 1, 50, 2.5", self.fake_connection.get_outgoing_message()) + + def test_set_ramp_segment_accepts_segment_five(self): + self.fake_connection.setup_response("0") + self.dut.set_ramp_segment(5, 50, 2.5) + self.assertIn("RSEGS 5, 50, 2.5", self.fake_connection.get_outgoing_message()) + + +class TestSafetyMethodValidation(TestWithFakeEMPowerSupply): + + def test_set_internal_water_rejects_invalid_mode(self): + with self.assertRaises(ValueError): + self.dut.set_internal_water(4) + + def test_set_internal_water_rejects_negative(self): + with self.assertRaises(ValueError): + self.dut.set_internal_water(-1) + + def test_set_magnet_water_rejects_invalid_mode(self): + with self.assertRaises(ValueError): + self.dut.set_magnet_water(5) + + def test_set_magnet_water_rejects_string(self): + with self.assertRaises(ValueError): + self.dut.set_magnet_water("auto") + + def test_set_display_brightness_rejects_invalid(self): + with self.assertRaises(ValueError): + self.dut.set_display_brightness(4) + + def test_set_front_panel_lock_rejects_invalid_state(self): + with self.assertRaises(ValueError): + self.dut.set_front_panel_lock(3, 123) + + def test_set_programming_mode_rejects_invalid(self): + with self.assertRaises(ValueError): + self.dut.set_programming_mode(3) From 1453ae581358af41b2b8ef39a845cfaf15609e82 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:35:07 -0500 Subject: [PATCH 6/7] Fix remaining validation gaps and extract magic numbers to constants Extract duplicated bounds (135.1 A, 0.0001-50.0 A/s) into class constants _MAX_CURRENT, _MIN_RAMP_RATE, _MAX_RAMP_RATE for maintainability. Add missing validation to set_ieee_488 (terminator, eoi_enable, address), set_ieee_interface_mode (mode 0-2), set_ramp_segments_enable (bool/0/1), and the code parameter in set_front_panel_lock. Fix _validate_number: remove over-indented docstring sections and unused return value. Fix set_ramp_segment Raises docstring to accurately describe the mixed validation (discrete for segment, numeric for current/ramp_rate). Co-Authored-By: Claude Opus 4.6 --- lakeshore/em_power_supply.py | 74 +++++++++++++++++++++++------------- 1 file changed, 48 insertions(+), 26 deletions(-) diff --git a/lakeshore/em_power_supply.py b/lakeshore/em_power_supply.py index 81fedfa..1fe2d78 100644 --- a/lakeshore/em_power_supply.py +++ b/lakeshore/em_power_supply.py @@ -9,15 +9,12 @@ def _validate_number(value, name): """Validate that a value is a finite number (rejects bool, NaN, and inf). - Args: - value: The value to validate. - name (str): Parameter name for error messages. - - Returns: - The validated value (unchanged). + Args: + value: The value to validate. + name (str): Parameter name for error messages. - Raises: - ValueError: If the value is not a finite number. + Raises: + ValueError: If the value is not a finite number. """ if isinstance(value, bool): raise ValueError(f"{name} must be a number, got bool") @@ -25,13 +22,16 @@ def _validate_number(value, name): raise ValueError(f"{name} must be a number, got {type(value).__name__}") if not math.isfinite(value): raise ValueError(f"{name} must be finite, got {value}") - return value class ElectromagnetPowerSupply(GenericInstrument): """Class object representing a Lake Shore Model 643 or 648 electromagnet power supply.""" vid_pid = [(0x1FB9, 0x0601), (0x1FB9, 0x0602)] # 643, 648 + _MAX_CURRENT = 135.1 # Model 648 maximum; Model 643 is 70.1 A + _MIN_RAMP_RATE = 0.0001 # A/s + _MAX_RAMP_RATE = 50.0 # A/s + def __init__(self, serial_number=None, com_port=None, @@ -281,12 +281,13 @@ def set_limits(self, max_current, max_ramp_rate): """ _validate_number(max_current, "Max current") _validate_number(max_ramp_rate, "Max ramp rate") - if max_current < 0 or max_current > 135.1: + if max_current < 0 or max_current > self._MAX_CURRENT: raise ValueError( - f"Max current {max_current} A is outside valid range (0 - 135.1 A)") - if max_ramp_rate < 0.0001 or max_ramp_rate > 50.0: + f"Max current {max_current} A is outside valid range (0 - {self._MAX_CURRENT} A)") + if max_ramp_rate < self._MIN_RAMP_RATE or max_ramp_rate > self._MAX_RAMP_RATE: raise ValueError( - f"Max ramp rate {max_ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") + f"Max ramp rate {max_ramp_rate} A/s is outside valid range " + f"({self._MIN_RAMP_RATE} - {self._MAX_RAMP_RATE} A/s)") self.command(f"LIMIT {max_current}, {max_ramp_rate}") def get_limits(self): @@ -313,9 +314,10 @@ def set_ramp_rate(self, ramp_rate): ValueError: If ramp_rate is not a finite number or out of range. """ _validate_number(ramp_rate, "Ramp rate") - if ramp_rate < 0.0001 or ramp_rate > 50.0: + if ramp_rate < self._MIN_RAMP_RATE or ramp_rate > self._MAX_RAMP_RATE: raise ValueError( - f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") + f"Ramp rate {ramp_rate} A/s is outside valid range " + f"({self._MIN_RAMP_RATE} - {self._MAX_RAMP_RATE} A/s)") self.command(f"RATE {ramp_rate}") def get_ramp_rate(self): @@ -337,19 +339,19 @@ def set_ramp_segment(self, segment, current, ramp_rate): ramp_rate (float): Specifies the rate at which the current will ramp. (0.0001 - 50.000 A/s). Raises: - ValueError: If any parameter is not a finite number or out of range. + ValueError: If segment is not 1-5, or current/ramp_rate is not a finite number or out of range. """ if segment not in range(1, 6): raise ValueError(f"Segment must be 1-5, got {segment}") _validate_number(current, "Current") _validate_number(ramp_rate, "Ramp rate") - max_current = 135.1 # Model 648 maximum; Model 643 is 70.1 A - if abs(current) > max_current: + if abs(current) > self._MAX_CURRENT: raise ValueError( - f"Current magnitude {abs(current)} A exceeds maximum safe value of {max_current} A") - if ramp_rate < 0.0001 or ramp_rate > 50.0: + f"Current magnitude {abs(current)} A exceeds maximum safe value of {self._MAX_CURRENT} A") + if ramp_rate < self._MIN_RAMP_RATE or ramp_rate > self._MAX_RAMP_RATE: raise ValueError( - f"Ramp rate {ramp_rate} A/s is outside valid range (0.0001 - 50.0 A/s)") + f"Ramp rate {ramp_rate} A/s is outside valid range " + f"({self._MIN_RAMP_RATE} - {self._MAX_RAMP_RATE} A/s)") self.command(f"RSEGS {segment}, {current}, {ramp_rate}") def get_ramp_segment(self, segment): @@ -370,7 +372,12 @@ def set_ramp_segments_enable(self, state): Args: state (bool): The state of the ramp segments enable. 0=Disabled and 1=Enabled. + + Raises: + ValueError: If state is not a bool or 0/1. """ + if state not in (True, False, 0, 1): + raise ValueError(f"Ramp segments enable must be True/False or 0/1, got {state}") self.command(f"RSEG {int(state)}") def get_ramp_segments_enable(self): @@ -396,10 +403,9 @@ def set_current(self, current): ValueError: If current is not a finite number or exceeds safe bounds. """ _validate_number(current, "Current") - max_current = 135.1 # Model 648 maximum; Model 643 is 70.1 A - if abs(current) > max_current: + if abs(current) > self._MAX_CURRENT: raise ValueError( - f"Current magnitude {abs(current)} A exceeds maximum safe value of {max_current} A") + f"Current magnitude {abs(current)} A exceeds maximum safe value of {self._MAX_CURRENT} A") self.command(f"SETI {current}") def get_current(self): @@ -506,10 +512,12 @@ def set_front_panel_lock(self, lock_state, code): code (int): Keypad lock code required to make changes to the lock state of the front panel. Raises: - ValueError: If lock_state is not 0, 1, or 2. + ValueError: If lock_state is not 0, 1, or 2, or code is not an integer. """ if lock_state not in (0, 1, 2): raise ValueError(f"Lock state must be 0, 1, or 2, got {lock_state}") + if not isinstance(code, int) or isinstance(code, bool): + raise ValueError(f"Lock code must be an integer, got {type(code).__name__}") self.command(f"LOCK {lock_state},{code}") def get_front_panel_status(self): @@ -556,7 +564,16 @@ def set_ieee_488(self, terminator, eoi_enable, address): terminator(int): the terminator. 0=, 1=, 2=, 3=no terminator (must have EOI enabled). eoi_enable(int): Sets EOI (End of Interrupt) mode. 0=Enabled, 1=Disabled. address (int): Specifies IEEE address. 1 - 30(0 and 31 are reserved). - """ + + Raises: + ValueError: If any parameter is out of its valid range. + """ + if terminator not in (0, 1, 2, 3): + raise ValueError(f"Terminator must be 0, 1, 2, or 3, got {terminator}") + if eoi_enable not in (0, 1): + raise ValueError(f"EOI enable must be 0 or 1, got {eoi_enable}") + if not isinstance(address, int) or isinstance(address, bool) or address < 1 or address > 30: + raise ValueError(f"IEEE address must be an integer from 1-30, got {address}") self.command(f"IEEE {terminator},{eoi_enable},{address}") def get_iee_488(self): @@ -576,7 +593,12 @@ def set_ieee_interface_mode(self, mode): Args: mode (int): Interface mode. 0, 1 or 2. 0=local, 1=remote, and 2=remote with local lockout. + + Raises: + ValueError: If mode is not 0, 1, or 2. """ + if mode not in (0, 1, 2): + raise ValueError(f"Interface mode must be 0, 1, or 2, got {mode}") self.command(f"MODE {mode}") def get_ieee_interface_mode(self): From 8c02eb7d23203f329d01a44f7158837d31c45c03 Mon Sep 17 00:00:00 2001 From: Justin Fichtner Date: Tue, 17 Feb 2026 19:37:34 -0500 Subject: [PATCH 7/7] Complete test coverage for EM power supply validation Add missing validation tests: bool/None/inf rejection for ramp_rate, bool rejection for limits, negative segment and ramp_rate below-min for ramp_segment. Add tests for newly validated methods: ieee_488, ieee_interface_mode, ramp_segments_enable, and front_panel_lock code parameter. Remove unused math import. Total: 108 tests. Co-Authored-By: Claude Opus 4.6 --- tests/test_em_power_supply.py | 61 ++++++++++++++++++++++++++++++++++- 1 file changed, 60 insertions(+), 1 deletion(-) diff --git a/tests/test_em_power_supply.py b/tests/test_em_power_supply.py index eede1d2..08d8cd8 100644 --- a/tests/test_em_power_supply.py +++ b/tests/test_em_power_supply.py @@ -1,4 +1,3 @@ -import math from tests.utils import TestWithFakeEMPowerSupply @@ -357,10 +356,22 @@ def test_set_ramp_rate_rejects_string(self): with self.assertRaises(ValueError): self.dut.set_ramp_rate("fast") + def test_set_ramp_rate_rejects_none(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(None) + + def test_set_ramp_rate_rejects_bool(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(True) + def test_set_ramp_rate_rejects_nan(self): with self.assertRaises(ValueError): self.dut.set_ramp_rate(float('nan')) + def test_set_ramp_rate_rejects_inf(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_rate(float('inf')) + def test_set_ramp_rate_rejects_zero(self): with self.assertRaises(ValueError): self.dut.set_ramp_rate(0) @@ -410,6 +421,10 @@ def test_set_limits_rejects_ramp_rate_above_max(self): with self.assertRaises(ValueError): self.dut.set_limits(50.0, 50.1) + def test_set_limits_rejects_bool_current(self): + with self.assertRaises(ValueError): + self.dut.set_limits(True, 3.0) + def test_set_limits_rejects_nan_current(self): with self.assertRaises(ValueError): self.dut.set_limits(float('nan'), 3.0) @@ -435,6 +450,10 @@ def test_set_ramp_segment_rejects_segment_zero(self): with self.assertRaises(ValueError): self.dut.set_ramp_segment(0, 50, 2.5) + def test_set_ramp_segment_rejects_segment_negative(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(-1, 50, 2.5) + def test_set_ramp_segment_rejects_segment_six(self): with self.assertRaises(ValueError): self.dut.set_ramp_segment(6, 50, 2.5) @@ -451,6 +470,10 @@ def test_set_ramp_segment_rejects_ramp_rate_above_max(self): with self.assertRaises(ValueError): self.dut.set_ramp_segment(1, 50, 50.1) + def test_set_ramp_segment_rejects_ramp_rate_below_min(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segment(1, 50, 0) + def test_set_ramp_segment_rejects_current_above_max(self): with self.assertRaises(ValueError): self.dut.set_ramp_segment(1, 200, 2.5) @@ -499,3 +522,39 @@ def test_set_front_panel_lock_rejects_invalid_state(self): def test_set_programming_mode_rejects_invalid(self): with self.assertRaises(ValueError): self.dut.set_programming_mode(3) + + def test_set_front_panel_lock_rejects_invalid_code_type(self): + with self.assertRaises(ValueError): + self.dut.set_front_panel_lock(1, "abc") + + def test_set_front_panel_lock_rejects_bool_code(self): + with self.assertRaises(ValueError): + self.dut.set_front_panel_lock(1, True) + + def test_set_ramp_segments_enable_rejects_invalid(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segments_enable(2) + + def test_set_ramp_segments_enable_rejects_string(self): + with self.assertRaises(ValueError): + self.dut.set_ramp_segments_enable("on") + + def test_set_ieee_488_rejects_invalid_terminator(self): + with self.assertRaises(ValueError): + self.dut.set_ieee_488(4, 0, 8) + + def test_set_ieee_488_rejects_invalid_eoi(self): + with self.assertRaises(ValueError): + self.dut.set_ieee_488(0, 2, 8) + + def test_set_ieee_488_rejects_address_zero(self): + with self.assertRaises(ValueError): + self.dut.set_ieee_488(0, 0, 0) + + def test_set_ieee_488_rejects_address_31(self): + with self.assertRaises(ValueError): + self.dut.set_ieee_488(0, 0, 31) + + def test_set_ieee_interface_mode_rejects_invalid(self): + with self.assertRaises(ValueError): + self.dut.set_ieee_interface_mode(3)