[tests] Test suite for node attributeChanged callback mechanism

Suite of tests to validate the behavior of desc.Node 'on{Attribute}Changed' behavior.

Author: yann-lty

tests/test_nodeAttributeChangedCallback.py

@@ -0,0 +1,162 @@
+# coding:utf-8
+
+from meshroom.core.graph import Graph, loadGraph
+from meshroom.core import desc, registerNodeType, unregisterNodeType
+from meshroom.core.node import Node
+
+
+class NodeWithAttributeChangedCallback(desc.Node):
+    """
+    A Node containing an input Attribute with an 'on{Attribute}Changed' method,
+    called whenever the value of this attribute is changed explicitly.
+    """
+
+    inputs = [
+        desc.IntParam(
+            name="input",
+            label="Input",
+            description="Attribute with a value changed callback (onInputChanged)",
+            value=0,
+            range=None,
+        ),
+        desc.IntParam(
+            name="affectedInput",
+            label="Affected Input",
+            description="Updated to input.value * 2 whenever 'input' is explicitly modified",
+            value=0,
+            range=None,
+        ),
+    ]
+
+    def onInputChanged(self, instance: Node):
+        instance.affectedInput.value = instance.input.value * 2
+
+    def processChunk(self, chunk):
+        pass  # No-op.
+
+
+
+class TestNodeWithAttributeChangedCallback:
+    @classmethod
+    def setup_class(cls):
+        registerNodeType(NodeWithAttributeChangedCallback)
+
+    @classmethod
+    def teardown_class(cls):
+        unregisterNodeType(NodeWithAttributeChangedCallback)
+
+    def test_assignValueTriggersCallback(self):
+        node = Node(NodeWithAttributeChangedCallback.__name__)
+        assert node.affectedInput.value == 0
+
+        node.input.value = 10
+        assert node.affectedInput.value == 20
+
+    def test_specifyDefaultValueDoesNotTriggerCallback(self):
+        node = Node(NodeWithAttributeChangedCallback.__name__, input=10)
+        assert node.affectedInput.value == 0
+
+    def test_assignDefaultValueDoesNotTriggerCallback(self):
+        node = Node(NodeWithAttributeChangedCallback.__name__, input=10)
+        node.input.value = 10
+        assert node.affectedInput.value == 0
+
+    def test_assignNonDefaultValueTriggersCallback(self):
+        node = Node(NodeWithAttributeChangedCallback.__name__, input=10)
+        node.input.value = 2
+        assert node.affectedInput.value == 4
+
+
+class TestAttributeCallbackTriggerInGraph:
+    @classmethod
+    def setup_class(cls):
+        registerNodeType(NodeWithAttributeChangedCallback)
+
+    @classmethod
+    def teardown_class(cls):
+        unregisterNodeType(NodeWithAttributeChangedCallback)
+
+    def test_connectionTriggersCallback(self):
+        graph = Graph("")
+        nodeA = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeB = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        assert nodeA.affectedInput.value == nodeB.affectedInput.value == 0
+
+        nodeA.input.value = 1
+        graph.addEdge(nodeA.input, nodeB.input)
+
+        assert nodeA.affectedInput.value == nodeB.affectedInput.value == 2
+
+    def test_connectedValueChangeTriggersCallback(self):
+        graph = Graph("")
+        nodeA = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeB = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        assert nodeA.affectedInput.value == nodeB.affectedInput.value == 0
+
+        graph.addEdge(nodeA.input, nodeB.input)
+        nodeA.input.value = 1
+
+        assert nodeA.affectedInput.value == 2
+        assert nodeB.affectedInput.value == 2
+
+    def test_defaultValueOnlyTriggersCallbackDownstream(self):
+        graph = Graph("")
+        nodeA = graph.addNewNode(NodeWithAttributeChangedCallback.__name__, input=1)
+        nodeB = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        assert nodeA.affectedInput.value == 0
+        assert nodeB.affectedInput.value == 0
+
+        graph.addEdge(nodeA.input, nodeB.input)
+
+        assert nodeA.affectedInput.value == 0
+        assert nodeB.affectedInput.value == 2
+
+    def test_valueChangeIsPropagatedAlongNodeChain(self):
+        graph = Graph("")
+        nodeA = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeB = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeC = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeD = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        graph.addEdges(
+            (nodeA.affectedInput, nodeB.input),
+            (nodeB.affectedInput, nodeC.input),
+            (nodeC.affectedInput, nodeD.input),
+        )
+
+        nodeA.input.value = 5
+
+        assert nodeA.affectedInput.value == nodeB.input.value == 10
+        assert nodeB.affectedInput.value == nodeC.input.value == 20
+        assert nodeC.affectedInput.value == nodeD.input.value == 40
+        assert nodeD.affectedInput.value == 80
+
+    def test_disconnectionTriggersCallback(self):
+        graph = Graph("")
+        nodeA = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+        nodeB = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        graph.addEdge(nodeA.input, nodeB.input)
+        nodeA.input.value = 5
+        assert nodeB.affectedInput.value == 10
+
+        graph.removeEdge(nodeB.input)
+
+        assert nodeB.input.value == 0
+        assert nodeB.affectedInput.value == 0
+
+    def test_loadingGraphDoesNotTriggerCallback(self, graphSavedOnDisk):
+        graph: Graph = graphSavedOnDisk
+        node = graph.addNewNode(NodeWithAttributeChangedCallback.__name__)
+
+        node.input.value = 5
+        node.affectedInput.value = 2
+        graph.save()
+
+        loadedGraph = loadGraph(graph.filepath)
+        loadedNode = loadedGraph.node(node.name)
+        assert loadedNode
+        assert loadedNode.affectedInput.value == 2