add new attribute to be added ot the config for object permissions

tests/integration/conftest.py

@@ -653,6 +653,7 @@ def config(self) -> WorkspaceConfig:
             include_job_ids=self.created_jobs,
             include_dashboard_ids=self.created_dashboards,
             include_query_ids=self.created_queries,
+            include_object_permissions=self.created_object_permissions,
         )
 
     @cached_property
@@ -779,6 +780,35 @@ def created_dashboards(self) -> list[str]:
                 raise ValueError(f"Unsupported dashboard: {dashboard}")
         return dashboard_ids
 
+    @property
+    def created_object_permissions(self) -> list[str]:
+        # Initialize include_object_permissions for the created fixtures
+        # Currently only supports the object types for which the fixtures exist
+        created_object_permissions = set()
+
+        # GenericPermissionsSupport
+        for cluster_policy in self._cluster_policies:
+            created_object_permissions.add(f"cluster_policies:{cluster_policy.policy_id}")
+        for job in self._jobs:
+            created_object_permissions.add(f"jobs:{job.job_id}")
+
+        # TableAclSupport
+        for table in self._tables:
+            created_object_permissions.add(f"TABLE:{table.full_name}")
+        for schema in self._schemas:
+            created_object_permissions.add(f"DATABASE:{schema.full_name}")
+        for catalog in self._catalogs:
+            created_object_permissions.add(f"CATALOG:{catalog.name}")
+        for udf in self._udfs:
+            created_object_permissions.add(f"FUNCTION:{udf.name}")
+
+        for secret_scope in self._secret_scopes:
+            created_object_permissions.add(f"secrets:{secret_scope}")
+        for query in self._queries:
+            created_object_permissions.add(f"queries:{query.id}")
+
+        return list(created_object_permissions)
+
     @cached_property
     def azure_service_principal_crawler(self) -> StaticServicePrincipalCrawler:
         return StaticServicePrincipalCrawler(
@@ -1087,34 +1117,6 @@ def config_transform(self) -> Callable[[WorkspaceConfig], WorkspaceConfig]:
     def include_object_permissions(self) -> list[str] | None:
         return None
 
-    def configure_include_object_permissions(self) -> None:
-        # Initialize include_object_permissions for the created fixtures
-        # Currently only supports the object types for which the fixtures exist
-
-        if not self.config.include_object_permissions:
-            self.config.include_object_permissions = []
-
-        # GenericPermissionsSupport
-        for cluster_policy in self._cluster_policies:
-            self.config.include_object_permissions.append(f"cluster_policies:{cluster_policy.policy_id}")
-        for job in self._jobs:
-            self.config.include_object_permissions.append(f"jobs:{job.job_id}")
-
-        # TableAclSupport
-        for table in self._tables:
-            self.config.include_object_permissions.append(f"TABLE:{table.full_name}")
-        for schema in self._schemas:
-            self.config.include_object_permissions.append(f"DATABASE:{schema.full_name}")
-        for catalog in self._catalogs:
-            self.config.include_object_permissions.append(f"CATALOG:{catalog.name}")
-        for udf in self._udfs:
-            self.config.include_object_permissions.append(f"FUNCTION:{udf.name}")
-
-        for secret_scope in self._secret_scopes:
-            self.config.include_object_permissions.append(f"secrets:{secret_scope}")
-        for query in self._queries:
-            self.config.include_object_permissions.append(f"queries:{query.id}")
-
     @cached_property
     def config(self) -> WorkspaceConfig:
         workspace_config = self.workspace_installer.configure()

tests/integration/workspace_access/test_workflows.py

@@ -51,7 +51,6 @@ def test_running_real_migrate_groups_job(
         scope=secret_scope, principal=ws_group.display_name, permission=AclPermission.WRITE
     )
 
-    installation_ctx.configure_include_object_permissions()
     installation_ctx.workspace_installation.run()
     # The crawlers should run as part of the assessment. To minimize the crawling here, we only crawl what is necessary
     # Tables crawler fails on `tacl` cluster used by the apply and validate permission tasks
@@ -125,7 +124,6 @@ def test_running_legacy_validate_groups_permissions_job(
     make_secret_scope_acl(scope=secret_scope, principal=ws_group_a.display_name, permission=AclPermission.WRITE)
 
     installation_ctx.__dict__['include_group_names'] = [ws_group_a.display_name]
-    installation_ctx.configure_include_object_permissions()
     installation_ctx.__dict__['config_transform'] = lambda c: replace(c, use_legacy_permission_migration=True)
     installation_ctx.workspace_installation.run()
     installation_ctx.permission_manager.snapshot()