feat: support serialisation of deployment nodes within deployment nodes

Author: yt-ms

src/structurizr/model/deployment_node.py

@@ -15,50 +15,136 @@
 
 """Provide a deployment node model."""
 
+from typing import TYPE_CHECKING, Iterable, List
 
-from typing import Iterable
+from pydantic import Field
 
 from .deployment_element import DeploymentElement, DeploymentElementIO
 
 
+if TYPE_CHECKING:
+    from .model import Model
+
 __all__ = ("DeploymentNode", "DeploymentNodeIO")
 
 
 class DeploymentNodeIO(DeploymentElementIO):
-    """Represent a deployment node."""
+    """
+    Represent a deployment node.
+
+    Attributes:
+        id: The ID of this deployment node in the model.
+        name: The name of this node.
+        description: A short description of this node.
+        environment (str):
+        tags: A comma separated list of tags associated with this node.
+        children: The deployment nodes that are direct children of this node.
+        properties: A set of arbitrary name-value properties.
+        relationships: The set of relationships from this node to
+                       other elements.
+        url (pydantic.HttpUrl):
+    """
+
+    class Config:
+        """Pydantic configuration for DeploymentNodeIO."""
+
+        # Prevent infinite recursion for `children` - see
+        # https://github.com/samuelcolvin/pydantic/issues/524
+        validate_assignment = "limited"
 
-    parent: "DeploymentNodeIO"
     technology: str = ""
     instances: int = 1
+    children: List["DeploymentNodeIO"] = Field(default=())
+
+
+DeploymentNodeIO.update_forward_refs()
 
 
 class DeploymentNode(DeploymentElement):
-    """Represent a deployment node."""
+    """
+    Represent a deployment node.
+
+    Attributes:
+        id: The ID of this deployment node in the model.
+        name: The name of this node.
+        description: A short description of this node.
+        environment (str):
+        tags: A comma separated list of tags associated with this node.
+        children: The deployment nodes that are direct children of this node.
+        properties: A set of arbitrary name-value properties.
+        relationships: The set of relationships from this node to
+                       other elements.
+        url (pydantic.HttpUrl):
+    """
 
     def __init__(
         self,
         *,
-        parent: "DeploymentNode",
+        parent: "DeploymentNode" = None,
         technology: str = "",
         instances: int = 1,
         children: Iterable["DeploymentNode"] = (),
         container_instances: Iterable["DeploymentNode"] = (),
-        **kwargs
+        **kwargs,
     ) -> None:
         """Initialize a deployment node."""
         super().__init__(**kwargs)
         self.parent = parent
         self.technology = technology
         self.instances = instances
-        self.children = set(children)
-        self.container_instances = set(container_instances)
+        self._children = set(children)
+        self._container_instances = set(container_instances)
+
+    @property
+    def children(self) -> Iterable["DeploymentNode"]:
+        """Return read-only list of child nodes."""
+        return list(self._children)
+
+    def add_deployment_node(self, **kwargs) -> "DeploymentNode":
+        """Add a new child deployment node to this node."""
+        node = DeploymentNode(**kwargs)
+        self += node
+        return node
+
+    def __iadd__(self, node: "DeploymentNode") -> "DeploymentNode":
+        """Add a newly constructed chile deployment node to this node."""
+        if node in self._children:
+            return self
+
+        # if self.get_component_with_name(component.name):
+        #     raise ValueError(
+        #         f"Component with name {component.name} already exists in {self}."
+        #     )
+
+        if node.parent is None:
+            node.parent = self
+        elif node.parent is not self:
+            raise ValueError(
+                f"DeploymentNode with name {node.name} already has parent "
+                f"{node.parent}. Cannot add to {self}."
+            )
+        self._children.add(node)
+        model = self.model
+        model += node
+        return self
 
     @classmethod
-    def hydrate(cls, deployment_node_io: DeploymentNodeIO) -> "DeploymentNode":
+    def hydrate(
+        cls,
+        deployment_node_io: DeploymentNodeIO,
+        model: "Model",
+        parent: "DeploymentNode" = None,
+    ) -> "DeploymentNode":
         """Hydrate a new DeploymentNode instance from its IO."""
-        # TODO (midnighter): Initialization requires `parent`.
-        return cls(
-            # parent=deployment_node_io.parent,
+        node = cls(
+            parent=parent,
             name=deployment_node_io.name,
             description=deployment_node_io.description,
         )
+        model += node
+
+        for child_io in deployment_node_io.children:
+            child_node = DeploymentNode.hydrate(child_io, model, node)
+            node += child_node
+
+        return node

tests/unit/model/test_deployment_node.py

@@ -0,0 +1,83 @@
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     https://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+"""Ensure the expected behaviour of the container element."""
+
+
+import pytest
+
+from structurizr.model.deployment_node import DeploymentNode, DeploymentNodeIO
+
+
+class MockModel:
+    """Implement a mock model for testing."""
+
+    def __init__(self):
+        """Initialize the mock, creating an empty node for tests."""
+        self.empty_node = DeploymentNode(name="Empty")
+        self.empty_node.set_model(self)
+
+    def __iadd__(self, node):
+        """Simulate the model assigning IDs to new elements."""
+        if not node.id:
+            node.id = "id"
+        node.set_model(self)
+        return self
+
+
+@pytest.fixture(scope="function")
+def model_with_node() -> MockModel:
+    """Provide an new empty model on demand for test cases to use."""
+    return MockModel()
+
+
+@pytest.mark.parametrize(
+    "attributes",
+    [
+        pytest.param({}, marks=pytest.mark.raises(exception=TypeError)),
+        {"name": "Node1", "technology": "tech1"},
+    ],
+)
+def test_deployment_node_init(attributes):
+    """Expect proper initialization from arguments."""
+    node = DeploymentNode(**attributes)
+    for attr, expected in attributes.items():
+        assert getattr(node, attr) == expected
+
+
+def test_deployment_node_adds_to_children(model_with_node):
+    """Test adding a child node to an existing one."""
+    top_node = model_with_node.empty_node
+    child = top_node.add_deployment_node(name="child")
+
+    assert child is not None
+    assert child.model == model_with_node
+    assert child.parent == top_node
+    assert child in top_node.children
+
+
+def test_deployment_node_serialization_of_recursive_nodes(model_with_node):
+    """Check that nodes within nodes are handled with (de)serialisation."""
+    top_node = model_with_node.empty_node
+    top_node.add_deployment_node(name="child")
+
+    io = DeploymentNodeIO.from_orm(top_node)
+    assert len(io.children) == 1
+    assert io.children[0].name == "child"
+
+    new_top_node = DeploymentNode.hydrate(io, model_with_node)
+    assert len(new_top_node.children) == 1
+    new_child = new_top_node.children[0]
+    assert new_child.name == "child"
+    assert new_child.parent is new_top_node
+    assert new_child.model is model_with_node