Addressed Integration Tests

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

@@ -281,7 +281,10 @@ def table_ownership(self) -> TableOwnership:
     def static_table_ownership(self) -> StaticTableOwnership:
         # Returns a static table ownership resolver
         return StaticTableOwnership(
-            self.administrator_locator, self.config.default_owner_group, self.connect_config.username
+            self.administrator_locator,
+            self.tables_crawler,
+            self.config.default_owner_group,
+            self.connect_config.username
         )
 
     @cached_property
@@ -335,7 +338,7 @@ def table_ownership_grant_loader(self) -> TableOwnershipGrantLoader:
     @cached_property
     def migrate_grants(self) -> MigrateGrants:
         # owner grants have to come first
-        ownership_loader: Callable[[], Iterable[Grant]] = self.table_ownership_grant_loader.load
+        ownership_loader: Callable[[], Iterable[Grant]] = self.static_table_ownership.load
         grant_loaders: list[Callable[[], Iterable[Grant]]] = [
             self.grants_crawler.snapshot,
             self.principal_acl.get_interactive_cluster_grants,

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

@@ -96,13 +96,34 @@ class StaticTableOwnership(Ownership[Table]):
     def __init__(
         self,
         administrator_locator: AdministratorLocator,
+        table_crawler: TablesCrawler,
         fixed_owner_group: str | None,
         application_principal: str | None,
     ) -> None:
+        self._tables_crawler = table_crawler
         self._fixed_owner_group = fixed_owner_group
         self._application_principal = application_principal
         super().__init__(administrator_locator)
 
+    def load(self) -> Iterable[Grant]:
+        for table in self._tables_crawler.snapshot():
+            owner = self._maybe_direct_owner(table)
+            table_name, view_name = self._names(table)
+            yield Grant(
+                principal=owner,
+                action_type='OWN',
+                catalog=table.catalog,
+                database=table.database,
+                table=table_name,
+                view=view_name,
+            )
+
+    @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:
         if self._fixed_owner_group:
             return self._fixed_owner_group

tests/integration/hive_metastore/test_ext_hms.py

@@ -40,6 +40,7 @@ def test_migration_job_ext_hms(ws, installation_ctx, prepare_tables_for_migratio
             r"Choose a cluster policy": "0",
         },
     )
+
     ext_hms_ctx.workspace_installation.run()
     ext_hms_ctx.deployed_workflows.run_workflow("migrate-tables")
     # assert the workflow is successful