diff --git a/infrahub_sdk/node/node.py b/infrahub_sdk/node/node.py index a47209dc..599816e6 100644 --- a/infrahub_sdk/node/node.py +++ b/infrahub_sdk/node/node.py @@ -1113,11 +1113,17 @@ async def _process_mutation_result( self.id = object_response["id"] self._existing = True + is_template = isinstance(self._schema, TemplateSchemaAPI) + for attr_name in self._attributes: attr = getattr(self, attr_name) if attr_name not in object_response or not attr.is_from_pool_attribute(): continue + # Templates store pool references without allocating values + if is_template: + continue + # Process allocated resource from a pool and update attribute attr.value = object_response[attr_name]["value"] @@ -1126,6 +1132,10 @@ async def _process_mutation_result( if rel_name not in object_response or not isinstance(rel, RelatedNode) or not rel.is_resource_pool: continue + # Templates store pool references without allocating + if is_template: + continue + # Process allocated resource from a pool and update related node allocated_resource = object_response[rel_name] related_node = RelatedNode( @@ -1996,11 +2006,17 @@ def _process_mutation_result( self.id = object_response["id"] self._existing = True + is_template = isinstance(self._schema, TemplateSchemaAPI) + for attr_name in self._attributes: attr = getattr(self, attr_name) if attr_name not in object_response or not attr.is_from_pool_attribute(): continue + # Templates store pool references without allocating values + if is_template: + continue + # Process allocated resource from a pool and update attribute attr.value = object_response[attr_name]["value"] @@ -2009,6 +2025,10 @@ def _process_mutation_result( if rel_name not in object_response or not isinstance(rel, RelatedNodeSync) or not rel.is_resource_pool: continue + # Templates store pool references without allocating + if is_template: + continue + # Process allocated resource from a pool and update related node allocated_resource = object_response[rel_name] related_node = RelatedNodeSync( diff --git a/tests/unit/sdk/pool/test_template_skip_pool_allocation.py b/tests/unit/sdk/pool/test_template_skip_pool_allocation.py new file mode 100644 index 00000000..dd02b55f --- /dev/null +++ b/tests/unit/sdk/pool/test_template_skip_pool_allocation.py @@ -0,0 +1,216 @@ +"""Templates store pool references without allocating values. + +When a template node is saved, _process_mutation_result should NOT overwrite +pool-sourced attributes or relationships with server-allocated values. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +import pytest + +from infrahub_sdk.node import InfrahubNode, InfrahubNodeSync +from infrahub_sdk.schema import AttributeSchemaAPI, RelationshipSchemaAPI, TemplateSchemaAPI +from infrahub_sdk.schema.main import AttributeKind, RelationshipKind + +if TYPE_CHECKING: + from infrahub_sdk import InfrahubClient, InfrahubClientSync + from infrahub_sdk.schema import NodeSchemaAPI + +POOL_ID = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" +NODE_ID = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" + +VLAN_MUTATION = "TemplateIpamVLANCreate" +DEVICE_MUTATION = "TemplateInfraDeviceCreate" + + +class TestTemplateAttributeSkipsPoolAllocation: + async def test_async_template_preserves_pool_reference_on_attribute( + self, + client: InfrahubClient, + vlan_template_schema: TemplateSchemaAPI, + ) -> None: + """After save, a template's from_pool attribute should keep the pool reference, not the allocated value.""" + node = InfrahubNode(client=client, schema=vlan_template_schema, data=_vlan_data_with_pool()) + + # Act + await node._process_mutation_result(VLAN_MUTATION, _vlan_mutation_response()) + + assert node.vlan_id.value != 42 + assert node.id == NODE_ID + + async def test_sync_template_preserves_pool_reference_on_attribute( + self, + client_sync: InfrahubClientSync, + vlan_template_schema: TemplateSchemaAPI, + ) -> None: + """After save, a template's from_pool attribute should keep the pool reference, not the allocated value.""" + node = InfrahubNodeSync(client=client_sync, schema=vlan_template_schema, data=_vlan_data_with_pool()) + + # Act + node._process_mutation_result(VLAN_MUTATION, _vlan_mutation_response()) + + assert node.vlan_id.value != 42 + assert node.id == NODE_ID + + async def test_async_regular_node_does_allocate_from_pool( + self, + client: InfrahubClient, + vlan_schema: NodeSchemaAPI, + ) -> None: + """Contrast: a regular node SHOULD update its attribute with the allocated value.""" + node = InfrahubNode(client=client, schema=vlan_schema, data=_vlan_data_with_pool()) + + # Act + await node._process_mutation_result("InfraVLANCreate", _node_vlan_mutation_response()) + + assert node.vlan_id.value == 42 + assert node.id == NODE_ID + + async def test_sync_regular_node_does_allocate_from_pool( + self, + client_sync: InfrahubClientSync, + vlan_schema: NodeSchemaAPI, + ) -> None: + """Contrast: a regular node SHOULD update its attribute with the allocated value.""" + node = InfrahubNodeSync(client=client_sync, schema=vlan_schema, data=_vlan_data_with_pool()) + + # Act + node._process_mutation_result("InfraVLANCreate", _node_vlan_mutation_response()) + + assert node.vlan_id.value == 42 + assert node.id == NODE_ID + + +class TestTemplateRelationshipSkipsPoolAllocation: + async def test_async_template_preserves_pool_reference_on_relationship( + self, + client: InfrahubClient, + device_template_schema: TemplateSchemaAPI, + ipaddress_pool_schema: NodeSchemaAPI, + ipam_ipprefix_schema: NodeSchemaAPI, + ipam_ipprefix_data: dict[str, Any], + ) -> None: + """After save, a template's pool relationship should not be replaced with the allocated resource.""" + ip_prefix = InfrahubNode(client=client, schema=ipam_ipprefix_schema, data=ipam_ipprefix_data) + ip_pool = InfrahubNode(client=client, schema=ipaddress_pool_schema, data=_pool_node_data(ip_prefix)) + node = InfrahubNode( + client=client, schema=device_template_schema, data={"name": "device-template", "primary_address": ip_pool} + ) + original_rel = node.primary_address + + # Act + await node._process_mutation_result(DEVICE_MUTATION, _device_mutation_response()) + + assert node.primary_address is original_rel + assert node.id == NODE_ID + + async def test_sync_template_preserves_pool_reference_on_relationship( + self, + client_sync: InfrahubClientSync, + device_template_schema: TemplateSchemaAPI, + ipaddress_pool_schema: NodeSchemaAPI, + ipam_ipprefix_schema: NodeSchemaAPI, + ipam_ipprefix_data: dict[str, Any], + ) -> None: + """After save, a template's pool relationship should not be replaced with the allocated resource.""" + ip_prefix = InfrahubNodeSync(client=client_sync, schema=ipam_ipprefix_schema, data=ipam_ipprefix_data) + ip_pool = InfrahubNodeSync(client=client_sync, schema=ipaddress_pool_schema, data=_pool_node_data(ip_prefix)) + node = InfrahubNodeSync( + client=client_sync, + schema=device_template_schema, + data={"name": "device-template", "primary_address": ip_pool}, + ) + original_rel = node.primary_address + + # Act + node._process_mutation_result(DEVICE_MUTATION, _device_mutation_response()) + + assert node.primary_address is original_rel + assert node.id == NODE_ID + + +# ────────────────────────────────────────────── +# Helpers +# ────────────────────────────────────────────── + + +def _vlan_data_with_pool() -> dict[str, Any]: + return {"name": "Standard VLAN", "vlan_id": {"from_pool": {"id": POOL_ID, "identifier": "vlan-id"}}} + + +def _vlan_mutation_response() -> dict[str, Any]: + return {VLAN_MUTATION: {"ok": True, "object": {"id": NODE_ID, "vlan_id": {"value": 42}}}} + + +def _node_vlan_mutation_response() -> dict[str, Any]: + return {"InfraVLANCreate": {"ok": True, "object": {"id": NODE_ID, "vlan_id": {"value": 42}}}} + + +def _device_mutation_response() -> dict[str, Any]: + return { + DEVICE_MUTATION: { + "ok": True, + "object": { + "id": NODE_ID, + "primary_address": { + "node": { + "__typename": "IpamIPAddress", + "id": "cccccccc-cccc-cccc-cccc-cccccccccccc", + "display_label": "10.0.0.1/32", + }, + }, + }, + }, + } + + +def _pool_node_data(ip_prefix: InfrahubNode | InfrahubNodeSync) -> dict[str, Any]: + return { + "id": POOL_ID, + "name": "Core loopbacks", + "default_address_type": "IpamIPAddress", + "default_prefix_length": 32, + "ip_namespace": "ip_namespace", + "resources": [ip_prefix], + } + + +# ────────────────────────────────────────────── +# Fixtures +# ────────────────────────────────────────────── + + +@pytest.fixture +async def vlan_template_schema() -> TemplateSchemaAPI: + return TemplateSchemaAPI( + name="VLAN", + namespace="TemplateIpam", + attributes=[ + AttributeSchemaAPI(name="name", kind=AttributeKind.TEXT, unique=True), + AttributeSchemaAPI(name="vlan_id", kind=AttributeKind.NUMBER), + ], + relationships=[], + ) + + +@pytest.fixture +async def device_template_schema() -> TemplateSchemaAPI: + return TemplateSchemaAPI( + name="Device", + namespace="TemplateInfra", + attributes=[ + AttributeSchemaAPI(name="name", kind=AttributeKind.TEXT, unique=True), + ], + relationships=[ + RelationshipSchemaAPI( + name="primary_address", + peer="IpamIPAddress", + label="Primary IP Address", + optional=True, + cardinality="one", + kind=RelationshipKind.ATTRIBUTE, + ), + ], + )