Addressed Integration Tests Issues

Author: FastLee

src/databricks/labs/ucx/contexts/application.py

@@ -284,7 +284,7 @@ def static_table_ownership(self) -> StaticTableOwnership:
             self.administrator_locator,
             self.tables_crawler,
             self.config.default_owner_group,
-            self.connect_config.username,
+            self.workspace_client.current_user.me().user_name,
         )
 
     @cached_property

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

@@ -106,12 +106,16 @@ def __init__(
         super().__init__(administrator_locator)
 
     def load(self) -> Iterable[Grant]:
+        databases = set()
+        owner = self._static_owner()
+        if not owner:
+            logger.warning("No owner found for tables and databases")
+            return
         for table in self._tables_crawler.snapshot():
-            owner = self._maybe_direct_owner(table)
             table_name, view_name = self._names(table)
-            if not owner:
-                logger.warning(f"No owner found for {table.key}")
-                continue
+
+            if table.database not in databases:
+                databases.add(table.database)
             yield Grant(
                 principal=owner,
                 action_type='OWN',
@@ -120,18 +124,30 @@ def load(self) -> Iterable[Grant]:
                 table=table_name,
                 view=view_name,
             )
+        for database in databases:
+            yield Grant(
+                principal=owner,
+                action_type='OWN',
+                catalog="hive_metastore",
+                database=database,
+                table=None,
+                view=None,
+            )
 
     @staticmethod
     def _names(table: Table) -> tuple[str | None, str | None]:
         if table.view_text:
             return None, table.name
         return table.name, None
 
-    def _maybe_direct_owner(self, record: Table) -> str | None:
+    def _static_owner(self) -> str | None:
         if self._fixed_owner_group:
             return self._fixed_owner_group
         return self._application_principal
 
+    def _maybe_direct_owner(self, record: Table) -> str | None:
+        return self._static_owner()
+
 
 class TableOwnershipGrantLoader:
     def __init__(self, tables_crawler: TablesCrawler, table_ownership: Ownership[Table]) -> None:

tests/integration/hive_metastore/test_catalog_schema.py

@@ -170,7 +170,8 @@ def test_create_catalog_schema_with_legacy_hive_metastore_privileges(
     def get_schema_permissions_list(full_name: str) -> PermissionsList:
         return ws.grants.get(SecurableType.SCHEMA, full_name)
 
-    assert ws.schemas.get(f"{dst_catalog_name}.{dst_schema_name}").owner == schema_owner.user_name
+    assert (ws.schemas.get(f"{dst_catalog_name}.{dst_schema_name}").owner ==
+            runtime_ctx.workspace_client.current_user.me().user_name)
     schema_grants = get_schema_permissions_list(f"{dst_catalog_name}.{dst_schema_name}")
     assert schema_grants.privilege_assignments is not None
     assert PrivilegeAssignment(table_owner.user_name, [Privilege.USE_SCHEMA]) in schema_grants.privilege_assignments