Merge pull request #5360 from jenshnielsen/jenshnielsen/cleanup_visa_…

…inst_tests

Instrument tests: Avoid leaking state between tests

qcodes/instrument/ip_to_visa.py

@@ -90,6 +90,23 @@ def close(self) -> None:
         if getattr(self, 'visa_handle', None):
             self.visa_handle.close()
 
+        if getattr(self, "visabackend", None) == "sim" and getattr(
+            self, "resource_manager", None
+        ):
+            # The pyvisa-sim visalib has a session attribute but the resource manager is not generic in the
+            # visalib type so we cannot get it in a type safe way
+            known_sessions = getattr(self.resource_manager.visalib, "sessions", ())
+            session_found = self.resource_manager.session in known_sessions
+
+            n_sessions = len(known_sessions)
+            # if this instrument is the last one or there are no connected instruments its safe to reset the device
+            if (session_found and n_sessions == 1) or n_sessions == 0:
+                # work around for https://github.com/pyvisa/pyvisa-sim/issues/83
+                # see other issues for more context
+                # https://github.com/QCoDeS/Qcodes/issues/5356 and
+                # https://github.com/pyvisa/pyvisa-sim/issues/82
+                self.resource_manager.visalib._init()
+
         # Instrument close
         if hasattr(self, 'connection') and hasattr(self.connection, 'close'):
             self.connection.close()

qcodes/instrument/visa.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+import warnings
 from collections.abc import Sequence
 from importlib.resources import as_file, files
 from typing import Any
@@ -257,6 +258,30 @@ def close(self) -> None:
         if getattr(self, 'visa_handle', None):
             self.visa_handle.close()
 
+        if getattr(self, "visabackend", None) == "sim" and getattr(
+            self, "resource_manager", None
+        ):
+            # The pyvisa-sim visalib has a session attribute but the resource manager is not generic in the
+            # visalib type so we cannot get it in a type safe way
+            known_sessions = getattr(self.resource_manager.visalib, "sessions", ())
+            session_found = self.resource_manager.session in known_sessions
+
+            n_sessions = len(known_sessions)
+            # if this instrument is the last one or there are no connected instruments its safe to reset the device
+            if (session_found and n_sessions == 1) or n_sessions == 0:
+                # work around for https://github.com/pyvisa/pyvisa-sim/issues/83
+                # see other issues for more context
+                # https://github.com/QCoDeS/Qcodes/issues/5356 and
+                # https://github.com/pyvisa/pyvisa-sim/issues/82
+                try:
+                    self.resource_manager.visalib._init()
+                except AttributeError:
+                    warnings.warn(
+                        "The installed version of pyvisa-sim does not have an `_init` method "
+                        "in its visa library implementation. Cannot reset simulated instrument state. "
+                        "On reconnect the instrument may retain settings set in this session."
+                    )
+
         super().close()
 
     def write_raw(self, cmd: str) -> None:

qcodes/instrument_drivers/Keithley/Keithley_s46.py

@@ -139,37 +139,42 @@ class KeithleyS46(VisaInstrument):
     def __init__(self, name: str, address: str, **kwargs: Any):
 
         super().__init__(name, address, terminator="\n", **kwargs)
+        try:
+            self.add_parameter(
+                "closed_channels",
+                get_cmd=":CLOS?",
+                get_parser=self._get_closed_channels_parser,
+            )
 
-        self.add_parameter(
-            "closed_channels",
-            get_cmd=":CLOS?",
-            get_parser=self._get_closed_channels_parser,
-        )
-
-        self._available_channels: list[str] = []
-
-        for relay_name, channel_count in zip(
-            KeithleyS46.relay_names, self.relay_layout
-        ):
-
-            relay_lock = KeithleyS46RelayLock(relay_name)
-
-            for channel_index in range(1, channel_count + 1):
-                # E.g. For channel 'B2', channel_index is 2
-                if channel_count > 1:
-                    alias = f"{relay_name}{channel_index}"
-                else:
-                    alias = relay_name  # For channels R1 to R8, we have one
-                    # channel per relay. Channel alias = relay name
-
-                self.add_parameter(
-                    alias,
-                    channel_number=KeithleyS46.channel_numbers[alias],
-                    lock=relay_lock,
-                    parameter_class=S46Parameter,
-                )
-
-                self._available_channels.append(alias)
+            self._available_channels: list[str] = []
+
+            for relay_name, channel_count in zip(
+                KeithleyS46.relay_names, self.relay_layout
+            ):
+
+                relay_lock = KeithleyS46RelayLock(relay_name)
+
+                for channel_index in range(1, channel_count + 1):
+                    # E.g. For channel 'B2', channel_index is 2
+                    if channel_count > 1:
+                        alias = f"{relay_name}{channel_index}"
+                    else:
+                        alias = relay_name  # For channels R1 to R8, we have one
+                        # channel per relay. Channel alias = relay name
+
+                    self.add_parameter(
+                        alias,
+                        channel_number=KeithleyS46.channel_numbers[alias],
+                        lock=relay_lock,
+                        parameter_class=S46Parameter,
+                    )
+
+                    self._available_channels.append(alias)
+        except RuntimeError as err:
+            # If we error on undesirable state we want to make sure
+            # we also close the visa connection
+            self.close()
+            raise err
 
     @staticmethod
     def _get_closed_channels_parser(reply: str) -> list[str]:

qcodes/tests/drivers/test_ami430_visa.py

@@ -162,6 +162,91 @@ def test_instantiation_compat_classes(request: FixtureRequest) -> None:
     assert driver._instrument_z is mag_z
 
 
