Nuke generate_template for generics

Author: gmazoyer

backend/infrahub/core/graph/__init__.py

@@ -1 +1 @@
-GRAPH_VERSION = 61
+GRAPH_VERSION = 62

backend/infrahub/core/migrations/graph/__init__.py

@@ -63,6 +63,7 @@
 from .m059_recompute_permission_display_labels import Migration059
 from .m060_template_number_pool_cleanup import Migration060
 from .m061_template_ip_pool_relationship_cleanup import Migration061
+from .m062_remove_generic_generate_template import Migration062
 
 if TYPE_CHECKING:
     from ..shared import MigrationTypes
@@ -130,6 +131,7 @@
     Migration059,
     Migration060,
     Migration061,
+    Migration062,
 ]
 
 

backend/infrahub/core/migrations/graph/m062_remove_generic_generate_template.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any, Sequence
+
+from infrahub.core.migrations.shared import MigrationResult
+from infrahub.core.query import Query, QueryType
+
+from ..shared import GraphMigration
+
+if TYPE_CHECKING:
+    from infrahub.database import InfrahubDatabase
+
+
+class RemoveGenericGenerateTemplateQuery(Query):
+    """Remove generate_template Attribute nodes from all SchemaGeneric-labeled instances."""
+
+    name = "migration_062_remove_generic_generate_template"
+    type: QueryType = QueryType.WRITE
+    insert_return = False
+
+    async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> None:  # noqa: ARG002
+        query = """
+        MATCH (sg:SchemaGeneric)-[:HAS_ATTRIBUTE]->(attr:Attribute {name: "generate_template"})
+        DETACH DELETE attr
+        """
+        self.add_to_query(query)
+
+
+class RemoveGenericGenerateTemplateSchemaAttributeQuery(Query):
+    """Remove the generate_template SchemaAttribute from the SchemaGeneric type definition."""
+
+    name = "migration_062_remove_generic_generate_template_schema_attribute"
+    type: QueryType = QueryType.WRITE
+    insert_return = False
+
+    async def query_init(self, db: InfrahubDatabase, **kwargs: dict[str, Any]) -> None:  # noqa: ARG002
+        query = """
+// Find the SchemaGeneric type definition (stored as SchemaNode)
+MATCH p1 = (sn:SchemaNode)-[:HAS_ATTRIBUTE]->(:Attribute {name: "name"})-[:HAS_VALUE]->(:AttributeValueIndexed {value: "Generic"})
+WHERE all(r IN relationships(p1) WHERE r.status = "active" AND r.to IS NULL)
+MATCH p2 = (sn)-[:HAS_ATTRIBUTE]->(:Attribute {name: "namespace"})-[:HAS_VALUE]->(:AttributeValueIndexed {value: "Schema"})
+WHERE all(r IN relationships(p2) WHERE r.status = "active" AND r.to IS NULL)
+WITH sn
+LIMIT 1
+// Find the generate_template SchemaAttribute child
+MATCH p3 = (sn)-[:IS_RELATED]-(rel:Relationship {name: "schema__node__attributes"})
+               -[:IS_RELATED]-(sa:SchemaAttribute)
+               -[:HAS_ATTRIBUTE]->(:Attribute {name: "name"})
+               -[:HAS_VALUE]->(:AttributeValueIndexed {value: "generate_template"})
+WHERE all(r IN relationships(p3) WHERE r.status = "active" AND r.to IS NULL)
+WITH sa, rel
+LIMIT 1
+
+// Delete the SchemaAttribute and its relationship node
+DETACH DELETE sa, rel
+        """
+        self.add_to_query(query)
+
+
+class Migration062(GraphMigration):
+    name: str = "062_remove_generic_generate_template"
+    minimum_version: int = 61
+    queries: Sequence[type[Query]] = [
+        RemoveGenericGenerateTemplateQuery,
+        RemoveGenericGenerateTemplateSchemaAttributeQuery,
+    ]
+
+    async def validate_migration(self, db: InfrahubDatabase) -> MigrationResult:  # noqa: ARG002
+        return MigrationResult()

backend/tests/component/core/migrations/graph/test_062_remove_generic_generate_template.py

