Add validation for hierarchy in RelationshipManager

backend/infrahub/core/relationship/model.py

@@ -734,6 +734,8 @@ async def init(
         if not isinstance(data, list):
             data = [data]
 
+        await rm._validate_hierarchy()
+
         for item in data:
             if not isinstance(item, (rm.rel_class, str, dict)) and not hasattr(item, "_schema"):
                 raise ValidationError({rm.name: f"Invalid data provided to form a relationship {item}"})
@@ -966,6 +968,8 @@ async def update(  # pylint: disable=too-many-branches
         else:
             list_data = data
 
+        await self._validate_hierarchy()
+
         # Reset the list of relationship and save the previous one to see if we can reuse some
         previous_relationships = {rel.peer_id: rel for rel in await self.get_relationships(db=db) if rel.peer_id}
         self._relationships.clear()
@@ -1023,6 +1027,8 @@ async def add(self, data: Union[dict[str, Any], Node], db: InfrahubDatabase) ->
         if not isinstance(data, (self.rel_class, dict)) and not hasattr(data, "_schema"):
             raise ValidationError({self.name: f"Invalid data provided to form a relationship {data}"})
 
+        await self._validate_hierarchy()
+
         previous_relationships = {rel.peer_id for rel in await self.get_relationships(db=db) if rel.peer_id}
 
         item_id = getattr(data, "id", None)
@@ -1148,3 +1154,14 @@ async def to_graphql(
             return None
 
         return await relationships[0].to_graphql(fields=fields, db=db, related_node_ids=related_node_ids)
+
+    async def _validate_hierarchy(self) -> None:
+        schema = self.node.get_schema()
+        if schema.is_profile_schema or not schema.hierarchy:  # type: ignore[union-attr]
+            return
+
+        if self.name == "parent" and not schema.parent:  # type: ignore[union-attr]
+            raise ValidationError({self.name: f"Not supported to assign a value to parent for {schema.kind}"})
+
+        if self.name == "children" and not schema.children:  # type: ignore[union-attr]
+            raise ValidationError({self.name: f"Not supported to assign some children for {schema.kind}"})

backend/tests/unit/conftest.py

@@ -35,6 +35,7 @@
 from infrahub.core.node.resource_manager.ip_address_pool import CoreIPAddressPool
 from infrahub.core.node.resource_manager.ip_prefix_pool import CoreIPPrefixPool
 from infrahub.core.protocols_base import CoreNode
+from infrahub.core.relationship import RelationshipManager
 from infrahub.core.schema import (
     GenericSchema,
     NodeSchema,
@@ -2062,6 +2063,42 @@ async def hierarchical_location_schema_simple(db: InfrahubDatabase, default_bran
     registry.schema.register_schema(schema=schema, branch=default_branch.name)
 
 
+@pytest.fixture
+async def location_generic_protocol():
+    class LocationGeneric(CoreNode):
+        name: String
+        status: StringOptional
+        things: RelationshipManager
+        parent: RelationshipManager
+        children: RelationshipManager
+
+    return LocationGeneric
+
+
+@pytest.fixture
+async def location_site_protocol(location_generic_protocol):
+    class LocationSite(location_generic_protocol):
+        pass
+
+    return LocationSite
+
+
+@pytest.fixture
+async def location_region_protocol(location_generic_protocol):
+    class LocationRegion(location_generic_protocol):
+        pass
+
+    return LocationRegion
+
+
+@pytest.fixture
+async def location_rack_protocol(location_generic_protocol):
+    class LocationRack(location_generic_protocol):
+        pass
+
+    return LocationRack
+
+
 @pytest.fixture
 async def hierarchical_location_schema(
     db: InfrahubDatabase, default_branch: Branch, hierarchical_location_schema_simple, register_core_models_schema

backend/tests/unit/core/hierarchy/test_hierarchy_create.py

@@ -0,0 +1,24 @@
+import pytest
+
+from infrahub.core import registry
+from infrahub.core.manager import NodeManager
+from infrahub.core.node import Node
+from infrahub.database import InfrahubDatabase
+from infrahub.exceptions import ValidationError
+
+
+async def test_create_node_with_invalid_hierarchy(db: InfrahubDatabase, hierarchical_location_data):
+    region_schema = registry.schema.get_node_schema(name="LocationRegion", duplicate=True)
+    rack_schema = registry.schema.get_node_schema(name="LocationRack", duplicate=True)
+    europe = await NodeManager.get_one(db=db, id=hierarchical_location_data["europe"].id)
+    city = await NodeManager.get_one(db=db, id=hierarchical_location_data["seattle"].id)
+
+    with pytest.raises(ValidationError) as exc:
+        region = await Node.init(db=db, schema=region_schema)
+        await region.new(db=db, name="region2", parent=city)
+    assert "Not supported to assign a value to parent" in str(exc.value)
+
+    with pytest.raises(ValidationError) as exc:
+        rack = await Node.init(db=db, schema=rack_schema)
+        await rack.new(db=db, name="rack2", parent=city, children=[europe])
+    assert "Not supported to assign some children" in str(exc.value)

backend/tests/unit/core/hierarchy/test_hierarchy_update.py

@@ -1,6 +1,9 @@
+import pytest
+
 from infrahub.core import registry
 from infrahub.core.manager import NodeManager
 from infrahub.database import InfrahubDatabase
+from infrahub.exceptions import ValidationError
 
 CHECK_HIERARCHY_QUERY = """
 MATCH ({uuid: $node_uuid})-[rel:IS_RELATED]-(rel_node:Relationship {name: "parent__child"})
@@ -30,3 +33,25 @@ async def test_update_node_with_hierarchy(db: InfrahubDatabase, hierarchical_loc
         assert result.get("rel").get("hierarchy") == site_schema.hierarchy
     nodes = await NodeManager.query(db=db, schema=site_schema, filters={"parent__name__value": "europe"})
     assert {node.name.value for node in nodes} == {"paris", "london", "seattle"}
+
+
+async def test_update_node_invalid_hierarchy(db: InfrahubDatabase, hierarchical_location_data):
+    city = await NodeManager.get_one(db=db, id=hierarchical_location_data["seattle"].id, raise_on_error=True)
+    region = await NodeManager.get_one(db=db, id=hierarchical_location_data["europe"].id, raise_on_error=True)
+    rack = await NodeManager.get_one(db=db, id=hierarchical_location_data["paris-r1"].id, raise_on_error=True)
+
+    with pytest.raises(ValidationError) as exc:
+        await region.parent.update(db=db, data=city)
+    assert "Not supported to assign a value to parent" in str(exc.value)
+
+    with pytest.raises(ValidationError) as exc:
+        await region.parent.add(db=db, data=city)
+    assert "Not supported to assign a value to parent" in str(exc.value)
+
+    with pytest.raises(ValidationError) as exc:
+        await rack.children.update(db=db, data=[region])
+    assert "Not supported to assign some children" in str(exc.value)
+
+    with pytest.raises(ValidationError) as exc:
+        await rack.children.update(db=db, data=[region])
+    assert "Not supported to assign some children" in str(exc.value)

changelog/4325.fixed.md

@@ -0,0 +1 @@
+Hierarchical node that don't have a parent or a children defined in the schema will properly enforce that constraint