3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-07 23:17:32 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Added tests for SendMaterialJob and refactored SendMaterialJob for

Browse source 
better testability. This is part of a larger project to create tests
for the UM3NetworkPrinting plugin in preparation for printing from the
cloud
This commit is contained in:
Marijn Deé 2018-11-16 14:16:45 +01:00
parent bfe5ae6f9c
commit e9e8c49b2d
3 changed files with 491 additions and 47 deletions
Download patch file Download diff file Expand all files Collapse all files

87
plugins/UM3NetworkPrinting/src/Models.py Normal file
View file

@ -0,0 +1,87 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
from typing import Optional
class BaseModel:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
def __eq__(self, other):
return self.__dict__ == other.__dict__ if type(self) == type(other) else False
## Represents an item in the cluster API response for installed materials.
class ClusterMaterial(BaseModel):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.version = int(self.version)
self.density = float(self.density)
guid: None # type: Optional[str]
material: None # type: Optional[str]
brand: None # type: Optional[str]
version = None # type: Optional[int]
color: None # type: Optional[str]
density: None # type: Optional[float]
class LocalMaterialProperties(BaseModel):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.density = float(self.density)
self.diameter = float(self.diameter)
self.weight = float(self.weight)
density: None # type: Optional[float]
diameter: None # type: Optional[float]
weight: None # type: Optional[int]
class LocalMaterial(BaseModel):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.properties = LocalMaterialProperties(**self.properties)
self.approximate_diameter = float(self.approximate_diameter)
self.version = int(self.version)
GUID: None # type: Optional[str]
id: None # type: Optional[str]
type: None # type: Optional[str]
status: None # type: Optional[str]
base_file: None # type: Optional[str]
setting_version: None # type: Optional[str]
version = None # type: Optional[int]
name: None # type: Optional[str]
brand: None # type: Optional[str]
material: None # type: Optional[str]
color_name: None # type: Optional[str]
description: None # type: Optional[str]
adhesion_info: None # type: Optional[str]
approximate_diameter: None # type: Optional[float]
properties: None # type: LocalMaterialProperties
definition: None # type: Optional[str]
compatible: None # type: Optional[bool]

122
plugins/UM3NetworkPrinting/src/SendMaterialJob.py
View file

