Fix tests.

controllerData/ioDefinition.py

@@ -12,7 +12,10 @@
 import logging
 from typing import Any
 
-from . import sensors
+try:
+    from . import sensors  # type: ignore
+except ImportError:
+    from . import sensors_sample as sensors
 
 log = logging.getLogger("TemperatureController")
 

controllerData/listener.py

@@ -17,10 +17,10 @@
 
 try:  # Qt for nice effects.
     from qtpy import QtCore
-    from qtpy.QtCore import Slot as pyqtSlot, Signal as pyqtSignal
+    from qtpy.QtCore import Signal as pyqtSignal
 except ModuleNotFoundError:
     from PyQt5 import QtCore
-    from PyQt5.QtCore import pyqtSlot, pyqtSignal
+    from PyQt5.QtCore import pyqtSignal
 
 from devices import intercom
 
@@ -158,7 +158,7 @@ def setValue(self, content):
             self.signals.pidChanged.emit(key.replace("pid", ""))
         intercom.sendMessage(self.connection, 'ACK')
 
-    def getValue(self, content):
+    def getValue(self, content: bytes) -> None:
         """Get some value."""
         keys = pickle.loads(content)
         assert hasattr(keys, '__iter__'), "The content has to be an iterable."

devices/intercom.py

@@ -23,7 +23,7 @@
 import logging
 import pickle
 import socket
-from socket import timeout as timeout
+from socket import timeout as timeout  # noqa: F401
 
 import zmq
 
@@ -66,19 +66,20 @@
 )
 
 
-def connect(address="127.0.0.1", port=12345, timeout=None):
+def connect(address="127.0.0.1", port=12345, timeout=None) -> socket.socket:  # noqa: F811
     """Create and return a TCP connection to `address` and `port` with `timeout`."""
     connection = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     connection.settimeout(timeout)
     connection.connect((address, port))
     return connection
 
 
-def readMessage(connection):
+def readMessage(connection: socket.socket) -> tuple[str, bytes]:
     """Receive and decode a message. Return `typ` and `content`."""
     headerLength = 3 + 5  # 3 letters for type, 5 for length
     status = 0
     readBuffer = b''
+    typ = ""
     while status < 2:
         read = connection.recv(1064)
         if not read:
@@ -92,9 +93,10 @@ def readMessage(connection):
         if status == 1 and len(readBuffer) >= length:  # Reading message.
             content = readBuffer[:length]
             return typ, content
+    raise TimeoutError
 
 
-def sendMessage(connection, typ, content=b''):
+def sendMessage(connection: socket.socket, typ: str, content: bytes = b'') -> None:
     """Encode and send a message with `typ` and `content`."""
     assert typ in validCommands, "Unknown type"
     header = f"{typ}{len(content):05}".encode('utf-8')
@@ -104,7 +106,7 @@ def sendMessage(connection, typ, content=b''):
 class Intercom:
     """An intercom channel using one-time connections."""
 
-    def __init__(self, address="127.0.0.1", port=12345, timeout=10):
+    def __init__(self, address="127.0.0.1", port=12345, timeout=10):  # noqa: F811
         """Save connection settings."""
         self.address = address, port, timeout
 

tests/ioDefinition_test.py

@@ -19,20 +19,20 @@
 
 # file to test
 from controllerData import ioDefinition
+from controllerData.ioDefinition import sensors  # type: ignore
 
 
 ioDefinition.log.addHandler(logging.StreamHandler())
 
 
 class Empty():
-    pass
+    def __init__(self):
+        self.readoutMethods = []
 
 
 @pytest.fixture
 def empty():
-    raw = Empty()
-    raw.readoutMethods = []
-    return raw
+    return Empty()
 
 
 @pytest.fixture
@@ -94,7 +94,9 @@ def mock_tinkerforge(monkeypatch):
     def returner(*args):
         return args
 
-    monkeypatch.setattr(ioDefinition, 'devices', {297: returner, 111: returner, 2115: returner}, False)
+    monkeypatch.setattr(
+        ioDefinition, "devices", {297: returner, 111: returner, 2115: returner}, False
+    )
     monkeypatch.setattr(ioDefinition, "IPConnection", Mock_IPConnection, False)
     monkeypatch.setattr(ioDefinition, 'BrickHAT', Mock_BrickHAT, False)
     monkeypatch.setattr(ioDefinition, 'BrickletAirQuality', Mock_BrickletAirQuality, False)
@@ -104,7 +106,8 @@ def returner(*args):
 @pytest.fixture
 def tf_device_pars():
     """Tinkerforge device parameters."""
