mirror of
https://github.com/Ultimaker/Cura.git
synced 2025-07-08 23:46:22 -06:00
Added tests for NetworkedPrinterOutputDevice
This commit is contained in:
parent
905787b9b3
commit
5f4907da5c
2 changed files with 136 additions and 3 deletions
|
@ -310,11 +310,11 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice):
|
||||||
def _createNetworkManager(self) -> None:
|
def _createNetworkManager(self) -> None:
|
||||||
Logger.log("d", "Creating network manager")
|
Logger.log("d", "Creating network manager")
|
||||||
if self._manager:
|
if self._manager:
|
||||||
self._manager.finished.disconnect(self.__handleOnFinished)
|
self._manager.finished.disconnect(self._handleOnFinished)
|
||||||
self._manager.authenticationRequired.disconnect(self._onAuthenticationRequired)
|
self._manager.authenticationRequired.disconnect(self._onAuthenticationRequired)
|
||||||
|
|
||||||
self._manager = QNetworkAccessManager()
|
self._manager = QNetworkAccessManager()
|
||||||
self._manager.finished.connect(self.__handleOnFinished)
|
self._manager.finished.connect(self._handleOnFinished)
|
||||||
self._last_manager_create_time = time()
|
self._last_manager_create_time = time()
|
||||||
self._manager.authenticationRequired.connect(self._onAuthenticationRequired)
|
self._manager.authenticationRequired.connect(self._onAuthenticationRequired)
|
||||||
|
|
||||||
|
@ -325,7 +325,7 @@ class NetworkedPrinterOutputDevice(PrinterOutputDevice):
|
||||||
if on_finished is not None:
|
if on_finished is not None:
|
||||||
self._onFinishedCallbacks[reply.url().toString() + str(reply.operation())] = on_finished
|
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.
|
# 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.
|
# As we don't want to keep them around forever, delete them if we get a reply.
|
||||||
if reply.operation() == QNetworkAccessManager.PostOperation:
|
if reply.operation() == QNetworkAccessManager.PostOperation:
|
||||||
|
|
133
tests/PrinterOutput/TestNetworkedPrinterOutputDevice.py
Normal file
133
tests/PrinterOutput/TestNetworkedPrinterOutputDevice.py
Normal file
|
@ -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
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue