IFC-15 Support for mandatory/optional parent relationship

backend/infrahub/core/schema/schema_branch.py

@@ -492,6 +492,8 @@ def process_pre_validation(self) -> None:
         self.process_branch_support()
         self.manage_profile_schemas()
         self.manage_profile_relationships()
+        self.add_hierarchy_generic()
+        self.add_hierarchy_node()
 
     def process_validate(self) -> None:
         self.validate_names()
@@ -512,8 +514,6 @@ def process_validate(self) -> None:
     def process_post_validation(self) -> None:
         self.cleanup_inherited_elements()
         self.add_groups()
-        self.add_hierarchy_generic()
-        self.add_hierarchy_node()
         self.generate_weight()
         self.process_labels()
         self.process_dropdowns()
@@ -633,7 +633,7 @@ def validate_schema_path(
                     and not (
                         schema_attribute_path.relationship_schema.name == "ip_namespace"
                         and isinstance(node_schema, NodeSchema)
-                        and (node_schema.is_ip_address() or node_schema.is_ip_prefix)
+                        and (node_schema.is_ip_address() or node_schema.is_ip_prefix())
                     )
                 ):
                     raise ValueError(
@@ -1509,7 +1509,7 @@ def add_groups(self) -> None:
             if changed:
                 self.set(name=node_name, schema=schema)
 
-    def _get_hierarchy_child_rel(self, peer: str, hierarchical: str, read_only: bool) -> RelationshipSchema:
+    def _get_hierarchy_child_rel(self, peer: str, hierarchical: str | None, read_only: bool) -> RelationshipSchema:
         return RelationshipSchema(
             name="children",
             identifier="parent__child",
@@ -1522,18 +1522,22 @@ def _get_hierarchy_child_rel(self, peer: str, hierarchical: str, read_only: bool
             read_only=read_only,
         )
 
-    def _get_hierarchy_parent_rel(self, peer: str, hierarchical: str, read_only: bool) -> RelationshipSchema:
+    def _get_hierarchy_parent_rel(
+        self, peer: str, hierarchical: str | None, read_only: bool, optional: bool
+    ) -> RelationshipSchema:
         return RelationshipSchema(
             name="parent",
             identifier="parent__child",
             peer=peer,
             kind=RelationshipKind.HIERARCHY,
             cardinality=RelationshipCardinality.ONE,
+            min_count=0 if optional else 1,
             max_count=1,
             branch=BranchSupportType.AWARE,
             direction=RelationshipDirection.OUTBOUND,
             hierarchical=hierarchical,
             read_only=read_only,
+            optional=optional,
         )
 
     def add_hierarchy_generic(self) -> None:
@@ -1548,7 +1552,9 @@ def add_hierarchy_generic(self) -> None:
 
             if "parent" not in generic.relationship_names:
                 generic.relationships.append(
-                    self._get_hierarchy_parent_rel(peer=generic_name, hierarchical=generic_name, read_only=read_only)
+                    self._get_hierarchy_parent_rel(
+                        peer=generic_name, hierarchical=generic_name, read_only=read_only, optional=True
+                    )
                 )
             if "children" not in generic.relationship_names:
                 generic.relationships.append(
@@ -1571,7 +1577,10 @@ def add_hierarchy_node(self) -> None:
                 if "parent" not in node.relationship_names:
                     node.relationships.append(
                         self._get_hierarchy_parent_rel(
-                            peer=node.parent, hierarchical=node.hierarchy, read_only=read_only
+                            peer=node.parent,
+                            hierarchical=node.hierarchy,
+                            read_only=read_only,
+                            optional=node.parent in [node_name] + self.generic_names,
                         )
                     )
                 else:

backend/tests/test_data/dataset01.py

@@ -95,8 +95,14 @@ async def load_data(db: InfrahubDatabase, nbr_devices: int = 0) -> None:
     # roles_dict = {}
 
     log.info("Creating Site")
+    continent = await Node.init(db=db, schema="LocationContinent")
+    await continent.new(db=db, name="Africa")
+    await continent.save(db=db)
+    country = await Node.init(db=db, schema="LocationCountry")
+    await country.new(db=db, name="Kingdom of Wakanda", parent=continent)
+    await country.save(db=db)
     site_hq = await Node.init(db=db, schema="LocationSite")
-    await site_hq.new(db=db, name="HQ")
+    await site_hq.new(db=db, name="HQ", parent=country)
     await site_hq.save(db=db)
 
     active_status = "active"

backend/tests/unit/core/constraint_validators/test_node_uniqueness.py

@@ -1,8 +1,11 @@
 import pytest
 
+from infrahub.core import registry
 from infrahub.core.branch import Branch
 from infrahub.core.node import Node
 from infrahub.core.node.constraints.attribute_uniqueness import NodeAttributeUniquenessConstraint
+from infrahub.core.node.constraints.grouped_uniqueness import NodeGroupedUniquenessConstraint
+from infrahub.core.schema import SchemaRoot
 from infrahub.database import InfrahubDatabase
 from infrahub.exceptions import ValidationError
 
@@ -29,3 +32,42 @@ async def test_node_validate_constraint_node_uniqueness_success(
     await alfred.new(db=db, name="Alfred", height=160)
 
     await constraint.check(alfred)
+
+
+async def test_hierarchical_uniqueness_constraint(
+    db: InfrahubDatabase, default_branch: Branch, hierarchical_location_schema_simple_unregistered: SchemaRoot
+):
+    site_schema = hierarchical_location_schema_simple_unregistered.get(name="LocationSite")
+    site_schema.human_friendly_id = ["parent__name__value", "name__value"]
+    site_schema.uniqueness_constraints = [["parent", "name__value"]]
+
+    rack_schema = hierarchical_location_schema_simple_unregistered.get(name="LocationRack")
+    rack_schema.human_friendly_id = ["parent__name__value", "name__value"]
+    rack_schema.uniqueness_constraints = [["parent", "name__value"]]
+
+    registry.schema.register_schema(schema=hierarchical_location_schema_simple_unregistered, branch=default_branch.name)
+    constraint = NodeGroupedUniquenessConstraint(db=db, branch=default_branch)
+
+    eu = await Node.init(db=db, schema="LocationRegion", branch=default_branch)
+    await eu.new(db=db, name="Europe")
+    await eu.save(db=db)
+    fr = await Node.init(db=db, schema="LocationSite", branch=default_branch)
+    await fr.new(db=db, name="France", parent=eu)
+    await fr.save(db=db)
+    uk = await Node.init(db=db, schema="LocationSite", branch=default_branch)
+    await uk.new(db=db, name="United Kingdom", parent=eu)
+    await fr.save(db=db)
+
+    th2 = await Node.init(db=db, schema="LocationRack", branch=default_branch)
+    await th2.new(db=db, name="th2-par", parent=fr)
+    await th2.save(db=db)
+
+    ld6 = await Node.init(db=db, schema="LocationRack", branch=default_branch)
+    await ld6.new(db=db, name="ld6-ldn", parent=uk)
+    await ld6.save(db=db)
+    await constraint.check(ld6)
+
+    ld6 = await Node.init(db=db, schema="LocationRack", branch=default_branch)
+    await ld6.new(db=db, name="ld6-ldn", parent=uk)
+    with pytest.raises(ValidationError, match=r"Violates uniqueness constraint 'parent-name' at parent"):
+        await constraint.check(ld6)

backend/tests/unit/core/schema_manager/test_manager_schema.py

@@ -17,6 +17,7 @@
     RelationshipKind,
     SchemaPathType,
 )
+from infrahub.core.node import Node
 from infrahub.core.schema import (
     AttributeSchema,
     GenericSchema,
@@ -2793,3 +2794,38 @@ async def test_process_deprecations(organization_schema):
     assert test_criticality.get_attribute(name="description").optional
     assert test_criticality.get_relationship(name="first").optional
     assert not test_criticality.get_relationship(name="second").optional
+
+
+async def test_hierarchical_validate_parent_children(
+    db: InfrahubDatabase, default_branch: Branch, hierarchical_location_schema_simple_unregistered: SchemaRoot
+):
+    site_schema = hierarchical_location_schema_simple_unregistered.get(name="LocationSite")
+    site_schema.human_friendly_id = ["parent__name__value", "name__value"]
+    site_schema.uniqueness_constraints = [["parent", "name__value"]]
+
+    registry.schema.register_schema(schema=hierarchical_location_schema_simple_unregistered, branch=default_branch.name)
+
+    schema_branch = registry.schema.get_schema_branch(name=default_branch.name)
+
+    with pytest.raises(ValueError, match=r"Unable to find the relationship"):
+        region_schema = schema_branch.get(name="LocationRegion", duplicate=False)
+        region_schema.get_relationship(name="parent")
+
+    with pytest.raises(ValueError, match=r"Unable to find the relationship"):
+        rack_schema = schema_branch.get(name="LocationRack", duplicate=False)
+        rack_schema.get_relationship(name="children")
+
+    eu: Node = await Node.init(db=db, schema="LocationRegion", branch=default_branch)
+    await eu.new(db=db, name="Europe")
+    await eu.save(db=db)
+
+    fr: Node = await Node.init(db=db, schema="LocationSite", branch=default_branch)
+    await fr.new(db=db, name="France", parent=eu)
+    await fr.save(db=db)
+
+    uk: Node = await Node.init(db=db, schema="LocationSite", branch=default_branch)
+    with pytest.raises(ValidationError, match=r"parent is mandatory"):
+        await uk.new(db=db, name="United Kingdom")
+
+    await uk.new(db=db, name="United Kingdom", parent=eu)
+    await uk.save(db=db)

changelog/3682.fixed.md

@@ -0,0 +1 @@
+Automatically mark hierarchical nodes `parent` relationship as optional if the parent is of the same kind or mandatory if the parent is of a different kind