import importlib import sys import types import unittest from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[2] MODULE_DIR = REPO_ROOT / "src" / "Mod" / "FreeCADExchange" if str(MODULE_DIR) not in sys.path: sys.path.insert(0, str(MODULE_DIR)) def _install_fake_freecad(): fake_freecad = types.ModuleType("FreeCAD") fake_freecad.ActiveDocument = None fake_freecad.Console = types.SimpleNamespace( PrintMessage=lambda *args, **kwargs: None, PrintWarning=lambda *args, **kwargs: None, PrintError=lambda *args, **kwargs: None, ) sys.modules["FreeCAD"] = fake_freecad class FakeViewObject: def __init__(self): self.Visibility = True class FakeObject: def __init__(self, name, type_id): self.Name = name self.Label = name self.TypeId = type_id self.PropertiesList = [] self.Group = [] self.InList = [] self.ViewObject = FakeViewObject() def isDerivedFrom(self, type_name): return self.TypeId == type_name def addProperty(self, prop_type, prop_name, group_name, description): if prop_name not in self.PropertiesList: self.PropertiesList.append(prop_name) def addObject(self, child): if child not in self.Group: self.Group.append(child) if self not in child.InList: child.InList.append(self) def removeObject(self, child): if child in self.Group: self.Group.remove(child) if self in child.InList: child.InList.remove(self) class FakeDocument: def __init__(self): self.Name = "FakeDoc" self.Objects = [] def addObject(self, type_name, name): obj = FakeObject(name, type_name) self.Objects.append(obj) return obj def getObject(self, name): for obj in self.Objects: if obj.Name == name: return obj return None def _reload_modules(): for name in [ "DeviceImport", "TerminalObjects", "WiringObjects", "StaleObjectSync", ]: sys.modules.pop(name, None) terminal_objects = importlib.import_module("TerminalObjects") wiring_objects = importlib.import_module("WiringObjects") stale_object_sync = importlib.import_module("StaleObjectSync") return terminal_objects, wiring_objects, stale_object_sync class StaleObjectSyncTest(unittest.TestCase): def test_marks_devices_terminals_and_wires_missing_from_payload_as_stale(self): _install_fake_freecad() terminal_objects, wiring_objects, stale_object_sync = _reload_modules() doc = FakeDocument() root = terminal_objects.ensure_root_group(doc, "project-1") active_cabinet = doc.addObject("App::DocumentObjectGroup", "QETCabinet_1") active_cabinet.addProperty("App::PropertyString", "QetCabinetInstanceId", "QET Exchange", "") active_cabinet.QetCabinetInstanceId = "1" root.addObject(active_cabinet) stale_cabinet = doc.addObject("App::DocumentObjectGroup", "QETCabinet_2") stale_cabinet.addProperty("App::PropertyString", "QetCabinetInstanceId", "QET Exchange", "") stale_cabinet.QetCabinetInstanceId = "2" root.addObject(stale_cabinet) active_device = doc.addObject("App::Part", "QETDevice_device_active") active_device.addProperty("App::PropertyString", "QetElementUuid", "QET Exchange", "") active_device.QetElementUuid = "device-active" active_device.addProperty("App::PropertyString", "QetInstanceId", "QET Exchange", "") active_device.QetInstanceId = "instance-active" root.addObject(active_device) stale_device = doc.addObject("App::Part", "QETDevice_device_stale") stale_device.addProperty("App::PropertyString", "QetElementUuid", "QET Exchange", "") stale_device.QetElementUuid = "device-stale" stale_device.addProperty("App::PropertyString", "QetInstanceId", "QET Exchange", "") stale_device.QetInstanceId = "instance-stale" root.addObject(stale_device) active_terminals = terminal_objects.ensure_terminal_group( doc, active_device, project_uuid="project-1", instance_id="instance-active", ) active_terminal = doc.addObject("Part::LocalCoordinateSystem", "QETTerminal_terminal_active") terminal_objects.set_terminal_semantics( active_terminal, "project-1", "device-active", "terminal-active", "instance-active", ) active_terminals.addObject(active_terminal) stale_terminals = terminal_objects.ensure_terminal_group( doc, stale_device, project_uuid="project-1", instance_id="instance-stale", ) stale_terminal = doc.addObject("Part::LocalCoordinateSystem", "QETTerminal_terminal_stale") terminal_objects.set_terminal_semantics( stale_terminal, "project-1", "device-stale", "terminal-stale", "instance-stale", ) stale_terminals.addObject(stale_terminal) active_task = wiring_objects.create_wire_task( doc, "project-1", "wire-active", "wire-active", "terminal-active", "terminal-b", "instance-active", "instance-b", ) stale_task = wiring_objects.create_wire_task( doc, "project-1", "wire-stale", "wire-stale", "terminal-stale", "terminal-c", "instance-stale", "instance-c", ) payload = { "project_uuid": "project-1", "cabinet": { "location_id": 1, }, "devices": [ { "device_instance_id": "instance-active", "display_tag": "QF1", "terminals": [ { "terminal_uuid": "terminal-active", "element_uuid": "device-active", } ], } ], "wires": [ { "wire_id": "wire-active", "start_terminal_uuid": "terminal-active", "end_terminal_uuid": "terminal-b", } ], } report = stale_object_sync.mark_stale_objects_from_payload(payload, doc) self.assertEqual(1, report["active_cabinets"]) self.assertEqual(1, report["stale_cabinets"]) self.assertEqual(1, report["active_devices"]) self.assertEqual(1, report["stale_devices"]) self.assertEqual(1, report["active_terminals"]) self.assertEqual(1, report["stale_terminals"]) self.assertEqual(1, report["active_wire_tasks"]) self.assertEqual(1, report["stale_wire_tasks"]) self.assertEqual("Active", active_cabinet.QetSyncStatus) self.assertEqual("Stale", stale_cabinet.QetSyncStatus) self.assertEqual("Active", active_device.QetSyncStatus) self.assertEqual("Stale", stale_device.QetSyncStatus) self.assertEqual("Active", active_terminal.QetSyncStatus) self.assertEqual("Stale", stale_terminal.QetSyncStatus) self.assertEqual("Active", active_task.QetSyncStatus) self.assertEqual("Stale", stale_task.QetSyncStatus) def test_device_stale_check_uses_instance_id_not_element_uuid(self): _install_fake_freecad() terminal_objects, _, stale_object_sync = _reload_modules() doc = FakeDocument() root = terminal_objects.ensure_root_group(doc, "project-1") device = doc.addObject("App::Part", "QETDevice_device_a") device.addProperty("App::PropertyString", "QetElementUuid", "QET Exchange", "") device.QetElementUuid = "shared-element" device.addProperty("App::PropertyString", "QetInstanceId", "QET Exchange", "") device.QetInstanceId = "instance-old" device.addProperty("App::PropertyString", "QetDisplayTag", "QET Exchange", "") device.QetDisplayTag = "QF1" root.addObject(device) payload = { "project_uuid": "project-1", "devices": [ { "device_instance_id": "instance-new", "display_tag": "QF1", "terminals": [ { "terminal_uuid": "terminal-a", "element_uuid": "shared-element", } ], } ], } report = stale_object_sync.mark_stale_objects_from_payload(payload, doc) self.assertEqual(0, report["active_devices"]) self.assertEqual(1, report["stale_devices"]) self.assertEqual("Stale", device.QetSyncStatus) self.assertEqual(1, len(report["stale_device_details"])) self.assertEqual("QF1", report["stale_device_details"][0]["label"]) self.assertEqual("instance-old", report["stale_device_details"][0]["instance_id"]) if __name__ == "__main__": unittest.main()