-    # 0: uid, 1: connected_uid, 2: position, 3: hardware_version, 4: firmware_version, 5: device_identifier, 6: enumeration_type
+    # 0: uid, 1: connected_uid, 2: position, 3: hardware_version, 4: firmware_version,
+    # 5: device_identifier, 6: enumeration_type
     return ["abc", "def", "i", 0, 0, 111, 1]
 
 
@@ -221,21 +224,21 @@ def test_stop_watchdog(self, skeleton):
         ioDefinition.InputOutput.close(skeleton)
         assert skeleton.tfDevices['abc'].args == (0, 0, False, False, False)
 
-    def test_close_sensors(self, empty, monkeypatch, calling):
-        monkeypatch.setattr('controllerData.sensors.close', calling)
+    def test_close_sensors(self, empty, monkeypatch: pytest.MonkeyPatch, calling):
+        monkeypatch.setattr(sensors, 'close', calling)
         ioDefinition.InputOutput.close(empty)
         assert called
 
-    def test_close_sensors_failed(self, skeletonP, monkeypatch, raising):
-        monkeypatch.setattr('controllerData.sensors.close', raising)
+    def test_close_sensors_failed(self, skeletonP, monkeypatch: pytest.MonkeyPatch, raising):
+        monkeypatch.setattr(sensors, 'close', raising)
         ioDefinition.InputOutput.close(skeletonP)
         pass  # TODO test exception log
 
 
 class Test_getSensors:
     @pytest.fixture
-    def sensorsGetData(self, monkeypatch):
-        monkeypatch.setattr('controllerData.sensors.getData', lambda *args: {})
+    def sensorsGetData(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(sensors, 'getData', lambda *args: {})
 
     def test_getSensors_Clean(self, empty, sensorsGetData):
         assert ioDefinition.InputOutput.getSensors(empty) == {}
@@ -247,15 +250,15 @@ def test_renew_watchdog(self, skeleton, sensorsGetData):
         assert skeleton.tfDevices['abc'].args == (30, 1, True, False, True)
 
     def test_call_sensors(self, empty, monkeypatch):
-        monkeypatch.setattr('controllerData.sensors.getData', lambda *args: {'test': True})
+        monkeypatch.setattr(sensors, 'getData', lambda *args: {'test': True})
         assert ioDefinition.InputOutput.getSensors(empty) == {'test': True}
 
     def test_call_sensors_no_dictionary(self, empty, monkeypatch):
-        monkeypatch.setattr('controllerData.sensors.getData', lambda *args: [])
+        monkeypatch.setattr(sensors, 'getData', lambda *args: [])
         assert ioDefinition.InputOutput.getSensors(empty) == {}
 
     def test_call_sensors_failed(self, skeletonP, monkeypatch, raising, caplog):
-        monkeypatch.setattr('controllerData.sensors.getData', raising)
+        monkeypatch.setattr(sensors, 'getData', raising)
         ioDefinition.InputOutput.getSensors(skeletonP)
         assert "Get sensors failed." in caplog.text
 
@@ -281,7 +284,7 @@ def test_tf(self, skeleton):
         assert skeleton.tfDevices['ao3'].voltage == 9
 
     def test_external(self, skeleton, monkeypatch, calling):
-        monkeypatch.setattr('controllerData.sensors.setOutput', calling)
+        monkeypatch.setattr(sensors, 'setOutput', calling)
         ioDefinition.InputOutput.setOutput(skeleton, 'out5', 2)
         assert calledArgs == (skeleton, 'out5', 2)
 

tests/listener_test.py

@@ -11,12 +11,7 @@
 
 # auxiliary for fixtures and tests
 import pickle
-try:  # Qt for nice effects.
-    from PyQt6 import QtCore
-    pyqt = 6
-except ModuleNotFoundError:
-    from PyQt5 import QtCore
-    pyqt = 5
+from qtpy import QtCore
 import socket
 
 from devices import intercom
@@ -82,10 +77,7 @@ def passing():
 @pytest.fixture
 def sets(monkeypatch):
     settings = QtCore.QSettings('NLOQO', "tests")
-    if pyqt == 5:
-        monkeypatch.setattr('PyQt5.QtCore.QSettings', lambda: settings)
-    if pyqt == 6:
-        monkeypatch.setattr('PyQt6.QtCore.QSettings', lambda: settings)
+    monkeypatch.setattr('qtpy.QtCore.QSettings', lambda: settings)
     yield settings
     settings.clear()
 
@@ -208,7 +200,7 @@ def test_getValue_wrong(self, ch):
         with pytest.raises(AssertionError):
             ch.getValue(pickle.dumps(5))
 
-    def test_some_value(self, ch, sets):
+    def test_some_value(self, ch: listener.ConnectionHandler, sets):
         sets.setValue('testing', 5)
         ch.getValue(pickle.dumps(["testing"]))
         assert pickle.loads(message[2]) == {'testing': 5}