Let create-catalog-schemas command run as collection (#2653)

## Changes
Let `validate-external-locations` command to run as collection

### Linked issues

Resolves #2609

### Functionality

- [x] modified existing command: `databricks labs ucx
create-catalog-schemas`

### Tests

- [x] manually tested
- [x] added unit tests
- [ ] ~added integration tests~ : Covering after #2507

Author: JCZuurmond

labs.yml

@@ -212,8 +212,12 @@ commands:
         description: Run the command for the collection of workspaces with ucx installed. Default is False.
 
   - name: create-catalogs-schemas
-    description: Create UC external catalogs and schemas based on the destinations created from create_table_mapping command.
-      This command is supposed to be run before migrating tables to UC.
+    description: |
+      Create UC external catalogs and schemas based on the destinations created from `create_table_mapping` command.
+      This command should be executed before migrating tables to Unity Catalog.
+    flags:
+      - name: run-as-collection
+        description: Run the command for the collection of workspaces with ucx installed. Default is False.
 
   - name: cluster-remap
     description: Re-mapping the cluster to UC

src/databricks/labs/ucx/cli.py

@@ -452,10 +452,20 @@ def migrate_locations(
 
 
 @ucx.command
-def create_catalogs_schemas(w: WorkspaceClient, prompts: Prompts):
+def create_catalogs_schemas(
+    w: WorkspaceClient,
+    prompts: Prompts,
+    ctx: WorkspaceContext | None = None,
+    run_as_collection: bool = False,
+    a: AccountClient | None = None,
+) -> None:
     """Create UC catalogs and schemas based on the destinations created from create_table_mapping command."""
-    ctx = WorkspaceContext(w)
-    ctx.catalog_schema.create_all_catalogs_schemas(prompts)
+    if ctx:
+        workspace_contexts = [ctx]
+    else:
+        workspace_contexts = _get_workspace_contexts(w, a, run_as_collection)
+    for workspace_context in workspace_contexts:
+        workspace_context.catalog_schema.create_all_catalogs_schemas(prompts)
 
 
 @ucx.command

src/databricks/labs/ucx/hive_metastore/catalog_schema.py

@@ -27,7 +27,7 @@ def __init__(
         self._principal_grants = principal_grants
         self._backend = sql_backend
 
-    def create_all_catalogs_schemas(self, prompts: Prompts):
+    def create_all_catalogs_schemas(self, prompts: Prompts) -> None:
         candidate_catalogs, candidate_schemas = self._get_missing_catalogs_schemas()
         for candidate_catalog in candidate_catalogs:
             try:

tests/unit/test_cli.py

@@ -713,19 +713,26 @@ def test_migrate_locations_gcp(ws):
         migrate_locations(ws, ctx=ctx)
 
 
-def test_create_catalogs_schemas(ws):
+@pytest.mark.parametrize("run_as_collection", [False, True])
+def test_create_catalogs_schemas_lists_catalogs(run_as_collection, workspace_clients, acc_client) -> None:
+    if not run_as_collection:
+        workspace_clients = [workspace_clients[0]]
+    for workspace_client in workspace_clients:
+        workspace_client.external_locations.list.return_value = [ExternalLocationInfo(url="s3://test")]
     prompts = MockPrompts({'.*': 's3://test'})
-    ws.external_locations.list.return_value = [ExternalLocationInfo(url="s3://test")]
-    create_catalogs_schemas(ws, prompts)
-    ws.catalogs.list.assert_called_once()
+
+    create_catalogs_schemas(workspace_clients[0], prompts, run_as_collection=run_as_collection, a=acc_client)
+
+    for workspace_client in workspace_clients:
+        workspace_client.catalogs.list.assert_called_once()
 
 
-def test_create_catalogs_schemas_handles_existing(ws, caplog):
+def test_create_catalogs_schemas_handles_existing(ws, caplog) -> None:
     prompts = MockPrompts({'.*': 's3://test'})
     ws.external_locations.list.return_value = [ExternalLocationInfo(url="s3://test")]
     ws.catalogs.create.side_effect = [BadRequest("Catalog 'test' already exists")]
     ws.schemas.create.side_effect = [BadRequest("Schema 'test' already exists")]
-    create_catalogs_schemas(ws, prompts)
+    create_catalogs_schemas(ws, prompts, ctx=WorkspaceContext(ws))
     ws.catalogs.list.assert_called_once()
 
     assert "Catalog test already exists. Skipping." in caplog.messages