TestCloudOutputDevice.py 7.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157
  1. # Copyright (c) 2018 Ultimaker B.V.
  2. # Cura is released under the terms of the LGPLv3 or higher.
  3. import json
  4. from unittest import TestCase
  5. from unittest.mock import patch, MagicMock
  6. from UM.Scene.SceneNode import SceneNode
  7. from cura.UltimakerCloudAuthentication import CuraCloudAPIRoot
  8. from cura.PrinterOutput.Models.PrinterOutputModel import PrinterOutputModel
  9. from ...src.Cloud import CloudApiClient
  10. from ...src.Cloud.CloudOutputDevice import CloudOutputDevice
  11. from ...src.Cloud.Models.CloudClusterResponse import CloudClusterResponse
  12. from .Fixtures import readFixture, parseFixture
  13. from .NetworkManagerMock import NetworkManagerMock
  14. class TestCloudOutputDevice(TestCase):
  15. maxDiff = None
  16. CLUSTER_ID = "RIZ6cZbWA_Ua7RZVJhrdVfVpf0z-MqaSHQE4v8aRTtYq"
  17. JOB_ID = "ABCDefGHIjKlMNOpQrSTUvYxWZ0-1234567890abcDE="
  18. HOST_NAME = "ultimakersystem-ccbdd30044ec"
  19. HOST_GUID = "e90ae0ac-1257-4403-91ee-a44c9b7e8050"
  20. HOST_VERSION = "5.2.0"
  21. FRIENDLY_NAME = "My Friendly Printer"
  22. STATUS_URL = "{}/connect/v1/clusters/{}/status".format(CuraCloudAPIRoot, CLUSTER_ID)
  23. PRINT_URL = "{}/connect/v1/clusters/{}/print/{}".format(CuraCloudAPIRoot, CLUSTER_ID, JOB_ID)
  24. REQUEST_UPLOAD_URL = "{}/cura/v1/jobs/upload".format(CuraCloudAPIRoot)
  25. def setUp(self):
  26. super().setUp()
  27. self.app = MagicMock()
  28. self.patches = [patch("UM.Qt.QtApplication.QtApplication.getInstance", return_value=self.app),
  29. patch("UM.Application.Application.getInstance", return_value=self.app)]
  30. for patched_method in self.patches:
  31. patched_method.start()
  32. self.cluster = CloudClusterResponse(self.CLUSTER_ID, self.HOST_GUID, self.HOST_NAME, is_online=True,
  33. status="active", host_version=self.HOST_VERSION,
  34. friendly_name=self.FRIENDLY_NAME)
  35. self.network = NetworkManagerMock()
  36. self.account = MagicMock(isLoggedIn=True, accessToken="TestAccessToken")
  37. self.onError = MagicMock()
  38. with patch.object(CloudApiClient, "QNetworkAccessManager", return_value = self.network):
  39. self._api = CloudApiClient.CloudApiClient(self.account, self.onError)
  40. self.device = CloudOutputDevice(self._api, self.cluster)
  41. self.cluster_status = parseFixture("getClusterStatusResponse")
  42. self.network.prepareReply("GET", self.STATUS_URL, 200, readFixture("getClusterStatusResponse"))
  43. def tearDown(self):
  44. try:
  45. super().tearDown()
  46. self.network.flushReplies()
  47. finally:
  48. for patched_method in self.patches:
  49. patched_method.stop()
  50. # We test for these in order to make sure the correct file type is selected depending on the firmware version.
  51. def test_properties(self):
  52. self.assertEqual(self.device.firmwareVersion, self.HOST_VERSION)
  53. self.assertEqual(self.device.name, self.FRIENDLY_NAME)
  54. def test_status(self):
  55. self.device._update()
  56. self.network.flushReplies()
  57. self.assertEqual([PrinterOutputModel, PrinterOutputModel], [type(printer) for printer in self.device.printers])
  58. controller_fields = {
  59. "_output_device": self.device,
  60. "can_abort": True,
  61. "can_control_manually": False,
  62. "can_pause": True,
  63. "can_pre_heat_bed": False,
  64. "can_pre_heat_hotends": False,
  65. "can_send_raw_gcode": False,
  66. "can_update_firmware": False,
  67. }
  68. self.assertEqual({printer["uuid"] for printer in self.cluster_status["data"]["printers"]},
  69. {printer.key for printer in self.device.printers})
  70. self.assertEqual([controller_fields, controller_fields],
  71. [printer.getController().__dict__ for printer in self.device.printers])
  72. self.assertEqual(["UM3PrintJobOutputModel"], [type(printer).__name__ for printer in self.device.printJobs])
  73. self.assertEqual({job["uuid"] for job in self.cluster_status["data"]["print_jobs"]},
  74. {job.key for job in self.device.printJobs})
  75. self.assertEqual({job["owner"] for job in self.cluster_status["data"]["print_jobs"]},
  76. {job.owner for job in self.device.printJobs})
  77. self.assertEqual({job["name"] for job in self.cluster_status["data"]["print_jobs"]},
  78. {job.name for job in self.device.printJobs})
  79. def test_remove_print_job(self):
  80. self.device._update()
  81. self.network.flushReplies()
  82. self.assertEqual(1, len(self.device.printJobs))
  83. self.cluster_status["data"]["print_jobs"].clear()
  84. self.network.prepareReply("GET", self.STATUS_URL, 200, self.cluster_status)
  85. self.device._last_request_time = None
  86. self.device._update()
  87. self.network.flushReplies()
  88. self.assertEqual([], self.device.printJobs)
  89. def test_remove_printers(self):
  90. self.device._update()
  91. self.network.flushReplies()
  92. self.assertEqual(2, len(self.device.printers))
  93. self.cluster_status["data"]["printers"].clear()
  94. self.network.prepareReply("GET", self.STATUS_URL, 200, self.cluster_status)
  95. self.device._last_request_time = None
  96. self.device._update()
  97. self.network.flushReplies()
  98. self.assertEqual([], self.device.printers)
  99. def test_print_to_cloud(self):
  100. active_machine_mock = self.app.getGlobalContainerStack.return_value
  101. active_machine_mock.getMetaDataEntry.side_effect = {"file_formats": "application/x-ufp"}.get
  102. request_upload_response = parseFixture("putJobUploadResponse")
  103. request_print_response = parseFixture("postJobPrintResponse")
  104. self.network.prepareReply("PUT", self.REQUEST_UPLOAD_URL, 201, request_upload_response)
  105. self.network.prepareReply("PUT", request_upload_response["data"]["upload_url"], 201, b"{}")
  106. self.network.prepareReply("POST", self.PRINT_URL, 200, request_print_response)
  107. file_handler = MagicMock()
  108. file_handler.getSupportedFileTypesWrite.return_value = [{
  109. "extension": "ufp",
  110. "mime_type": "application/x-ufp",
  111. "mode": 2
  112. }, {
  113. "extension": "gcode.gz",
  114. "mime_type": "application/gzip",
  115. "mode": 2,
  116. }]
  117. file_handler.getWriterByMimeType.return_value.write.side_effect = \
  118. lambda stream, nodes: stream.write(str(nodes).encode())
  119. scene_nodes = [SceneNode()]
  120. expected_mesh = str(scene_nodes).encode()
  121. self.device.requestWrite(scene_nodes, file_handler=file_handler, file_name="FileName")
  122. self.network.flushReplies()
  123. self.assertEqual(
  124. {"data": {"content_type": "application/x-ufp", "file_size": len(expected_mesh), "job_name": "FileName"}},
  125. json.loads(self.network.getRequestBody("PUT", self.REQUEST_UPLOAD_URL).decode())
  126. )
  127. self.assertEqual(expected_mesh,
  128. self.network.getRequestBody("PUT", request_upload_response["data"]["upload_url"]))
  129. self.assertIsNone(self.network.getRequestBody("POST", self.PRINT_URL))