feat(gooddata-pipelines): runtime validation of provisioning inputs

Author: janmatzek

gooddata-pipelines/gooddata_pipelines/provisioning/entities/users/permissions.py

@@ -35,6 +35,11 @@ class PermissionProvisioner(
     source_group_incremental: list[PermissionIncrementalLoad]
     source_group_full: list[PermissionFullLoad]
 
+    FULL_LOAD_TYPE: type[PermissionFullLoad] = PermissionFullLoad
+    INCREMENTAL_LOAD_TYPE: type[PermissionIncrementalLoad] = (
+        PermissionIncrementalLoad
+    )
+
     def _get_ws_declaration(self, ws_id: str) -> PermissionDeclaration:
         users: TargetsPermissionDict = {}
         user_groups: TargetsPermissionDict = {}

gooddata-pipelines/gooddata_pipelines/provisioning/entities/users/user_groups.py

@@ -29,6 +29,11 @@ class UserGroupProvisioner(
     source_group_full: list[UserGroupFullLoad]
     upstream_user_groups: list[CatalogUserGroup]
 
+    FULL_LOAD_TYPE: type[UserGroupFullLoad] = UserGroupFullLoad
+    INCREMENTAL_LOAD_TYPE: type[UserGroupIncrementalLoad] = (
+        UserGroupIncrementalLoad
+    )
+
     @staticmethod
     def _is_changed(
         group: UserGroupModel, existing_group: CatalogUserGroup

gooddata-pipelines/gooddata_pipelines/provisioning/entities/users/users.py

@@ -24,13 +24,15 @@ class UserProvisioner(Provisioning[UserFullLoad, UserIncrementalLoad]):
     """Provisioning class for users in GoodData workspaces.
 
     This class handles the creation, update, and deletion of users
-    based on the provided source data. Use the `full_load` or `incremental_load`
-    methods to run the provisioning.
+    based on the provided source data.
     """
 
     source_group_incremental: list[UserIncrementalLoad]
     source_group_full: list[UserFullLoad]
 
+    FULL_LOAD_TYPE: type[UserFullLoad] = UserFullLoad
+    INCREMENTAL_LOAD_TYPE: type[UserIncrementalLoad] = UserIncrementalLoad
+
     def __init__(self, host: str, token: str) -> None:
         super().__init__(host, token)
         self.upstream_user_cache: dict[UserId, UserModel] = {}

gooddata-pipelines/gooddata_pipelines/provisioning/entities/workspaces/workspace.py

@@ -35,6 +35,11 @@ class WorkspaceProvisioner(
     source_group_full: list[WorkspaceFullLoad]
     source_group_incremental: list[WorkspaceIncrementalLoad]
 
+    FULL_LOAD_TYPE: type[WorkspaceFullLoad] = WorkspaceFullLoad
+    INCREMENTAL_LOAD_TYPE: type[WorkspaceIncrementalLoad] = (
+        WorkspaceIncrementalLoad
+    )
+
     def __init__(self, *args: str, **kwargs: str) -> None:
         """Creates an instance of the WorkspaceProvisioner.
 

gooddata-pipelines/gooddata_pipelines/provisioning/provisioning.py

@@ -23,6 +23,8 @@ class Provisioning(Generic[TFullLoadSourceData, TIncrementalSourceData]):
     TProvisioning = TypeVar("TProvisioning", bound="Provisioning")
     source_group_full: list[TFullLoadSourceData]
     source_group_incremental: list[TIncrementalSourceData]
+    FULL_LOAD_TYPE: type[TFullLoadSourceData]
+    INCREMENTAL_LOAD_TYPE: type[TIncrementalSourceData]
 
     def __init__(self, host: str, token: str) -> None:
         self.source_id: set[str] = set()
@@ -80,6 +82,17 @@ def _create_groups(
             ids_to_create=ids_to_create,
         )
 
+    def _validate_source_data_type(
+        self,
+        source_data: list[TFullLoadSourceData] | list[TIncrementalSourceData],
+        model: type[TFullLoadSourceData] | type[TIncrementalSourceData],
+    ) -> None:
+        """Validates data type of the source data."""
+        if not all(isinstance(record, model) for record in source_data):
+            raise TypeError(
+                f"Not all elements in source data are instances of {model.__name__}"
+            )
+
     def _provision_incremental_load(self) -> None:
         raise NotImplementedError(
             "Provisioning method to be implemented in the subclass."
@@ -103,9 +116,10 @@ def full_load(self, source_data: list[TFullLoadSourceData]) -> None:
         - All child workspaces not declared under the parent workspace in the
         source data are deleted
         """
-        self.source_group_full = source_data
 
         try:
+            self._validate_source_data_type(source_data, self.FULL_LOAD_TYPE)
+            self.source_group_full = source_data
             self._provision_full_load()
             self.logger.info("Provisioning completed.")
         except Exception as e:
@@ -120,10 +134,11 @@ def incremental_load(
         based on the source data provided. Only changes requested in the source
         data will be applied.
         """
-        # TODO: validate the data type of source group at runtime
-        self.source_group_incremental = source_data
-
         try:
+            self._validate_source_data_type(
+                source_data, self.INCREMENTAL_LOAD_TYPE
+            )
+            self.source_group_incremental = source_data
             self._provision_incremental_load()
             self.logger.info("Provisioning completed.")
         except Exception as e:

gooddata-pipelines/tests/conftest.py

@@ -6,7 +6,15 @@
 import boto3
 import pytest
 from moto import mock_aws
-
+from pytest_mock import MockerFixture
+
+from gooddata_pipelines import (
+    PermissionProvisioner,
+    UserDataFilterProvisioner,
+    UserGroupProvisioner,
+    UserProvisioner,
+    WorkspaceProvisioner,
+)
 from gooddata_pipelines.api import GoodDataApi
 
 TEST_DATA_DIR = str((Path(__file__).parent / "data").absolute())
@@ -69,3 +77,53 @@ def error(self, msg: str) -> None:
             print(msg)
 
     return MockLogger()
+
+
+@pytest.fixture
+def workspace_provisioner(mocker: MockerFixture):
+    provisioner_instance = WorkspaceProvisioner.create("host", "token")
+
+    # Patch the API
+    mocker.patch.object(provisioner_instance, "_api", return_value=None)
+
+    return provisioner_instance
+
+
+@pytest.fixture
+def user_provisioner(mocker: MockerFixture):
+    provisioner_instance = UserProvisioner.create("host", "token")
+
+    # Patch the API
+    mocker.patch.object(provisioner_instance, "_api", return_value=None)
+
+    return UserProvisioner.create("host", "token")
+
+
+@pytest.fixture
+def user_group_provisioner(mocker: MockerFixture):
+    provisioner_instance = UserGroupProvisioner.create("host", "token")
+
+    # Patch the API
+    mocker.patch.object(provisioner_instance, "_api", return_value=None)
+
+    return provisioner_instance
+
+
+@pytest.fixture
+def permission_provisioner(mocker: MockerFixture) -> PermissionProvisioner:
+    provisioner_instance = PermissionProvisioner.create("host", "token")
+
+    # Patch the API
+    mocker.patch.object(provisioner_instance, "_api", return_value=None)
+
+    return provisioner_instance
+
+
+@pytest.fixture
+def user_data_filter_provisioner(mocker: MockerFixture):
+    provisioner_instance = UserDataFilterProvisioner.create("host", "token")
+
+    # Patch the API
+    mocker.patch.object(provisioner_instance, "_api", return_value=None)
+
+    return provisioner_instance

gooddata-pipelines/tests/provisioning/entities/users/test_permissions.py

@@ -402,18 +402,6 @@ def mock_upstream_perms(ws_id: str) -> CatalogDeclarativeWorkspacePermissions:
     return UPSTREAM_WS_PERMISSIONS[ws_id]
 
 
-@pytest.fixture
-def permission_provisioner(mocker: MockerFixture) -> PermissionProvisioner:
-    provisioner_instance = PermissionProvisioner.create(
-        host="https://localhost:3000", token="token"
-    )
-
-    # Patch the API
-    mocker.patch.object(provisioner_instance, "_api", return_value=None)
-
-    return provisioner_instance
-
-
 def parse_expected_permissions(
     raw_data: dict,
 ) -> dict[str, list[CatalogDeclarativeSingleWorkspacePermission]]:

gooddata-pipelines/tests/provisioning/entities/workspaces/test_provisioning.py

@@ -1,12 +1,13 @@
 # (C) 2025 GoodData Corporation
 
+from gooddata_pipelines import WorkspaceProvisioner
 from gooddata_pipelines.provisioning.provisioning import Provisioning
 
-MOCK_PROVISIONER: Provisioning = Provisioning.create("host", "token")
 
-
-def test_create_groups_base_case() -> None:
-    provisioner: Provisioning = MOCK_PROVISIONER
+def test_create_groups_base_case(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
+    provisioner: Provisioning = workspace_provisioner
     test_source_ids = {"source_id_1", "source_id_2", "source_id_3"}
     test_panther_ids = {"source_id_2", "source_id_3", "source_id_4"}
 
@@ -17,8 +18,10 @@ def test_create_groups_base_case() -> None:
     assert id_groups.ids_to_create == {"source_id_1"}
 
 
-def test_create_groups_empty_sets() -> None:
-    provisioner: Provisioning = MOCK_PROVISIONER
+def test_create_groups_empty_sets(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
+    provisioner: Provisioning = workspace_provisioner
     test_source_ids: set[str] = set()
     test_panther_ids: set[str] = set()
 
@@ -29,8 +32,10 @@ def test_create_groups_empty_sets() -> None:
     assert id_groups.ids_to_create == set()
 
 
-def test_create_groups_no_overlap() -> None:
-    provisioner: Provisioning = MOCK_PROVISIONER
+def test_create_groups_no_overlap(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
+    provisioner: Provisioning = workspace_provisioner
     test_source_ids = {"source_id_1", "source_id_2"}
     test_panther_ids = {"source_id_3", "source_id_4"}
 
@@ -41,8 +46,10 @@ def test_create_groups_no_overlap() -> None:
     assert id_groups.ids_to_create == {"source_id_1", "source_id_2"}
 
 
-def test_create_groups_full_overlap() -> None:
-    provisioner: Provisioning = MOCK_PROVISIONER
+def test_create_groups_full_overlap(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
+    provisioner: Provisioning = workspace_provisioner
     test_source_ids = {"source_id_1", "source_id_2"}
     test_panther_ids = {"source_id_1", "source_id_2"}
 

gooddata-pipelines/tests/provisioning/entities/workspaces/test_workspace.py

@@ -1,5 +1,6 @@
 # (C) 2025 GoodData Corporation
 
+import pytest
 from gooddata_sdk.catalog.workspace.entity_model.workspace import (
     CatalogWorkspace,
 )
@@ -9,13 +10,11 @@
     WorkspaceFullLoad,
 )
 
-MOCK_WORKSPACE_PROVISIONER: WorkspaceProvisioner = WorkspaceProvisioner.create(
-    "host", "token"
-)
-
 
-def test_find_workspaces_to_update_same_ids_and_names() -> None:
-    provisioner: WorkspaceProvisioner = MOCK_WORKSPACE_PROVISIONER
+@pytest.fixture
+def test_find_workspaces_to_update_same_ids_and_names(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
     ids_in_both_systems = {"workspace_id1", "workspace_id2"}
     source_group: list[WorkspaceFullLoad] = [
         WorkspaceFullLoad(
@@ -42,15 +41,16 @@ def test_find_workspaces_to_update_same_ids_and_names() -> None:
         ),
     ]
 
-    workspaces_to_update = provisioner._find_workspaces_to_update(
+    workspaces_to_update = workspace_provisioner._find_workspaces_to_update(
         source_group, panther_group, ids_in_both_systems
     )
 
     assert workspaces_to_update == set()
 
 
-def test_find_workspaces_to_update_different_ids() -> None:
-    provisioner: WorkspaceProvisioner = MOCK_WORKSPACE_PROVISIONER
+def test_find_workspaces_to_update_different_ids(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
     ids_in_both_systems = {"workspace_id1", "workspace_id2"}
     source_group: list[WorkspaceFullLoad] = [
         WorkspaceFullLoad(
@@ -77,15 +77,16 @@ def test_find_workspaces_to_update_different_ids() -> None:
         ),
     ]
 
-    workspaces_to_update = provisioner._find_workspaces_to_update(
+    workspaces_to_update = workspace_provisioner._find_workspaces_to_update(
         source_group, panther_group, ids_in_both_systems
     )
 
     assert workspaces_to_update == set()
 
 
-def test_find_workspaces_to_update_same_ids_different_names() -> None:
-    provisioner: WorkspaceProvisioner = MOCK_WORKSPACE_PROVISIONER
+def test_find_workspaces_to_update_same_ids_different_names(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
     ids_in_both_systems: set[str] = {"workspace_id1", "workspace_id2"}
     source_group: list[WorkspaceFullLoad] = [
         WorkspaceFullLoad(
@@ -112,15 +113,16 @@ def test_find_workspaces_to_update_same_ids_different_names() -> None:
         ),
     ]
 
-    workspaces_to_update = provisioner._find_workspaces_to_update(
+    workspaces_to_update = workspace_provisioner._find_workspaces_to_update(
         source_group, panther_group, ids_in_both_systems
     )
 
     assert workspaces_to_update == {"workspace_id1", "workspace_id2"}
 
 
-def test_find_workspaces_to_update_no_panther() -> None:
-    provisioner: WorkspaceProvisioner = MOCK_WORKSPACE_PROVISIONER
+def test_find_workspaces_to_update_no_panther(
+    workspace_provisioner: WorkspaceProvisioner,
+) -> None:
     ids_in_both_systems: set[str] = set()
     source_group: list[WorkspaceFullLoad] = [
         WorkspaceFullLoad(
@@ -136,7 +138,7 @@ def test_find_workspaces_to_update_no_panther() -> None:
     ]
     panther_group: list[CatalogWorkspace] = []
 
-    workspaces_to_update = provisioner._find_workspaces_to_update(
+    workspaces_to_update = workspace_provisioner._find_workspaces_to_update(
         source_group, panther_group, ids_in_both_systems
     )