@ -1,99 +1,129 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import json #To understand the list of materials from the printer reply.
import os #To walk over material files.
import os.path #To filter on material files.
from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest #To listen to the reply from the printer.
from typing import Any, Dict, Set, TYPE_CHECKING
import urllib.parse #For getting material IDs from their file names.
import json # To understand the list of materials from the printer reply.
import os # To walk over material files.
import os.path # To filter on material files.
import urllib.parse # For getting material IDs from their file names.
from typing import Dict, TYPE_CHECKING
from UM.Job import Job #The interface we're implementing.
from PyQt5.QtNetwork import QNetworkReply, QNetworkRequest # To listen to the reply from the printer.
from UM.Job import Job # The interface we're implementing.
from UM.Logger import Logger
from UM.MimeTypeDatabase import MimeTypeDatabase #To strip the extensions of the material profile files.
from UM.MimeTypeDatabase import MimeTypeDatabase # To strip the extensions of the material profile files.
from UM.Resources import Resources
from UM.Settings.ContainerRegistry import ContainerRegistry #To find the GUIDs of materials.
from cura.CuraApplication import CuraApplication #For the resource types.
from UM.Settings.ContainerRegistry import ContainerRegistry # To find the GUIDs of materials.
from cura.CuraApplication import CuraApplication # For the resource types.
from plugins.UM3NetworkPrinting.src.Models import ClusterMaterial, LocalMaterial
if TYPE_CHECKING:
from .ClusterUM3OutputDevice import ClusterUM3OutputDevice
## Asynchronous job to send material profiles to the printer.
#
# This way it won't freeze up the interface while sending those materials.
class SendMaterialJob(Job):
def __init__(self, device: "ClusterUM3OutputDevice") -> None:
super().__init__()
self.device = device #type: ClusterUM3OutputDevice
self.device = device # type: ClusterUM3OutputDevice
def run(self) -> None:
self.device.get("materials/", on_finished = self.sendMissingMaterials)
self.device.get("materials/", on_finished=self.sendMissingMaterials)
def sendMissingMaterials(self, reply: QNetworkReply) -> None:
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200: #Got an error from the HTTP request.
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200: # Got an error from the HTTP request.
Logger.log("e", "Couldn't request current material storage on printer. Not syncing materials.")
return
remote_materials_list = reply.readAll().data().decode("utf-8")
# Collect materials from the printer's reply
try:
remote_materials_list = json.loads(remote_materials_list)
remote_materials_by_guid = self._parseReply(reply)
except json.JSONDecodeError:
Logger.log("e", "Request material storage on printer: I didn't understand the printer's answer.")
return
try:
remote_materials_by_guid = {material["guid"]: material for material in remote_materials_list} #Index by GUID.
except KeyError:
Logger.log("e", "Request material storage on printer: Printer's answer was missing GUIDs.")
return
container_registry = ContainerRegistry.getInstance()
local_materials_list = filter(lambda material: ("GUID" in material and "version" in material and "id" in material), container_registry.findContainersMetadata(type = "material"))
local_materials_by_guid = {material["GUID"]: material for material in local_materials_list if material["id"] == material["base_file"]}
for material in local_materials_list: #For each GUID get the material with the highest version number.
try:
if int(material["version"]) > local_materials_by_guid[material["GUID"]]["version"]:
local_materials_by_guid[material["GUID"]] = material
except ValueError:
Logger.log("e", "Material {material_id} has invalid version number {number}.".format(material_id = material["id"], number = material["version"]))
continue
# Collect local materials
local_materials_by_guid = self._getLocalMaterials()
materials_to_send = set() #type: Set[Dict[str, Any]]
for guid, material in local_materials_by_guid.items():
if guid not in remote_materials_by_guid:
materials_to_send.add(material["id"])
continue
try:
if int(material["version"]) > remote_materials_by_guid[guid]["version"]:
materials_to_send.add(material["id"])
continue
except KeyError:
Logger.log("e", "Current material storage on printer was an invalid reply (missing version).")
return
# Find out what materials are new or updated annd must be sent to the printer
materials_to_send = {
material.id
for guid, material in local_materials_by_guid.items()
if guid not in remote_materials_by_guid or
material.version > remote_materials_by_guid[guid].version
}
# Send materials to the printer
self.sendMaterialsToPrinter(materials_to_send)
def sendMaterialsToPrinter(self, materials_to_send):
for file_path in Resources.getAllResourcesOfType(CuraApplication.ResourceTypes.MaterialInstanceContainer):
try:
mime_type = MimeTypeDatabase.getMimeTypeForFile(file_path)
except MimeTypeDatabase.MimeTypeNotFoundError:
continue #Not the sort of file we'd like to send then.
continue # Not the sort of file we'd like to send then.
_, file_name = os.path.split(file_path)
material_id = urllib.parse.unquote_plus(mime_type.stripExtension(file_name))
if material_id not in materials_to_send:
continue
parts = []
with open(file_path, "rb") as f:
parts.append(self.device._createFormPart("name=\"file\"; filename=\"{file_name}\"".format(file_name = file_name), f.read()))
parts.append(
self.device._createFormPart("name=\"file\"; filename=\"{file_name}\"".format(file_name=file_name),
f.read()))
signature_file_path = file_path + ".sig"
if os.path.exists(signature_file_path):
_, signature_file_name = os.path.split(signature_file_path)
with open(signature_file_path, "rb") as f:
parts.append(self.device._createFormPart("name=\"signature_file\"; filename=\"{file_name}\"".format(file_name = signature_file_name), f.read()))
parts.append(self.device._createFormPart(
"name=\"signature_file\"; filename=\"{file_name}\"".format(file_name=signature_file_name),
f.read()))
Logger.log("d", "Syncing material {material_id} with cluster.".format(material_id = material_id))
self.device.postFormWithParts(target = "materials/", parts = parts, on_finished = self.sendingFinished)
Logger.log("d", "Syncing material {material_id} with cluster.".format(material_id=material_id))
self.device.postFormWithParts(target="materials/", parts=parts, on_finished=self.sendingFinished)
def sendingFinished(self, reply: QNetworkReply):
if reply.attribute(QNetworkRequest.HttpStatusCodeAttribute) != 200:
Logger.log("e", "Received error code from printer when syncing material: {code}".format(code = reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)))
Logger.log("e", "Received error code from printer when syncing material: {code}".format(
code=reply.attribute(QNetworkRequest.HttpStatusCodeAttribute)))
Logger.log("e", reply.readAll().data().decode("utf-8"))
## Parse the reply from the printer
#
# Parses the reply to a "/materials" request to the printer
#
# \return a dictionary of ClustMaterial objects by GUID
# \throw json.JSONDecodeError Raised when the reply does not contain a valid json string
# \throw KeyErrror Raised when on of the materials does not include a valid guid
@classmethod
def _parseReply(cls, reply: QNetworkReply) -> Dict[str, ClusterMaterial]:
remote_materials_list = json.loads(reply.readAll().data().decode("utf-8"))
return {material["guid"]: ClusterMaterial(**material) for material in remote_materials_list}
## Retrieves a list of local materials
#
# Only the new newest version of the local materials is returned
#
# \return a dictionary of LocalMaterial objects by GUID
@classmethod
def _getLocalMaterials(cls):
result = {}
for material in ContainerRegistry.getInstance().findContainersMetadata(type="material"):
try:
localMaterial = LocalMaterial(**material)
if localMaterial.GUID not in result or localMaterial.version > result.get(localMaterial.GUID).version:
result[localMaterial.GUID] = localMaterial
except (ValueError):
Logger.log("e", "Material {material_id} has invalid version number {number}.".format(
material_id=material["id"], number=material["version"]))
return result

