[NEAT-126] Force deploy containers/views (#362)

* feat: force-deploy for all causes

* tests: wip new test

* tests: completed test'

* refactor: improved implementation of force functionality

* refactor: added shell

* tests: Added failing test

* feat: Implemented force deploy for containers

cognite/neat/utils/cdf_loaders/_data_modeling.py

@@ -1,5 +1,4 @@
-import re
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from graphlib import TopologicalSorter
 from typing import Any, Literal, cast
 
@@ -50,6 +49,27 @@ def in_space(self, item: T_WriteClass | T_WritableCogniteResource, space: set[st
     def sort_by_dependencies(self, items: list[T_WriteClass]) -> list[T_WriteClass]:
         return items
 
+    def _create_force(
+        self,
+        items: Sequence[T_WriteClass],
+        tried_force_deploy: set[T_ID],
+        create_method: Callable[[Sequence[T_WriteClass]], T_WritableCogniteResourceList],
+    ) -> T_WritableCogniteResourceList:
+        try:
+            return create_method(items)
+        except CogniteAPIError as e:
+            failed_items = {failed.as_id() for failed in e.failed if hasattr(failed, "as_id")}
+            to_redeploy = [
+                item for item in items if item.as_id() in failed_items and item.as_id() not in tried_force_deploy  # type: ignore[attr-defined]
+            ]
+            if not to_redeploy:
+                # Avoid infinite loop
+                raise e
+            ids = [item.as_id() for item in to_redeploy]  # type: ignore[attr-defined]
+            tried_force_deploy.update(ids)
+            self.delete(ids)
+            return self._create_force(to_redeploy, tried_force_deploy, create_method)
+
 
 class SpaceLoader(DataModelingLoader[str, SpaceApply, Space, SpaceApplyList, SpaceList]):
     resource_name = "spaces"
@@ -75,28 +95,20 @@ class ViewLoader(DataModelingLoader[ViewId, ViewApply, View, ViewApplyList, View
     resource_name = "views"
 
     def __init__(self, client: CogniteClient, existing_handling: Literal["fail", "skip", "update", "force"] = "fail"):
-        self.client = client
+        super().__init__(client)
         self.existing_handling = existing_handling
         self._interfaces_by_id: dict[ViewId, View] = {}
+        self._tried_force_deploy: set[ViewId] = set()
 
     @classmethod
     def get_id(cls, item: View | ViewApply) -> ViewId:
         return item.as_id()
 
     def create(self, items: Sequence[ViewApply]) -> ViewList:
-        try:
+        if self.existing_handling == "force":
+            return self._create_force(items, self._tried_force_deploy, self.client.data_modeling.views.apply)
+        else:
             return self.client.data_modeling.views.apply(items)
-        except CogniteAPIError as e:
-            if self.existing_handling == "force" and e.message.startswith("Cannot update view"):
-                res = re.search(r"(?<=\')(.*?)(?=\')", e.message)
-                if res is None or ":" not in res.group(1) or "/" not in res.group(1):
-                    raise e
-                view_id_str = res.group(1)
-                space, external_id_version = view_id_str.split(":")
-                external_id, version = external_id_version.split("/")
-                self.delete([ViewId(space, external_id, version)])
-                return self.create(items)
-            raise e
 
     def retrieve(self, ids: SequenceNotStr[ViewId]) -> ViewList:
         return self.client.data_modeling.views.retrieve(cast(Sequence, ids))
@@ -174,6 +186,11 @@ def _retrieve_view_ancestors(self, parents: list[ViewId], cache: dict[ViewId, Vi
 class ContainerLoader(DataModelingLoader[ContainerId, ContainerApply, Container, ContainerApplyList, ContainerList]):
     resource_name = "containers"
 
+    def __init__(self, client: CogniteClient, existing_handling: Literal["fail", "skip", "update", "force"] = "fail"):
+        super().__init__(client)
+        self.existing_handling = existing_handling
+        self._tried_force_deploy: set[ContainerId] = set()
+
     @classmethod
     def get_id(cls, item: Container | ContainerApply) -> ContainerId:
         return item.as_id()
@@ -193,7 +210,10 @@ def sort_by_dependencies(self, items: Sequence[ContainerApply]) -> list[Containe
         ]
 
     def create(self, items: Sequence[ContainerApply]) -> ContainerList:
-        return self.client.data_modeling.containers.apply(items)
+        if self.existing_handling == "force":
+            return self._create_force(items, self._tried_force_deploy, self.client.data_modeling.containers.apply)
+        else:
+            return self.client.data_modeling.containers.apply(items)
 
     def retrieve(self, ids: SequenceNotStr[ContainerId]) -> ContainerList:
         return self.client.data_modeling.containers.retrieve(cast(Sequence, ids))

tests/tests_integration/conftest.py

@@ -7,7 +7,7 @@
 from tests.config import ROOT
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 def cognite_client() -> CogniteClient:
     load_dotenv(ROOT / ".env", override=True)
 

tests/tests_integration/test_utils/test_cdf_loaders.py

@@ -0,0 +1,95 @@
+import pytest
+from cognite.client import CogniteClient
+from cognite.client import data_modeling as dm
+
+from cognite.neat.utils.cdf_loaders import ContainerLoader, ViewLoader
+
+
+@pytest.fixture(scope="session")
+def space(cognite_client: CogniteClient) -> dm.Space:
+    space = dm.SpaceApply(
+        space="test_space", description="This space is used by Neat for integration tests", name="Test Space"
+    )
+    return cognite_client.data_modeling.spaces.apply(space)
+
+
+@pytest.fixture(scope="session")
+def container_props(cognite_client: CogniteClient, space: dm.Space) -> dm.Container:
+    container = dm.ContainerApply(
+        space=space.space,
+        external_id="PropContainer",
+        properties={
+            "name": dm.ContainerProperty(type=dm.Text()),
+            "other": dm.ContainerProperty(type=dm.DirectRelation()),
+            "number": dm.ContainerProperty(type=dm.Int64()),
+            "float": dm.ContainerProperty(type=dm.Float64()),
+        },
+    )
+    return cognite_client.data_modeling.containers.apply(container)
+
+
+class TestViewLoader:
+    def test_force_create(self, cognite_client: CogniteClient, container_props: dm.Container, space: dm.Space) -> None:
+        container_id = container_props.as_id()
+        original = dm.ViewApply(
+            space=space.space,
+            external_id="test_view",
+            version="1",
+            properties={
+                "name": dm.MappedPropertyApply(container=container_id, container_property_identifier="name"),
+                "other": dm.MappedPropertyApply(
+                    container=container_id,
+                    container_property_identifier="other",
+                    source=dm.ViewId(space.space, "test_view", "1"),
+                ),
+                "count": dm.MappedPropertyApply(container=container_id, container_property_identifier="number"),
+            },
+        )
+        retrieved_list = cognite_client.data_modeling.views.retrieve(original.as_id())
+        if not retrieved_list:
+            cognite_client.data_modeling.views.apply(original)
+            existing = original
+        else:
+            existing = retrieved_list[0]
+        modified = dm.ViewApply.load(original.dump_yaml())
+        # Change the type for each time the test runs to require a force update
+        new_prop = "float" if existing.properties["count"].container_property_identifier == "number" else "number"
+        modified.properties["count"] = dm.MappedPropertyApply(
+            container=container_id, container_property_identifier=new_prop
+        )
+
+        loader = ViewLoader(cognite_client, existing_handling="force")
+        new_created = loader.create([modified])[0]
+
+        assert new_created.as_id() == original.as_id(), "The view version should be the same"
+        assert (
+            new_created.properties["count"].container_property_identifier == new_prop
+        ), "The property should have been updated"
+
+
+class TestContainerLoader:
+    def test_force_create(self, cognite_client: CogniteClient, space: dm.Space) -> None:
+        original = dm.ContainerApply(
+            space=space.space,
+            external_id="test_container",
+            properties={
+                "name": dm.ContainerProperty(type=dm.Text()),
+                "number": dm.ContainerProperty(type=dm.Int64()),
+            },
+        )
+        retrieved = cognite_client.data_modeling.containers.retrieve(original.as_id())
+        if retrieved is None:
+            cognite_client.data_modeling.containers.apply(original)
+            existing = original
+        else:
+            existing = retrieved
+        modified = dm.ContainerApply.load(original.dump_yaml())
+        # Change the type for each time the test runs to require a force update
+        new_prop = dm.Float64() if isinstance(existing.properties["number"].type, dm.Int64) else dm.Int64()
+        modified.properties["number"] = dm.ContainerProperty(type=new_prop)
+
+        loader = ContainerLoader(cognite_client, existing_handling="force")
+        new_created = loader.create([modified])[0]
+
+        assert new_created.as_id() == original.as_id(), "The container version should be the same"
+        assert new_created.properties["number"].type == new_prop, "The property should have been updated"