diff --git a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py index bbeb87d5e3..0e33a71249 100644 --- a/cura/PrinterOutput/NetworkedPrinterOutputDevice.py +++ b/cura/PrinterOutput/NetworkedPrinterOutputDevice.py @@ -310,11 +310,11 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): def _createNetworkManager(self) -> None: Logger.log("d", "Creating network manager") if self._manager: - self._manager.finished.disconnect(self.__handleOnFinished) + self._manager.finished.disconnect(self._handleOnFinished) self._manager.authenticationRequired.disconnect(self._onAuthenticationRequired) self._manager = QNetworkAccessManager() - self._manager.finished.connect(self.__handleOnFinished) + self._manager.finished.connect(self._handleOnFinished) self._last_manager_create_time = time() self._manager.authenticationRequired.connect(self._onAuthenticationRequired) @@ -325,7 +325,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice): if on_finished is not None: self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = on_finished - def __handleOnFinished(self, reply: QNetworkReply) -> None: + def _handleOnFinished(self, reply: QNetworkReply) -> None: # Due to garbage collection, we need to cache certain bits of post operations. # As we don't want to keep them around forever, delete them if we get a reply. if reply.operation() == QNetworkAccessManager.PostOperation: diff --git a/tests/PrinterOutput/TestNetworkedPrinterOutputDevice.py b/tests/PrinterOutput/TestNetworkedPrinterOutputDevice.py new file mode 100644 index 0000000000..b3f7277051 --- /dev/null +++ b/tests/PrinterOutput/TestNetworkedPrinterOutputDevice.py @@ -0,0 +1,133 @@ +import time +from unittest.mock import MagicMock + +from PyQt5.QtNetwork import QNetworkAccessManager +from PyQt5.QtCore import QUrl +from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState +from cura.PrinterOutputDevice import ConnectionState + + +def test_properties(): + properties = { b"firmware_version": b"12", b"printer_type": b"BHDHAHHADAD", b"address": b"ZOMG", b"name": b":(", b"testProp": b"zomg"} + + output_device = NetworkedPrinterOutputDevice(device_id = "test", address = "127.0.0.1", properties = properties) + assert output_device.address == "ZOMG" + assert output_device.firmwareVersion == "12" + assert output_device.printerType == "BHDHAHHADAD" + assert output_device.ipAddress == "127.0.0.1" + assert output_device.name == ":(" + assert output_device.key == "test" + assert output_device.getProperties() == properties + + assert output_device.getProperty("testProp") == "zomg" + assert output_device.getProperty("whateverr") == "" + + +def test_authenticationState(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + + output_device.setAuthenticationState(AuthState.Authenticated) + + assert output_device.authenticationState == AuthState.Authenticated + + +def test_post(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + mocked_network_manager = MagicMock() + output_device._manager = mocked_network_manager + + # Create a fake reply (we cant use a QReply, since those are abstract C++) + reply = MagicMock() + reply.operation = MagicMock(return_value=QNetworkAccessManager.PostOperation) + reply.url = MagicMock(return_value=QUrl("127.0.0.1")) + mocked_network_manager.post = MagicMock(return_value = reply) + + mocked_callback_handler = MagicMock() + output_device.post("whatever", "omgzomg", on_finished = mocked_callback_handler.onFinished) + + # So we now fake that the request was sucesful. + output_device._handleOnFinished(reply) + + # We expect to get a callback regarding this. + mocked_callback_handler.onFinished.assert_called_once_with(reply) + + +def test_get(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + mocked_network_manager = MagicMock() + output_device._manager = mocked_network_manager + + # Create a fake reply (we cant use a QReply, since those are abstract C++) + reply = MagicMock() + reply.operation = MagicMock(return_value=QNetworkAccessManager.PostOperation) + reply.url = MagicMock(return_value=QUrl("127.0.0.1")) + mocked_network_manager.get = MagicMock(return_value=reply) + + mocked_callback_handler = MagicMock() + output_device.get("whatever", on_finished=mocked_callback_handler.onFinished) + + # So we now fake that the request was sucesful. + output_device._handleOnFinished(reply) + + # We expect to get a callback regarding this. + mocked_callback_handler.onFinished.assert_called_once_with(reply) + + +def test_delete(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + mocked_network_manager = MagicMock() + output_device._manager = mocked_network_manager + + # Create a fake reply (we cant use a QReply, since those are abstract C++) + reply = MagicMock() + reply.operation = MagicMock(return_value=QNetworkAccessManager.PostOperation) + reply.url = MagicMock(return_value=QUrl("127.0.0.1")) + mocked_network_manager.deleteResource = MagicMock(return_value=reply) + + mocked_callback_handler = MagicMock() + output_device.delete("whatever", on_finished=mocked_callback_handler.onFinished) + + # So we now fake that the request was sucesful. + output_device._handleOnFinished(reply) + + # We expect to get a callback regarding this. + mocked_callback_handler.onFinished.assert_called_once_with(reply) + + +def test_put(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + mocked_network_manager = MagicMock() + output_device._manager = mocked_network_manager + + # Create a fake reply (we cant use a QReply, since those are abstract C++) + reply = MagicMock() + reply.operation = MagicMock(return_value=QNetworkAccessManager.PostOperation) + reply.url = MagicMock(return_value=QUrl("127.0.0.1")) + mocked_network_manager.put = MagicMock(return_value = reply) + + mocked_callback_handler = MagicMock() + output_device.put("whatever", "omgzomg", on_finished = mocked_callback_handler.onFinished) + + # So we now fake that the request was sucesful. + output_device._handleOnFinished(reply) + + # We expect to get a callback regarding this. + mocked_callback_handler.onFinished.assert_called_once_with(reply) + + +def test_timeout(): + output_device = NetworkedPrinterOutputDevice(device_id="test", address="127.0.0.1", properties={}) + output_device.setConnectionState(ConnectionState.Connected) + + assert output_device.connectionState == ConnectionState.Connected + output_device._update() + # Pretend we didn't get any response for 15 seconds + output_device._last_response_time = time.time() - 15 + # But we did recently ask for a response! + output_device._last_request_time = time.time() - 5 + output_device._update() + + # The connection should now be closed, since it went into timeout. + assert output_device.connectionState == ConnectionState.Closed + +