Skip group migration when no groups are available after preparation s…

…tep. (#363)

Fixes #360

src/databricks/labs/ucx/runtime.py

@@ -190,9 +190,12 @@ def migrate_permissions(cfg: WorkspaceConfig):
     See [interactive tutorial here](https://app.getreprise.com/launch/myM3VNn/)."""
     toolkit = GroupMigrationToolkit(cfg)
     toolkit.prepare_environment()
-    toolkit.apply_permissions_to_backup_groups()
-    toolkit.replace_workspace_groups_with_account_groups()
-    toolkit.apply_permissions_to_account_groups()
+    if toolkit.has_groups():
+        toolkit.apply_permissions_to_backup_groups()
+        toolkit.replace_workspace_groups_with_account_groups()
+        toolkit.apply_permissions_to_account_groups()
+    else:
+        logger.info("Skipping group migration as no groups were found.")
 
 
 @task("migrate-groups-cleanup", depends_on=[migrate_permissions])

src/databricks/labs/ucx/workspace_access/groups.py

@@ -181,6 +181,9 @@ def prepare_groups_in_environment(self):
         self._set_migration_groups(group_names)
         logger.info("Environment prepared successfully")
 
+    def has_groups(self) -> bool:
+        return len(self._migration_state.groups) > 0
+
     @property
     def migration_groups_provider(self) -> GroupMigrationState:
         assert len(self._migration_state.groups) > 0, "Migration groups were not loaded or initialized"

src/databricks/labs/ucx/workspace_access/migration.py

@@ -114,6 +114,9 @@ def _configure_logger(level: str):
         ucx_logger = logging.getLogger("databricks.labs.ucx")
         ucx_logger.setLevel(level)
 
+    def has_groups(self) -> bool:
+        return self._group_manager.has_groups()
+
     def prepare_environment(self):
         self._group_manager.prepare_groups_in_environment()
 

tests/unit/workspace_access/test_groups.py

@@ -286,6 +286,19 @@ def test_prepare_groups_in_environment_with_conf_in_auto_mode_should_populate_mi
     assert manager._migration_state.groups == [group_info]
 
 
+def test_prepare_groups_in_environment_with_no_groups():
+    client = Mock()
+    client.groups.list.return_value = iter([])
+    client.api_client.do.return_value = {
+        "Resources": [],
+    }
+
+    group_conf = GroupsConfig(auto=True)
+    manager = GroupManager(client, group_conf)
+    manager.prepare_groups_in_environment()
+    assert not manager.has_groups()
+
+
 def test_replace_workspace_groups_with_account_groups_should_call_delete_and_do():
     client = Mock()