Fix tests and ci

.github/workflows/_dist.yml

@@ -15,6 +15,11 @@ jobs:
       - name: Install uv
         uses: astral-sh/setup-uv@v7
 
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+
       - name: Build sdist and wheel
         run: >
           export SOURCE_DATE_EPOCH=$(git log -1 --pretty=%ct) &&

src/dls_pmac_control/__main__.py

@@ -48,14 +48,12 @@ class Controlform(QMainWindow, UiControlForm):
     disable_polling_status_signal = pyqtSignal(bool)
 
     def __init__(self, options, parent=None):
-        super().__init__()
+        super().__init__(parent)
 
         signal.signal(2, self.signal_handler)
         # setup signals
 
-        QMainWindow.__init__(self, parent)
         self.setup_ui(self)
-        # self.parent = parent
 
         self.greenLedOn = QPixmap(path.join(path.dirname(__file__), "greenLedOn.png"))
         self.greenLedOff = QPixmap(path.join(path.dirname(__file__), "greenLedOff.png"))
@@ -498,21 +496,21 @@ def pmac_energise_axis(self):
         self.energiseScreen = Energiseform(self.pmac, self)
         self.energiseScreen.show()
 
-    def status_screen(self):
+    def show_status_screen(self):
         # if power pmac
         if isinstance(self.pmac, PPmacSshInterface):
             self.ppmacstatusScreen.show()
         else:
             self.status_screen.show()
 
-    def cs_status_screen(self):
+    def show_cs_status_screen(self):
         # if power pmac
         if isinstance(self.pmac, PPmacSshInterface):
             self.PpmacCSStatusScreen.show()
         else:
             self.cs_status_screen.show()
 
-    def global_status_screen(self):
+    def show_global_status_screen(self):
         # if power pmac
         if isinstance(self.pmac, PPmacSshInterface):
             self.PpmacGlobalStatusScreen.show()

src/dls_pmac_control/comms_thread.py

@@ -2,6 +2,7 @@
 import time
 import traceback
 from queue import Queue
