1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
# Copyright (c) 2021 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import time
from unittest.mock import MagicMock, patch
from PyQt6.QtNetwork import QNetworkAccessManager
from PyQt6.QtCore import QUrl
from cura.PrinterOutput.NetworkedPrinterOutputDevice import NetworkedPrinterOutputDevice, AuthState
from cura.PrinterOutput.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"}
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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 can't use a QReply, since those are abstract C++)
reply = MagicMock()
reply.operation = MagicMock(return_value=QNetworkAccessManager.Operation.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 successful.
output_device._handleOnFinished(reply)
# We expect to get a callback regarding this.
mocked_callback_handler.onFinished.assert_called_once_with(reply)
def test_get():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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 can't use a QReply, since those are abstract C++)
reply = MagicMock()
reply.operation = MagicMock(return_value=QNetworkAccessManager.Operation.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 successful.
output_device._handleOnFinished(reply)
# We expect to get a callback regarding this.
mocked_callback_handler.onFinished.assert_called_once_with(reply)
def test_delete():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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 can't use a QReply, since those are abstract C++)
reply = MagicMock()
reply.operation = MagicMock(return_value=QNetworkAccessManager.Operation.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 successful.
output_device._handleOnFinished(reply)
# We expect to get a callback regarding this.
mocked_callback_handler.onFinished.assert_called_once_with(reply)
def test_put():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
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 can't use a QReply, since those are abstract C++)
reply = MagicMock()
reply.operation = MagicMock(return_value=QNetworkAccessManager.Operation.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 successful.
output_device._handleOnFinished(reply)
# We expect to get a callback regarding this.
mocked_callback_handler.onFinished.assert_called_once_with(reply)
def test_timeout():
with patch("UM.Qt.QtApplication.QtApplication.getInstance"):
output_device = NetworkedPrinterOutputDevice(device_id = "test", address = "127.0.0.1", properties = {})
with patch("cura.CuraApplication.CuraApplication.getInstance"):
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
with patch("cura.CuraApplication.CuraApplication.getInstance"):
output_device._update()
# The connection should now be closed, since it went into timeout.
assert output_device.connectionState == ConnectionState.Closed
|
