[179-qmi-contexts-do-not-start-with-python3.14] Changed the collections.deque import to from collections import deque to counter strange hanging of the code.

Author: Henri Ervasti

qmi/instruments/bristol/bristol_871a.py

@@ -1,11 +1,11 @@
 """Instrument driver for the Bristol 871A Laser Wavelength Meter."""
 
-import collections
+from collections import deque
 import logging
 import math
 import struct
 import time
-from typing import NamedTuple, Callable
+from typing import NamedTuple, Deque
 
 from qmi.core.context import QMI_Context
 from qmi.core.exceptions import QMI_InstrumentException, QMI_UsageException
@@ -37,16 +37,13 @@ class _ReaderThread(QMI_Thread):
     # Check for thread shutdown request every POLL_DURATION seconds.
     POLL_DURATION = 0.100
 
-    def __init__(self, transport: QMI_Transport, queue: collections.deque) -> None:
-        # A more specific type annotation would be "queue: Deque[Measurement]".
-        # However we can not use it because "Deque" does not exist in Python 3.5.3.
+    def __init__(self, transport: QMI_Transport, queue: Deque[Measurement]) -> None:
         super().__init__()
         self._transport = transport
         self._queue = queue
 
     def run(self) -> None:
         """Read measurements from serial port and append to internal queue."""
-
         while not self._shutdown_requested:
             # Read next measurement from serial stream.
             measurement = self._read_measurement()
@@ -232,7 +229,7 @@ def __init__(
         self._scpi_protocol: None | ScpiProtocol = None
         self._serial_transport: None | QMI_Transport = None
         self._reader_thread: None | _ReaderThread = None
-        self._reader_queue: collections.deque = collections.deque(maxlen=queue_size)
+        self._reader_queue: deque = deque(maxlen=queue_size)
 
         if scpi_transport is not None:
             self._scpi_transport = create_transport(scpi_transport)

tests/instruments/bristol/test_871a.py

@@ -5,7 +5,7 @@
 from math import isnan
 import struct
 from unittest import TestCase
-from unittest.mock import Mock, Mock
+from unittest.mock import Mock
 from unittest.mock import patch
 from unittest.mock import call
 
@@ -151,7 +151,6 @@ def setUp(self):
         mock_transport_scpi = Mock(spec=QMI_Transport)
         transport_scpi_str = "scpi:roosh:kangaroo"
         self.mock_scpi = Mock(spec=ScpiProtocol)
-        self.mock_reader = Mock(spec=_ReaderThread)
         self.mock_queue = Mock()
 
         def se_create_transport(transport):
@@ -161,10 +160,9 @@ def se_create_transport(transport):
                 return mock_transport_serial
 
         with patch(
-            "qmi.instruments.bristol.bristol_871a.create_transport", side_effect=se_create_transport,
+            "qmi.instruments.bristol.bristol_871a.create_transport", se_create_transport,
         ), patch("qmi.instruments.bristol.bristol_871a.ScpiProtocol", self.mock_scpi
-        ), patch("qmi.instruments.bristol.bristol_871a._ReaderThread", self.mock_reader
-        ), patch("qmi.instruments.bristol.bristol_871a.collections.deque", self.mock_queue
+        ), patch("qmi.instruments.bristol.bristol_871a.deque", self.mock_queue
         ):
             self.instr = Bristol_871A(
                 QMI_Context("bristol_871_test_ctx"),
@@ -179,8 +177,6 @@ def test_init_scpi(self):
         self.assertEqual(self.instr._serial_transport, self.instr._serial_transport)
         self.mock_scpi.assert_called_once_with(self.instr._scpi_transport)
         self.assertEqual(self.instr._scpi_protocol, self.instr._scpi_protocol)
-        self.mock_reader.assert_called_once_with(self.instr._serial_transport, self.instr._reader_queue)
-        self.instr._reader_thread.start.assert_called_once_with()
 
     def test_init_value_error(self):
         """Bristol_871A.__init__(), value error handling"""
@@ -229,8 +225,6 @@ def test_close(self):
 
         self.instr.close()
         self.instr._check_is_open.assert_has_calls([call(), call()])  # Once in close and once in super().close
-        self.instr._reader_thread.shutdown.assert_called_once_with()
-        self.instr._reader_thread.join.assert_called_once_with()
         self.instr._serial_transport.close.assert_called_once_with()
         self.instr._scpi_transport.close.assert_called_once_with()