@@ -0,0 +1,91 @@
+import pytest
+
+from infrahub.core import registry
+from infrahub.core.branch import Branch
+from infrahub.core.constants import SchemaPathType
+from infrahub.core.migrations.graph.m062_remove_generic_generate_template import Migration062
+from infrahub.core.migrations.schema.node_attribute_add import NodeAttributeAddMigration
+from infrahub.core.migrations.shared import InternalSchemaMigration, MigrationInput
+from infrahub.core.path import SchemaPath
+from infrahub.core.timestamp import Timestamp
+from infrahub.database import InfrahubDatabase
+
+COUNT_GENERATE_TEMPLATE_ON_GENERICS = """
+MATCH (sg:SchemaGeneric)-[r:HAS_ATTRIBUTE]->(attr:Attribute {name: "generate_template"})
+WHERE r.status = "active" AND r.to IS NULL
+RETURN count(attr) AS count
+"""
+
+COUNT_GENERATE_TEMPLATE_SCHEMA_ATTRIBUTE = """
+MATCH p1 = (sn:SchemaNode)-[:HAS_ATTRIBUTE]->(:Attribute {name: "name"})
+    -[:HAS_VALUE]->(:AttributeValueIndexed {value: "Generic"})
+WHERE all(r IN relationships(p1) WHERE r.status = "active" AND r.to IS NULL)
+MATCH p2 = (sn)-[:HAS_ATTRIBUTE]->(:Attribute {name: "namespace"})
+    -[:HAS_VALUE]->(:AttributeValueIndexed {value: "Schema"})
+WHERE all(r IN relationships(p2) WHERE r.status = "active" AND r.to IS NULL)
+WITH sn
+LIMIT 1
+MATCH p3 = (sn)-[:IS_RELATED]-(:Relationship {name: "schema__node__attributes"})
+    -[:IS_RELATED]-(sa:SchemaAttribute)
+    -[:HAS_ATTRIBUTE]->(:Attribute {name: "name"})
+    -[:HAS_VALUE]->(:AttributeValueIndexed {value: "generate_template"})
+WHERE all(r IN relationships(p3) WHERE r.status = "active" AND r.to IS NULL)
+RETURN count(sa) AS count
+"""
+
+
+@pytest.fixture
+async def migration_062_data(
+    db: InfrahubDatabase, reset_registry: None, default_branch: Branch, register_core_schema_db: None
+) -> None:
+    """Add generate_template attribute to SchemaGeneric nodes, simulating the old DB state."""
+    internal_schema_branch = InternalSchemaMigration.get_internal_schema()
+    schema_node = internal_schema_branch.get_node(name="SchemaNode")
+    schema_generic = internal_schema_branch.get_node(name="SchemaGeneric")
+
+    # Build a SchemaGeneric definition that includes generate_template
+    schema_generic_with_attr = internal_schema_branch.get_node(name="SchemaGeneric")
+    generate_template_attr = schema_node.get_attribute(name="generate_template")
+    schema_generic_with_attr.attributes.append(generate_template_attr)
+
+    add_migration = NodeAttributeAddMigration(
+        new_node_schema=schema_generic_with_attr,
+        previous_node_schema=schema_generic,
+        schema_path=SchemaPath(
+            schema_kind="SchemaGeneric", path_type=SchemaPathType.ATTRIBUTE, field_name="generate_template"
+        ),
+    )
+    await add_migration.execute(migration_input=MigrationInput(db=db), branch=default_branch)
+
+    # Also add generate_template SchemaAttribute to the SchemaGeneric type definition
+    # This simulates the old DB state where update_core_schema stored the attribute definition
+    schema_generic_def = registry.schema.get(name="SchemaGeneric", branch=default_branch, duplicate=True)
+    gt_attr = generate_template_attr.duplicate()
+    gt_attr.id = None
+    schema_generic_def.attributes.append(gt_attr)
+    await registry.schema.update_node_in_db(
+        node=schema_generic_def, branch=default_branch, db=db, at=Timestamp(), user_id="migration-test"
+    )
+
+
+async def test_migration_062(
+    db: InfrahubDatabase, reset_registry: None, default_branch: Branch, migration_062_data: None
+) -> None:
+    result_before = await db.execute_query(query=COUNT_GENERATE_TEMPLATE_ON_GENERICS)
+    assert result_before[0].get("count") > 0
+
+    schema_attr_before = await db.execute_query(query=COUNT_GENERATE_TEMPLATE_SCHEMA_ATTRIBUTE)
+    assert schema_attr_before[0].get("count") > 0
+
+    migration = Migration062.init()
+    execution_result = await migration.execute(migration_input=MigrationInput(db=db))
+    assert not execution_result.errors
+
+    validation_result = await migration.validate_migration(db=db)
+    assert not validation_result.errors
+
+    result_after = await db.execute_query(query=COUNT_GENERATE_TEMPLATE_ON_GENERICS)
+    assert result_after[0].get("count") == 0
+
+    schema_attr_after = await db.execute_query(query=COUNT_GENERATE_TEMPLATE_SCHEMA_ATTRIBUTE)
+    assert schema_attr_after[0].get("count") == 0