+from typing import Any, cast
 
 from dls_pmaclib.dls_pmacremote import (
     PmacEthernetInterface,
@@ -66,8 +67,10 @@ def start(self):
 
     @pyqtSlot()
     def stop(self):
-        if self.timer:
+        if self.timer is not None:
             self.timer.stop()
+            self.timer.deleteLater()
+            self.timer = None
         self.finished.emit()
 
     @pyqtSlot(list)
@@ -220,14 +223,14 @@ def generate_cmd(self):
         return cmd
 
     def poll_status(self) -> None:
-        if self.parent.pmac is None:
+        if cast(Any, self.parent).pmac is None:
             return None
 
-        if not self.parent.pmac.isConnectionOpen:
+        if not cast(Any, self.parent).pmac.isConnectionOpen:
             return None
 
         cmd = self.generate_cmd()
-        (send_command_response, success) = self.parent.pmac.sendCommand(cmd)
+        (send_command_response, success) = cast(Any, self.parent).pmac.sendCommand(cmd)
 
         if success:
             parsed_poll_response_status = self.parsed_poll_response(
@@ -242,7 +245,6 @@ def poll_status(self) -> None:
 
     def update_func(self):
         if self.parent.pmac is None or not self.parent.pmac.isConnectionOpen:
-            time.sleep(0.1)
             return
 
         self.poll_status()

src/dls_pmac_control/ui_form_control.py

@@ -643,7 +643,7 @@ def setup_ui(self, control_form):
         self.btnJogStop.clicked.connect(control_form.jog_stop)
         self.btnJogTo.clicked.connect(control_form.jog_go_to_position)
         self.btnSettings.clicked.connect(control_form.jog_parameters)
-        self.btnStatus.clicked.connect(control_form.status_screen)
+        self.btnStatus.clicked.connect(control_form.show_status_screen)
         self.btnJogNeg.clicked.connect(control_form.jog_neg)
         self.table.cellDoubleClicked["int", "int"].connect(
             control_form.chooseMotorFromTable
@@ -654,8 +654,8 @@ def setup_ui(self, control_form):
         self.btnLoadFile.clicked.connect(control_form.pmac_load_config)
         self.btnKillAll.clicked.connect(control_form.kill_all_motors)
         self.btnEnergise.clicked.connect(control_form.pmac_energise_axis)
-        self.btnCSStatus.clicked.connect(control_form.cs_status_screen)
-        self.btnGlobalStatus.clicked.connect(control_form.global_status_screen)
+        self.btnCSStatus.clicked.connect(control_form.show_cs_status_screen)
+        self.btnGlobalStatus.clicked.connect(control_form.show_global_status_screen)
         self.btnHome.clicked.connect(control_form.jog_home)
         self.rbUseSerial.clicked.connect(control_form.use_serial)
         QtCore.QMetaObject.connectSlotsByName(control_form)

tests/conftest.py

@@ -2,6 +2,7 @@
 from typing import Any
 
 import pytest
+from PyQt6.QtWidgets import QApplication
 
 # Prevent pytest from catching exceptions when debugging in vscode so that break on
 # exception works correctly (see: https://github.com/pytest-dev/pytest/issues/7409)
@@ -19,3 +20,11 @@ def pytest_exception_interact(call: pytest.CallInfo[Any]):
     @pytest.hookimpl(tryfirst=True)
     def pytest_internalerror(excinfo: pytest.ExceptionInfo[Any]):
         raise excinfo.value
+
+
+@pytest.fixture(scope="session", autouse=True)
+def qapp():
+    app = QApplication.instance()
+    if app is None:
+        app = QApplication([])
+    return app

tests/test_commsthread.py

@@ -1,9 +1,10 @@
 import unittest
 from unittest.mock import Mock, patch
 
+from PyQt6.QtTest import QTest
 from PyQt6.QtWidgets import QMainWindow
 
-from dls_pmac_control.comms_thread import CommsThread
+from dls_pmac_control.comms_thread import CommsWorker
 
 
 class DummyTestWidget(QMainWindow):
@@ -25,18 +26,16 @@ def __init__(self, parent=None):
 
 class CommsthreadTest(unittest.TestCase):
     @patch("threading.Lock")
-    @patch("threading.Thread")
     @patch("queue.Queue")
-    def setUp(self, mock_queue, mock_thread, mock_lock):
+    def setUp(self, mock_queue, mock_lock):
         self.test_widget = DummyTestWidget()
-        self.obj = CommsThread(self.test_widget)
+        self.obj = CommsWorker(self.test_widget)
 
     def test_init(self):
         assert self.obj.parent == self.test_widget
         assert self.obj.CSNum == 1
         assert self.obj.gen is None
-        assert self.obj.updateReadyEvent is None
-        assert self.obj.disablePollingStatus is False
+        # assert self.obj.disablePollingStatus is False
         assert self.obj.max_pollrate is None
         assert self.obj.lineNumber == 0
         assert self.obj._watch_window == {}
@@ -74,61 +73,77 @@ def test_send_complete(self, mock_custom, mock_event):
         assert mock_custom.called
         assert mock_event.called
 
-    @patch("dls_pmac_control.comms_thread.CommsThread.update_func")
-    def test_update_thread(self, mock_updatefunc):
-        mock_updatefunc.return_value = True
-        self.obj.update_thread()
+    @patch("dls_pmac_control.comms_thread.CommsWorker.update_func")
+    def test_update_func_called(self, mock_updatefunc):
+        self.obj.start()
+        QTest.qWait(150)
         assert mock_updatefunc.called
 
 
 class UpdatefuncTest(unittest.TestCase):
-    @patch("threading.Lock")
-    @patch("threading.Thread")
-    @patch("queue.Queue")
-    def setUp(self, mock_queue, mock_thread, mock_lock):
+    def setUp(self):
         self.test_widget = DummyTestWidget()
-        self.obj = CommsThread(self.test_widget)
+        self.obj = CommsWorker(self.test_widget)
 
-    @patch("queue.Queue.get")
-    def test_update_func_die(self, mock_get):
-        mock_get.return_value = ("die", "data")
-        assert self.obj.update_func() is True
-        mock_get.assert_called_with(block=False)
+    def test_update_func_sendseries(self):
+        data = ["item1", "item2", "item3"]
 
-    @patch("PyQt6.QtCore.QCoreApplication.postEvent")
-    @patch("dls_pmac_control.comms_thread.CustomEvent")
-    @patch("queue.Queue.put")
-    @patch("queue.Queue.get")
-    def test_update_func_sendseries(self, mock_get, mock_put, mock_custom, mock_event):
-        mock_get.return_value = ("sendSeries", "data")
-        self.obj.parent.pmac.isConnectionOpen = True
-        self.obj.disablePollingStatus = None
-        assert self.obj.update_func() is None
-        self.test_widget.pmac.sendSeries.assert_called_with("data")
-        mock_get.assert_called_with(block=False)
-
-    @patch("queue.Queue.get")
-    def test_update_func_disable(self, mock_get):
-        mock_get.return_value = ("disablePollingStatus", True)
-        self.obj.gen = False
-        self.obj.parent.pmac.isConnectionOpen = True
-        ret = self.obj.update_func()
-        assert self.obj.disablePollingStatus is True
-        assert ret is None
-        mock_get.assert_called_with(block=False)
+        self.obj.send_series(data)
 
-    @patch("PyQt6.QtCore.QCoreApplication.postEvent")
-    @patch("dls_pmac_control.comms_thread.CustomEvent")
-    @patch("queue.Queue.put")
-    @patch("queue.Queue.get")
-    def test_update_func_cancel(self, mock_get, mock_put, mock_custom, mock_event):
-        mock_get.return_value = ("cancelSendSeries", "data")
-        self.obj.parent.pmac.isConnectionOpen = True
-        self.obj.disablePollingStatus = False
-        self.obj.gen = False
-        ret = self.obj.update_func()
-        assert ret is None
-        assert mock_put.call_count == 5
-        assert mock_custom.called
-        assert mock_event.called
-        mock_get.assert_called_with(block=False)
+        self.test_widget.pmac.sendSeries.assert_called_once_with(data)
+
+        self.assertIs(self.obj.gen, self.test_widget.pmac.sendSeries.return_value)
+
+    def test_update_func_sendseries_exception(self):
+        data = ["item1", "item2", "item3"]
+
+        self.test_widget.pmac.sendSeries.side_effect = Exception("Error")
+
+        with patch("dls_pmac_control.comms_thread.traceback.print_exc") as mock_print:
+            with patch.object(self.obj, "send_complete") as mock_complete:
+                self.obj.send_series(data)
+
+        mock_complete.assert_called_once_with("Couldn't start download")
+        mock_print.assert_called_once()
+
+    def test_update_func_disable_polling_status(self):
+        self.assertFalse(self.obj.disablePollingStatusValue)
+
+        self.obj.disable_polling_status(True)
+        self.assertTrue(self.obj.disablePollingStatusValue)
+
+        self.obj.disable_polling_status(False)
+        self.assertFalse(self.obj.disablePollingStatusValue)
+
+    def test_update_func_cancel(self):
+        self.obj.gen = Mock()
+
+        with patch.object(self.obj, "send_complete") as mock_complete:
+            self.obj.cancel_send_series()
+
+        self.obj.gen.close.assert_called_once_with()
+        mock_complete.assert_called_once_with("Download cancelled by the user")
+
+    def test_update_func_cancel_no_mock(self):
+        self.obj.gen = None
+
+        with patch.object(self.obj, "send_complete") as mock_complete:
+            self.obj.cancel_send_series()
+
+        mock_complete.assert_not_called()
+
+    def test_update_func_poll_status(self):
+        self.test_widget.pmac.isConnectionOpen = True
+
+        with patch.object(self.obj, "poll_status") as mock_poll:
+            self.obj.update_func()
+
+        mock_poll.assert_called_once_with()
+
+    def test_update_func_poll_status_exited(self):
+        self.test_widget.pmac.isConnectionOpen = False
+
+        with patch.object(self.obj, "poll_status") as mock_poll:
+            self.obj.update_func()
+
+        mock_poll.assert_not_called()

tests/test_csstatus.py

@@ -18,8 +18,8 @@ def __init__(self, parent=None):
         self.redLedOn = QPixmap(path.join(path.dirname(__file__), "redLedOn.png"))
         self.redLedOff = QPixmap(path.join(path.dirname(__file__), "redLedOff.png"))
         self.pmac = Mock()
-        self.commsThread = Mock()
-        self.commsThread.CSNum = 1
+        self.comms_worker = Mock()
+        self.comms_worker.CSNum = 1
 
 
 class CSStatusTest(unittest.TestCase):

tests/test_login.py

@@ -1,20 +1,17 @@
-import sys
 import unittest
 from unittest.mock import patch
 
 from PyQt6.QtCore import Qt
 from PyQt6.QtTest import QTest
-from PyQt6.QtWidgets import QApplication, QWidget
+from PyQt6.QtWidgets import QWidget
 
 from dls_pmac_control.login import Loginform
 
-app = QApplication(sys.argv)
-test_widget = QWidget()
-
 
 class LoginTest(unittest.TestCase):
     def setUp(self):
-        self.obj = Loginform(test_widget, "", "")
+        self.test_widget = QWidget()
+        self.obj = Loginform(self.test_widget, "", "")
 
     def test_inital_form(self):
         self.assertEqual(self.obj.lneUsername.text(), "")