Backport fix for inherited attribute / relationships not being updated (

#4059)

* Fix for inherited attribute / relationships not being updated (#4005)

---------

Co-authored-by: Damien Garros <[email protected]>

backend/infrahub/core/schema/attribute_schema.py

@@ -81,6 +81,14 @@ def convert_enum_to_value(self, data: Any) -> Any:
             return data.value
         return data
 
+    def update_from_generic(self, other: AttributeSchema) -> None:
+        fields_to_exclude = ("id", "order_weight", "branch", "inherited")
+        for name in self.model_fields:
+            if name in fields_to_exclude:
+                continue
+            if getattr(self, name) != getattr(other, name):
+                setattr(self, name, getattr(other, name))
+
     async def get_query_filter(
         self,
         name: str,

backend/infrahub/core/schema/basenode_schema.py

@@ -298,6 +298,10 @@ def get_filter_name_id_map(self) -> dict[str, str]:
     def valid_input_names(self) -> list[str]:
         return self.attribute_names + self.relationship_names + NODE_METADATA_ATTRIBUTES
 
+    @property
+    def valid_local_names(self) -> list[str]:
+        return self.local_attribute_names + self.local_relationship_names + NODE_METADATA_ATTRIBUTES
+
     @property
     def attribute_names(self) -> list[str]:
         return [item.name for item in self.attributes]
@@ -326,10 +330,18 @@ def mandatory_relationship_names(self) -> list[str]:
     def local_attributes(self) -> list[AttributeSchema]:
         return [item for item in self.attributes if not item.inherited]
 
+    @property
+    def local_attribute_names(self) -> list[str]:
+        return [item.name for item in self.local_attributes]
+
     @property
     def local_relationships(self) -> list[RelationshipSchema]:
         return [item for item in self.relationships if not item.inherited]
 
+    @property
+    def local_relationship_names(self) -> list[str]:
+        return [item.name for item in self.local_relationships]
+
     @property
     def unique_attributes(self) -> list[AttributeSchema]:
         return [item for item in self.attributes if item.unique]

backend/infrahub/core/schema/node_schema.py

@@ -60,8 +60,10 @@ def validate_inheritance(self, interface: GenericSchema) -> None:
                 )
 
     def inherit_from_interface(self, interface: GenericSchema) -> None:
-        existing_inherited_attributes = {item.name: idx for idx, item in enumerate(self.attributes) if item.inherited}
-        existing_inherited_relationships = {
+        existing_inherited_attributes: dict[str, int] = {
+            item.name: idx for idx, item in enumerate(self.attributes) if item.inherited
+        }
+        existing_inherited_relationships: dict[str, int] = {
             item.name: idx for idx, item in enumerate(self.relationships) if item.inherited
         }
         existing_inherited_fields = list(existing_inherited_attributes.keys()) + list(
@@ -82,7 +84,7 @@ def inherit_from_interface(self, interface: GenericSchema) -> None:
                 setattr(self, prop_name, getattr(interface, prop_name))
 
         for attribute in interface.attributes:
-            if attribute.name in self.valid_input_names:
+            if attribute.name in self.valid_local_names:
                 continue
 
             new_attribute = attribute.duplicate()
@@ -92,10 +94,10 @@ def inherit_from_interface(self, interface: GenericSchema) -> None:
                 self.attributes.append(new_attribute)
             else:
                 item_idx = existing_inherited_attributes[attribute.name]
-                self.attributes[item_idx] = new_attribute
+                self.attributes[item_idx].update_from_generic(other=new_attribute)
 
         for relationship in interface.relationships:
-            if relationship.name in self.valid_input_names:
+            if relationship.name in self.valid_local_names:
                 continue
 
             new_relationship = relationship.duplicate()
@@ -105,7 +107,7 @@ def inherit_from_interface(self, interface: GenericSchema) -> None:
                 self.relationships.append(new_relationship)
             else:
                 item_idx = existing_inherited_relationships[relationship.name]
-                self.relationships[item_idx] = new_relationship
+                self.relationships[item_idx].update_from_generic(other=new_relationship)
 
     def get_hierarchy_schema(self, db: InfrahubDatabase, branch: Optional[Union[Branch, str]] = None) -> GenericSchema:
         if not self.hierarchy:

backend/infrahub/core/schema/relationship_schema.py

@@ -55,6 +55,14 @@ def get_query_arrows(self) -> QueryArrows:
 
         return QueryArrows(left=QueryArrowOutband(), right=QueryArrowInband())
 
+    def update_from_generic(self, other: RelationshipSchema) -> None:
+        fields_to_exclude = ("id", "order_weight", "branch", "inherited", "filters")
+        for name in self.model_fields:
+            if name in fields_to_exclude:
+                continue
+            if getattr(self, name) != getattr(other, name):
+                setattr(self, name, getattr(other, name))
+
     async def get_query_filter(
         self,
         db: InfrahubDatabase,

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

@@ -163,6 +163,28 @@ async def test_schema_branch_process_inheritance_node_level(animal_person_schema
     assert dog.icon == animal.icon
 
 
+async def test_schema_branch_process_inheritance_update_inherited_elements(animal_person_schema_dict):
+    schema = SchemaBranch(cache={}, name="test")
+    schema.load_schema(schema=SchemaRoot(**animal_person_schema_dict))
+
+    schema.process_inheritance()
+
+    dog = schema.get(name="TestDog")
+    assert dog.get_attribute(name="name").description is None
+    assert dog.get_relationship(name="owner").optional is False
+
+    updated_schema = animal_person_schema_dict
+    updated_schema["generics"][0]["attributes"][0]["description"] = "new description"
+    updated_schema["generics"][0]["relationships"][0]["optional"] = True
+
+    schema.load_schema(schema=SchemaRoot(**updated_schema))
+    schema.process_inheritance()
+
+    dog = schema.get(name="TestDog")
+    assert dog.get_attribute(name="name").description == "new description"
+    assert dog.get_relationship(name="owner").optional is True
+
+
 async def test_schema_branch_process_humain_friendly_id(animal_person_schema_dict):
     schema = SchemaBranch(cache={}, name="test")
     schema.load_schema(schema=SchemaRoot(**animal_person_schema_dict))

changelog/4004.fixed.md

@@ -0,0 +1 @@
+Fix logic that prevented existing inherited attribute / relationships from being updated