Migrate token authorization to new fixtures (#175)

Author: nfx

src/databricks/labs/ucx/providers/mixins/fixtures.py

@@ -131,8 +131,7 @@ def _path(ws, path):
             [PermissionLevel.CAN_READ, PermissionLevel.CAN_RUN, PermissionLevel.CAN_EDIT, PermissionLevel.CAN_MANAGE],
             _path,
         ),
-        ("tokens_authorization", "authorization", [PermissionLevel.CAN_USE], _simple),
-        ("passwords_authorization", "authorization", [PermissionLevel.CAN_USE], _simple),
+        ("authorization", "authorization", [PermissionLevel.CAN_USE], _simple),
         (
             "warehouse",
             "sql/warehouses",

tests/integration/conftest.py

@@ -234,24 +234,6 @@ def instance_profiles(env: EnvironmentInfo, ws: WorkspaceClient) -> list[Instanc
     logger.debug("Test instance profiles deleted")
 
 
-@pytest.fixture
-def tokens(ws: WorkspaceClient, env: EnvironmentInfo) -> list[AccessControlRequest]:
-    logger.debug("Adding token-level permissions to groups")
-
-    token_permissions = [
-        AccessControlRequest(group_name=ws_group.display_name, permission_level=PermissionLevel.CAN_USE)
-        for ws_group, _ in random.sample(env.groups, k=min(len(env.groups), NUM_TEST_TOKENS))
-    ]
-
-    ws.permissions.update(
-        request_object_type=RequestObjectType.AUTHORIZATION,
-        request_object_id="tokens",
-        access_control_list=token_permissions,
-    )
-
-    yield token_permissions
-
-
 @pytest.fixture
 def workspace_objects(ws: WorkspaceClient, env: EnvironmentInfo) -> WorkspaceObjects:
     logger.info(f"Creating test workspace objects under /{env.test_uid}")
@@ -309,12 +291,10 @@ def workspace_objects(ws: WorkspaceClient, env: EnvironmentInfo) -> WorkspaceObj
 
 @pytest.fixture
 def verifiable_objects(
-    tokens,
     workspace_objects,
 ) -> list[tuple[list, str, RequestObjectType | None]]:
     _verifiable_objects = [
         (workspace_objects, "workspace_objects", None),
-        (tokens, "tokens", RequestObjectType.AUTHORIZATION),
     ]
     yield _verifiable_objects
 

tests/integration/test_e2e.py

@@ -5,13 +5,8 @@
 
 import pytest
 from databricks.sdk import WorkspaceClient
-from databricks.sdk.service import workspace
-from databricks.sdk.service.iam import (
-    AccessControlRequest,
-    AccessControlResponse,
-    Permission,
-    PermissionLevel,
-)
+from databricks.sdk.service import iam, workspace
+from databricks.sdk.service.iam import PermissionLevel
 from pyspark.errors import AnalysisException
 
 from databricks.labs.ucx.config import (
@@ -72,33 +67,6 @@ def _verify_group_permissions(
             toolkit.permissions_manager.verify_applied_scope_acls(
                 scope_name, toolkit.group_manager.migration_groups_provider, target
             )
-
-    elif id_attribute in ("tokens", "passwords"):
-        _typed_objects: list[AccessControlRequest] = objects
-        ws_permissions = [
-            AccessControlResponse(
-                all_permissions=[
-                    Permission(permission_level=o.permission_level, inherited=False, inherited_from_object=None)
-                ],
-                group_name=o.group_name,
-            )
-            for o in _typed_objects
-        ]
-
-        target_permissions = list(
-            filter(
-                lambda p: p.group_name
-                in [getattr(g, target).display_name for g in toolkit.group_manager.migration_groups_provider.groups],
-                ws.permissions.get(
-                    request_object_type=request_object_type, request_object_id=id_attribute
-                ).access_control_list,
-            )
-        )
-
-        sorted_ws = sorted(ws_permissions, key=lambda p: p.group_name)
-        sorted_target = sorted(target_permissions, key=lambda p: p.group_name)
-
-        assert [p.all_permissions for p in sorted_ws] == [p.all_permissions for p in sorted_target]
     else:
         for _object in objects:
             toolkit.permissions_manager.verify_applied_permissions(
@@ -146,6 +114,7 @@ def test_e2e(
     make_pipeline_permissions,
     make_secret_scope,
     make_secret_scope_acl,
+    make_authorization_permissions,
     make_warehouse,
     make_warehouse_permissions,
 ):
@@ -196,7 +165,7 @@ def test_e2e(
         group_name=ws_group.display_name,
     )
     verifiable_objects.append(
-        ([model], "experiment_id", RequestObjectType.EXPERIMENTS),
+        ([model], "id", RequestObjectType.REGISTERED_MODELS),
     )
 
     experiment = make_experiment()
@@ -237,6 +206,15 @@ def test_e2e(
     make_secret_scope_acl(scope=scope, principal=ws_group.display_name, permission=workspace.AclPermission.WRITE)
     verifiable_objects.append(([scope], "secret_scopes", None))
 
+    make_authorization_permissions(
+        object_id="tokens",
+        permission_level=PermissionLevel.CAN_USE,
+        group_name=ws_group.display_name,
+    )
+    verifiable_objects.append(
+        ([iam.ObjectPermissions(object_id="tokens")], "object_id", RequestObjectType.AUTHORIZATION)
+    )
+
     warehouse = make_warehouse()
     make_warehouse_permissions(
         object_id=warehouse.id,