327
tests/plugins/UM3NetworkPrinting/TestSendMaterialJob.py Normal file
View file

@ -0,0 +1,327 @@
# Copyright (c) 2018 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher.
import io
import json
from typing import Any, List
from unittest import TestCase, mock
from unittest.mock import patch, call
from PyQt5.QtCore import QByteArray
from UM.Logger import Logger
from UM.MimeTypeDatabase import MimeType
from UM.Settings.ContainerRegistry import ContainerInterface, ContainerRegistryInterface, \
DefinitionContainerInterface, ContainerRegistry
from plugins.UM3NetworkPrinting.src.ClusterUM3OutputDevice import ClusterUM3OutputDevice
from plugins.UM3NetworkPrinting.src.Models import ClusterMaterial
from plugins.UM3NetworkPrinting.src.SendMaterialJob import SendMaterialJob
# All log entries written to Log.log by the class-under-test are written to this list. It is cleared before each test
# run and check afterwards
_logentries = []
def new_log(*args):
_logentries.append(args)
class TestContainerRegistry(ContainerRegistryInterface):
def __init__(self):
self.containersMetaData = None
def findContainers(self, *, ignore_case: bool = False, **kwargs: Any) -> List[ContainerInterface]:
raise NotImplementedError()
def findDefinitionContainers(self, **kwargs: Any) -> List[DefinitionContainerInterface]:
raise NotImplementedError()
@classmethod
def getApplication(cls) -> "Application":
raise NotImplementedError()
def getEmptyInstanceContainer(self) -> "InstanceContainer":
raise NotImplementedError()
def isReadOnly(self, container_id: str) -> bool:
raise NotImplementedError()
def setContainersMetadata(self, value):
self.containersMetaData = value
def findContainersMetadata(self, type):
return self.containersMetaData
class FakeDevice(ClusterUM3OutputDevice):
def _createFormPart(self, content_header, data, content_type=None):
return "xxx"
class TestSendMaterialJob(TestCase):
_LOCALMATERIAL_WHITE = {'type': 'material', 'status': 'unknown', 'id': 'generic_pla_white',
'base_file': 'generic_pla_white', 'setting_version': 5, 'name': 'White PLA',
'brand': 'Generic', 'material': 'PLA', 'color_name': 'White',
'GUID': 'badb0ee7-87c8-4f3f-9398-938587b67dce', 'version': '1', 'color_code': '#ffffff',
'description': 'Test PLA White', 'adhesion_info': 'Use glue.', 'approximate_diameter': '3',
'properties': {'density': '1.00', 'diameter': '2.85', 'weight': '750'},
'definition': 'fdmprinter', 'compatible': True}
_LOCALMATERIAL_BLACK = {'type': 'material', 'status': 'unknown', 'id': 'generic_pla_black',
'base_file': 'generic_pla_black', 'setting_version': 5, 'name': 'Yellow CPE',
'brand': 'Ultimaker', 'material': 'CPE', 'color_name': 'Black',
'GUID': '5fbb362a-41f9-4818-bb43-15ea6df34aa4', 'version': '1', 'color_code': '#000000',
'description': 'Test PLA Black', 'adhesion_info': 'Use glue.', 'approximate_diameter': '3',
'properties': {'density': '1.01', 'diameter': '2.85', 'weight': '750'},
'definition': 'fdmprinter', 'compatible': True}
_REMOTEMATERIAL_WHITE = {
"guid": "badb0ee7-87c8-4f3f-9398-938587b67dce",
"material": "PLA",
"brand": "Generic",
"version": 1,
"color": "White",
"density": 1.00
}
_REMOTEMATERIAL_BLACK = {
"guid": "5fbb362a-41f9-4818-bb43-15ea6df34aa4",
"material": "PLA",
"brand": "Generic",
"version": 2,
"color": "Black",
"density": 1.00
}
def setUp(self):
# Make sure the we start with clean (log) slate
_logentries.clear()
def tearDown(self):
# If there are still log entries that were not checked something is wrong or we must add checks for them
self.assertEqual(len(_logentries), 0)
@patch("plugins.UM3NetworkPrinting.src.ClusterUM3OutputDevice")
def test_run(self, device_mock):
with mock.patch.object(Logger, 'log', new=new_log):
job = SendMaterialJob(device_mock)
job.run()
device_mock.get.assert_called_with("materials/", on_finished=job.sendMissingMaterials)
self.assertEqual(0, len(_logentries))
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_withFailedRequest(self, reply_mock):
reply_mock.attribute.return_value = 404
with mock.patch.object(Logger, 'log', new=new_log):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0)])
self._assertLogEntries([('e', "Couldn't request current material storage on printer. Not syncing materials.")],
_logentries)
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_withBadJsonAnswer(self, reply_mock):
reply_mock.attribute.return_value = 200
reply_mock.readAll.return_value = QByteArray(b'Six sick hicks nick six slick bricks with picks and sticks.')
with mock.patch.object(Logger, 'log', new=new_log):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries(
[('e', "Request material storage on printer: I didn't understand the printer's answer.")],
_logentries)
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_withMissingGuid(self, reply_mock):
reply_mock.attribute.return_value = 200
remoteMaterialWithoutGuid = self._REMOTEMATERIAL_WHITE.copy()
del remoteMaterialWithoutGuid["guid"]
reply_mock.readAll.return_value = QByteArray(json.dumps([remoteMaterialWithoutGuid]).encode("ascii"))
with mock.patch.object(Logger, 'log', new=new_log):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries(
[('e', "Request material storage on printer: Printer's answer was missing GUIDs.")],
_logentries)
@patch("UM.Resources.Resources.getAllResourcesOfType", lambda _: [])
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_WithInvalidVersionInLocalMaterial(self, reply_mock):
reply_mock.attribute.return_value = 200
reply_mock.readAll.return_value = QByteArray(json.dumps([self._REMOTEMATERIAL_WHITE]).encode("ascii"))
containerRegistry = TestContainerRegistry()
localMaterialWhiteWithInvalidVersion = self._LOCALMATERIAL_WHITE.copy()
localMaterialWhiteWithInvalidVersion["version"] = "one"
containerRegistry.setContainersMetadata([localMaterialWhiteWithInvalidVersion])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries([('e', "Material generic_pla_white has invalid version number one.")], _logentries)
@patch("UM.Resources.Resources.getAllResourcesOfType", lambda _: [])
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_WithMultipleLocalVersionsLowFirst(self, reply_mock):
reply_mock.attribute.return_value = 200
reply_mock.readAll.return_value = QByteArray(json.dumps([self._REMOTEMATERIAL_WHITE]).encode("ascii"))
containerRegistry = TestContainerRegistry()
localMaterialWhiteWithHigherVersion = self._LOCALMATERIAL_WHITE.copy()
localMaterialWhiteWithHigherVersion["version"] = "2"
containerRegistry.setContainersMetadata([self._LOCALMATERIAL_WHITE, localMaterialWhiteWithHigherVersion])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries([], _logentries)
@patch("UM.Resources.Resources.getAllResourcesOfType", lambda _: [])
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendMissingMaterials_MaterialMissingOnPrinter(self, reply_mock):
reply_mock.attribute.return_value = 200
reply_mock.readAll.return_value = QByteArray(
json.dumps([self._REMOTEMATERIAL_WHITE]).encode("ascii"))
containerRegistry = TestContainerRegistry()
containerRegistry.setContainersMetadata([self._LOCALMATERIAL_WHITE, self._LOCALMATERIAL_BLACK])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries([], _logentries)
@patch("builtins.open", lambda a, b: io.StringIO("<xml></xml>"))
@patch("UM.MimeTypeDatabase.MimeTypeDatabase.getMimeTypeForFile",
lambda _: MimeType(name="application/x-ultimaker-material-profile", comment="Ultimaker Material Profile",
suffixes=["xml.fdm_material"]))
@patch("UM.Resources.Resources.getAllResourcesOfType", lambda _: ["/materials/generic_pla_white.xml.fdm_material"])
@patch("plugins.UM3NetworkPrinting.src.ClusterUM3OutputDevice")
def test_sendMaterialsToPrinter(self, device):
with mock.patch.object(Logger, "log", new=new_log):
SendMaterialJob(device).sendMaterialsToPrinter({'generic_pla_white'})
self._assertLogEntries([("d", "Syncing material generic_pla_white with cluster.")], _logentries)
@patch("PyQt5.QtNetwork.QNetworkReply")
def xtest_sendMissingMaterials(self, reply_mock):
reply_mock.attribute.return_value = 200
reply_mock.readAll.return_value = QByteArray(
json.dumps([self._REMOTEMATERIAL_WHITE], self._REMOTEMATERIAL_BLACK).encode("ascii"))
containerRegistry = TestContainerRegistry()
containerRegistry.setContainersMetadata([self._LOCALMATERIAL_WHITE, self._LOCALMATERIAL_BLACK])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
SendMaterialJob(None).sendMissingMaterials(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.readAll()])
self._assertLogEntries([], _logentries)
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendingFinished_success(self, reply_mock) -> None:
reply_mock.attribute.return_value = 200
with mock.patch.object(Logger, 'log', new=new_log):
SendMaterialJob(None).sendingFinished(reply_mock)
reply_mock.attribute.assert_called_once_with(0)
self.assertEqual(0, len(_logentries))
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_sendingFinished_failed(self, reply_mock) -> None:
reply_mock.attribute.return_value = 404
reply_mock.readAll.return_value = QByteArray(b'Six sick hicks nick six slick bricks with picks and sticks.')
with mock.patch.object(Logger, 'log', new=new_log):
SendMaterialJob(None).sendingFinished(reply_mock)
reply_mock.attribute.assert_called_with(0)
self.assertEqual(reply_mock.method_calls, [call.attribute(0), call.attribute(0), call.readAll()])
self._assertLogEntries([
("e", "Received error code from printer when syncing material: 404"),
("e", "Six sick hicks nick six slick bricks with picks and sticks.")
], _logentries)
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_parseReply(self, reply_mock):
reply_mock.readAll.return_value = QByteArray(json.dumps([self._REMOTEMATERIAL_WHITE]).encode("ascii"))
response = SendMaterialJob._parseReply(reply_mock)
self.assertTrue(len(response) == 1)
self.assertEqual(next(iter(response.values())), ClusterMaterial(**self._REMOTEMATERIAL_WHITE))
@patch("PyQt5.QtNetwork.QNetworkReply")
def test_parseReplyWithInvalidMaterial(self, reply_mock):
remoteMaterialWithInvalidVersion = self._REMOTEMATERIAL_WHITE.copy()
remoteMaterialWithInvalidVersion["version"] = "one"
reply_mock.readAll.return_value = QByteArray(json.dumps([remoteMaterialWithInvalidVersion]).encode("ascii"))
with self.assertRaises(ValueError):
SendMaterialJob._parseReply(reply_mock)
def test__getLocalMaterials(self):
containerRegistry = TestContainerRegistry()
containerRegistry.setContainersMetadata([self._LOCALMATERIAL_WHITE, self._LOCALMATERIAL_BLACK])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
local_materials = SendMaterialJob(None)._getLocalMaterials()
self.assertTrue(len(local_materials) == 2)
def test__getLocalMaterialsWithMultipleVersions(self):
containerRegistry = TestContainerRegistry()
localMaterialWithNewerVersion = self._LOCALMATERIAL_WHITE.copy()
localMaterialWithNewerVersion["version"] = 2
containerRegistry.setContainersMetadata([self._LOCALMATERIAL_WHITE, localMaterialWithNewerVersion])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
local_materials = SendMaterialJob(None)._getLocalMaterials()
self.assertTrue(len(local_materials) == 1)
self.assertTrue(list(local_materials.values())[0].version == 2)
containerRegistry.setContainersMetadata([localMaterialWithNewerVersion, self._LOCALMATERIAL_WHITE])
with mock.patch.object(Logger, "log", new=new_log):
with mock.patch.object(ContainerRegistry, "getInstance", lambda: containerRegistry):
local_materials = SendMaterialJob(None)._getLocalMaterials()
self.assertTrue(len(local_materials) == 1)
self.assertTrue(list(local_materials.values())[0].version == 2)
def _assertLogEntries(self, first, second):
"""
Inspects the two sets of log entry tuples and fails when they are not the same
:param first: The first set of tuples
:param second: The second set of tuples
"""
self.assertEqual(len(first), len(second))
while len(first) > 0:
e1, m1 = first[0]
e2, m2 = second[0]
self.assertEqual(e1, e2)
self.assertEqual(m1, m2)
first.pop(0)
second.pop(0)