Add integartion tests for metadata

Author: ogenstad

tests/integration/test_infrahub_client.py

@@ -11,6 +11,7 @@
 from infrahub_sdk.constants import InfrahubClientMode
 from infrahub_sdk.exceptions import BranchNotFoundError, URLNotFoundError
 from infrahub_sdk.node import InfrahubNode
+from infrahub_sdk.node.metadata import NodeMetadata, RelationshipMetadata
 from infrahub_sdk.playback import JSONPlayback
 from infrahub_sdk.recorder import JSONRecorder
 from infrahub_sdk.schema import GenericSchema, NodeSchema, ProfileSchemaAPI
@@ -328,6 +329,89 @@ async def create_person_with_tags(clt: InfrahubClient, nbr_tags: int) -> None:
         )
         assert len(group.members.peers) == 2
 
+    async def test_node_metadata_not_fetched_by_default(
+        self, client: InfrahubClient, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = await client.get(kind=TESTING_CAT, id=cat_luna.id)
+        assert node.get_node_metadata() is None
+
+    async def test_node_metadata_with_get(
+        self, client: InfrahubClient, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = await client.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        metadata = node.get_node_metadata()
+        assert isinstance(metadata, NodeMetadata)
+        assert metadata.created_at is not None
+        assert metadata.updated_at is not None
+        assert metadata.created_by.display_label == "Admin"
+
+    async def test_node_metadata_with_all(self, client: InfrahubClient, base_dataset: None) -> None:
+        nodes = await client.all(kind=TESTING_CAT, include_metadata=True)
+        assert nodes
+
+        for node in nodes:
+            metadata = node.get_node_metadata()
+            assert isinstance(metadata, NodeMetadata)
+            assert metadata.created_at is not None
+            assert metadata.updated_at is not None
+            assert metadata.created_by.display_label == "Admin"
+
+    async def test_attribute_metadata(self, client: InfrahubClient, base_dataset: None, cat_luna: InfrahubNode) -> None:
+        node = await client.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        assert node.name.updated_by.display_label == "Admin"
+        assert node.breed.updated_by.display_label == "Admin"
+        original_name_updated_at = node.name.updated_at
+        original_breed_updated_at = node.breed.updated_at
+
+        node.name.value = "Luna Updated"
+        await node.save()
+
+        assert original_name_updated_at is not None
+
+        node_after = await client.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+        assert node_after.name.value == "Luna Updated"
+        assert node_after.name.updated_by.display_label == "Admin"
+        assert node_after.name.updated_at is not None
+        assert node_after.name.updated_at > original_name_updated_at
+        assert node_after.breed.updated_at == original_breed_updated_at
+
+    async def test_relationship_metadata_cardinality_one(
+        self, client: InfrahubClient, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = await client.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        rel_metadata = node.owner.get_relationship_metadata()
+        assert isinstance(rel_metadata, RelationshipMetadata)
+        assert rel_metadata.updated_at is not None
+        assert rel_metadata.updated_by.display_label == "Admin"
+
+    async def test_relationship_metadata_cardinality_many(
+        self, client: InfrahubClient, base_dataset: None, person_ethan: InfrahubNode
+    ) -> None:
+        # Use include=["animals"] rather than prefetch_relationships=True.
+        # prefetch_relationships=True recursively fetches each animal's peer relationships,
+        # which in turn include person_ethan's null favorite_animal cardinality-one relationship.
+        # The server GraphQL schema marks NestedEdgedTestingAnimal.node_metadata as non-nullable,
+        # so requesting include_metadata=True on that null edge causes the server to silently
+        # return empty edges. include=["animals"] fetches only the animals relationship shallowly.
+        # Also exclude favorite_animal (null cardinality-one inbound) for the same reason.
+        node = await client.get(
+            kind=TESTING_PERSON,
+            id=person_ethan.id,
+            include_metadata=True,
+            include=["animals"],
+            exclude=["favorite_animal"],
+        )
+
+        assert node.animals.peers
+        for peer in node.animals.peers:
+            rel_metadata = peer.get_relationship_metadata()
+            assert isinstance(rel_metadata, RelationshipMetadata)
+            assert rel_metadata.updated_at is not None
+            assert rel_metadata.updated_by.display_label == "Admin"
+
     @pytest.mark.xfail(reason="https://github.com/opsmill/infrahub-sdk-python/issues/733")
     async def test_recorder_with_playback_rewrite_host(
         self, base_dataset: None, tmp_path: Path, infrahub_port: int

tests/integration/test_infrahub_client_sync.py

@@ -10,6 +10,7 @@
 from infrahub_sdk.constants import InfrahubClientMode
 from infrahub_sdk.exceptions import BranchNotFoundError, URLNotFoundError
 from infrahub_sdk.node import InfrahubNodeSync
+from infrahub_sdk.node.metadata import NodeMetadata, RelationshipMetadata
 from infrahub_sdk.playback import JSONPlayback
 from infrahub_sdk.recorder import JSONRecorder
 from infrahub_sdk.schema import GenericSchema, NodeSchema, ProfileSchemaAPI
@@ -330,6 +331,74 @@ def create_person_with_tags(clt: InfrahubClientSync, nbr_tags: int) -> None:
         )
         assert len(group.members.peers) == 2
 
+    def test_node_metadata_not_fetched_by_default(
+        self, client_sync: InfrahubClientSync, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = client_sync.get(kind=TESTING_CAT, id=cat_luna.id)
+        assert node.get_node_metadata() is None
+
+    def test_node_metadata_with_get(
+        self, client_sync: InfrahubClientSync, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = client_sync.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        metadata = node.get_node_metadata()
+        assert isinstance(metadata, NodeMetadata)
+        assert metadata.created_at is not None
+        assert metadata.updated_at is not None
+        assert metadata.created_by.display_label == "Admin"
+
+    def test_attribute_metadata(
+        self, client_sync: InfrahubClientSync, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = client_sync.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        assert node.name.updated_by.display_label == "Admin"
+        assert node.breed.updated_by.display_label == "Admin"
+        original_name_updated_at = node.name.updated_at
+        original_breed_updated_at = node.breed.updated_at
+        assert original_name_updated_at is not None
+
+        node.name.value = "Luna Updated"
+        node.save()
+
+        node_after = client_sync.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+        assert node_after.name.value == "Luna Updated"
+        assert node_after.name.updated_by.display_label == "Admin"
+        assert node_after.name.updated_at is not None
+        assert node_after.name.updated_at > original_name_updated_at
+        assert node_after.breed.updated_at == original_breed_updated_at
+
+    def test_relationship_metadata_cardinality_one(
+        self, client_sync: InfrahubClientSync, base_dataset: None, cat_luna: InfrahubNode
+    ) -> None:
+        node = client_sync.get(kind=TESTING_CAT, id=cat_luna.id, include_metadata=True)
+
+        rel_metadata = node.owner.get_relationship_metadata()
+        assert isinstance(rel_metadata, RelationshipMetadata)
+        assert rel_metadata.updated_at is not None
+        assert rel_metadata.updated_by.display_label == "Admin"
+
+    def test_relationship_metadata_cardinality_many(
+        self, client_sync: InfrahubClientSync, base_dataset: None, person_ethan: InfrahubNode
+    ) -> None:
+        # Use include=["animals"] rather than prefetch_relationships=True — see async counterpart
+        # in test_infrahub_client.py for full explanation.
+        node = client_sync.get(
+            kind=TESTING_PERSON,
+            id=person_ethan.id,
+            include_metadata=True,
+            include=["animals"],
+            exclude=["favorite_animal"],
+        )
+
+        assert node.animals.peers
+        for peer in node.animals.peers:
+            rel_metadata = peer.get_relationship_metadata()
+            assert isinstance(rel_metadata, RelationshipMetadata)
+            assert rel_metadata.updated_at is not None
+            assert rel_metadata.updated_by.display_label == "Admin"
+
     @pytest.mark.xfail(reason="https://github.com/opsmill/infrahub-sdk-python/issues/733")
     def test_recorder_with_playback_rewrite_host(self, base_dataset: None, tmp_path: Path, infrahub_port: int) -> None:
         # Create a fresh client for recording to ensure clean state (no cached schema)