+def test_visa_interaction(request: FixtureRequest) -> None:
+    """
+    Test that closing one instrument we can still use the other simulated instruments.
+    """
+    request.addfinalizer(AMIModel4303D.close_all)
+    mag_x = AMIModel430(
+        "x", address="GPIB::1::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_y = AMIModel430(
+        "y", address="GPIB::2::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_z = AMIModel430(
+        "z", address="GPIB::3::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+
+    default_field_value = 0.123
+
+    assert mag_x.field() == default_field_value
+    assert mag_y.field() == default_field_value
+    assert mag_z.field() == default_field_value
+
+    mag_x.field(0.1)
+    mag_y.field(0.2)
+    mag_z.field(0.3)
+
+    assert mag_x.field() == 0.1
+    assert mag_y.field() == 0.2
+    assert mag_z.field() == 0.3
+
+    mag_x.close()
+    # closing x should not change y or z
+    assert mag_y.field() == 0.2
+    assert mag_z.field() == 0.3
+
+
+def test_sim_visa_reset_on_fully_closed(request: FixtureRequest) -> None:
+    """
+    Test that closing all instruments defined in a yaml file will reset the
+    state of all the instruments.
+    """
+    request.addfinalizer(AMIModel4303D.close_all)
+    mag_x = AMIModel430(
+        "x", address="GPIB::1::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_y = AMIModel430(
+        "y", address="GPIB::2::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_z = AMIModel430(
+        "z", address="GPIB::3::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+
+    default_field_value = 0.123
+
+    assert mag_x.field() == default_field_value
+    assert mag_y.field() == default_field_value
+    assert mag_z.field() == default_field_value
+
+    mag_x.field(0.1)
+    mag_y.field(0.2)
+    mag_z.field(0.3)
+
+    assert mag_x.field() == 0.1
+    assert mag_y.field() == 0.2
+    assert mag_z.field() == 0.3
+
+    mag_x.close()
+    mag_y.close()
+    mag_z.close()
+
+    # all are closed so instruments should be reset
+    mag_x = AMIModel430(
+        "x", address="GPIB::1::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_y = AMIModel430(
+        "y", address="GPIB::2::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+    mag_z = AMIModel430(
+        "z", address="GPIB::3::INSTR", pyvisa_sim_file="AMI430.yaml", terminator="\n"
+    )
+
+    assert mag_x.field() == default_field_value
+    assert mag_y.field() == default_field_value
+    assert mag_z.field() == default_field_value
+
+
 def test_instantiation_from_name_of_nonexistent_ami_instrument(
     magnet_axes_instances, request: FixtureRequest
 ) -> None:

qcodes/tests/drivers/test_keithley_s46.py

@@ -23,7 +23,7 @@ def calc_channel_nr(alias: str) -> int:
     ])
 
 
-@pytest.fixture(scope='module')
+@pytest.fixture(scope="function")
 def s46_six():
     """
     A six channel-per-relay instrument
@@ -38,7 +38,7 @@ def s46_six():
         driver.close()
 
 
-@pytest.fixture(scope='module')
+@pytest.fixture(scope="function")
 def s46_four():
     """
     A four channel-per-relay instrument

qcodes/tests/drivers/test_keysight_34980a.py

@@ -2,12 +2,12 @@
 import logging
 
 import pytest
-from pytest import LogCaptureFixture
+from pytest import FixtureRequest, LogCaptureFixture
 
 from qcodes.instrument_drivers.Keysight.keysight_34980a import Keysight34980A
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture(scope="function")
 def switch_driver():
     inst = Keysight34980A(
         "keysight_34980A_sim",
@@ -21,20 +21,25 @@ def switch_driver():
         inst.close()
 
 
-def test_safety_interlock_during_init(switch_driver, caplog: LogCaptureFixture) -> None:
+def test_safety_interlock_during_init(
+    request: FixtureRequest, caplog: LogCaptureFixture
+) -> None:
     """
     to check if a warning would show when initialize the instrument with a
-    module in safety interlock state. This test has to be placed first if
-    the scope is set to be "module".
+    module in safety interlock state.
     """
-    msg = [
-        x.message for x in caplog.get_records('setup')
-        if x.levelno == logging.WARNING
-    ]
-    assert "safety interlock" in msg[0]
+    with caplog.at_level(logging.WARNING):
+        inst = Keysight34980A(
+            "keysight_34980A_sim",
+            address="GPIB::1::INSTR",
+            pyvisa_sim_file="keysight_34980A.yaml",
+        )
+    request.addfinalizer(inst.close)
 
+    assert "safety interlock" in caplog.records[0].msg
 
-def test_get_idn(switch_driver) -> None:
+
+def test_get_idn(switch_driver: Keysight34980A) -> None:
     """
     to check if the instrument attributes are set correctly after getting
     the IDN
@@ -47,7 +52,7 @@ def test_get_idn(switch_driver) -> None:
     }
 
 
-def test_scan_slots(switch_driver) -> None:
+def test_scan_slots(switch_driver: Keysight34980A) -> None:
     """
     to check if the submodule attributes are set correctly after scanning
     every slot
@@ -69,11 +74,15 @@ def test_scan_slots(switch_driver) -> None:
     }
 
 
-def test_safety_interlock(switch_driver, caplog: LogCaptureFixture) -> None:
+def test_safety_interlock(
+    switch_driver: Keysight34980A, caplog: LogCaptureFixture
+) -> None:
     """
     to check if a warning would show when talk to a module that is in safety
     interlock state
     """
-    switch_driver.module[3].write('*CLS')
+    module = switch_driver.module[3]
+    assert module is not None
+    module.write("*CLS")
     with caplog.at_level(logging.DEBUG):
         assert "safety interlock" in caplog.text