[CHORE] Moving QueryLinter input objects to signature (#3250)

## Changes
Follow the more common pattern of expect objects in a class signature.
Those objects are passed on from the context. Makes testing more
straightforward

Breaking down the linked PR below.

### Linked issues
Progresses #3045
Breaks up #3112

### Tests

- [x] modified unit tests
- [x] modified integration tests

Author: JCZuurmond

src/databricks/labs/ucx/assessment/workflows.py

@@ -195,7 +195,7 @@ def assess_dashboards(self, ctx: RuntimeContext):
 
         Also, stores direct filesystem accesses for display in the migration dashboard.
         """
-        ctx.query_linter.refresh_report(ctx.sql_backend, ctx.inventory_database)
+        ctx.query_linter.refresh_report()
 
     @job_task
     def assess_workflows(self, ctx: RuntimeContext):

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

@@ -535,7 +535,9 @@ def workflow_linter(self) -> WorkflowLinter:
     def query_linter(self) -> QueryLinter:
         return QueryLinter(
             self.workspace_client,
-            TableMigrationIndex([]),  # TODO: bring back self.tables_migrator.index()
+            self.sql_backend,
+            self.inventory_database,
+            TableMigrationIndex([]),
             self.directfs_access_crawler_for_queries,
             self.used_tables_crawler_for_queries,
             self.config.include_dashboard_ids,

src/databricks/labs/ucx/progress/workflows.py

@@ -145,7 +145,7 @@ def assess_dashboards(self, ctx: RuntimeContext):
         """Scans all dashboards for migration issues in SQL code of embedded widgets.
         Also stores direct filesystem accesses for display in the migration dashboard."""
         # TODO: Ensure these are captured in the history log.
-        ctx.query_linter.refresh_report(ctx.sql_backend, ctx.inventory_database)
+        ctx.query_linter.refresh_report()
 
     @job_task(depends_on=[verify_prerequisites])
     def assess_workflows(self, ctx: RuntimeContext):

src/databricks/labs/ucx/source_code/queries.py

@@ -1,6 +1,6 @@
 import dataclasses
 import logging
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 
@@ -46,41 +46,62 @@ class QueryLinter:
     def __init__(
         self,
         ws: WorkspaceClient,
+        sql_backend: SqlBackend,
+        inventory_database: str,
         migration_index: TableMigrationIndex,
         directfs_crawler: DirectFsAccessCrawler,
         used_tables_crawler: UsedTablesCrawler,
         include_dashboard_ids: list[str] | None,
         debug_listing_upper_limit: int | None = None,
     ):
         self._ws = ws
+        self._sql_backend = sql_backend
         self._migration_index = migration_index
         self._directfs_crawler = directfs_crawler
         self._used_tables_crawler = used_tables_crawler
         self._include_dashboard_ids = include_dashboard_ids
         self._debug_listing_upper_limit = debug_listing_upper_limit
 
-    def refresh_report(self, sql_backend: SqlBackend, inventory_database: str) -> None:
+        self._catalog = "hive_metastore"
+        self._schema = inventory_database
+        self._table = "query_problems"
+
+    @property
+    def _full_name(self) -> str:
+        """Generates the full name of the table.
+
+        Returns:
+            str: The full table name.
+        """
+        return f"{self._catalog}.{self._schema}.{self._table}"
+
+    def refresh_report(self) -> None:
         assessment_start = datetime.now(timezone.utc)
         context = _ReportingContext()
         self._lint_dashboards(context)
         self._lint_queries(context)
         assessment_end = datetime.now(timezone.utc)
-        self._dump_problems(context, sql_backend, inventory_database)
-        self._dump_dfsas(context, assessment_start, assessment_end)
-        self._dump_used_tables(context, assessment_start, assessment_end)
+        self._dump_problems(context.all_problems)
+        self._dump_dfsas(context.all_dfsas, assessment_start, assessment_end)
+        self._dump_used_tables(context.all_tables, assessment_start, assessment_end)
 
-    def _dump_problems(self, context: _ReportingContext, sql_backend: SqlBackend, inventory_database: str) -> None:
-        logger.info(f"Saving {len(context.all_problems)} linting problems...")
-        sql_backend.save_table(
-            f'{escape_sql_identifier(inventory_database)}.query_problems',
-            context.all_problems,
+    def _dump_problems(self, problems: Sequence[QueryProblem]) -> None:
+        logger.info(f"Saving {len(problems)} linting problems...")
+        self._sql_backend.save_table(
+            escape_sql_identifier(self._full_name),
+            problems,
             QueryProblem,
             mode='overwrite',
         )
 
-    def _dump_dfsas(self, context: _ReportingContext, assessment_start: datetime, assessment_end: datetime) -> None:
+    def _dump_dfsas(
+        self,
+        dfsas: Sequence[DirectFsAccess],
+        assessment_start: datetime,
+        assessment_end: datetime,
+    ) -> None:
         processed_dfsas = []
-        for dfsa in context.all_dfsas:
+        for dfsa in dfsas:
             dfsa = dataclasses.replace(
                 dfsa,
                 assessment_start_timestamp=assessment_start,
@@ -91,18 +112,18 @@ def _dump_dfsas(self, context: _ReportingContext, assessment_start: datetime, as
 
     def _dump_used_tables(
         self,
-        context: _ReportingContext,
+        used_tables: Sequence[UsedTable],
         assessment_start: datetime,
         assessment_end: datetime,
     ) -> None:
         processed_tables = []
-        for table in context.all_tables:
-            table = dataclasses.replace(
-                table,
+        for used_table in used_tables:
+            used_table = dataclasses.replace(
+                used_table,
                 assessment_start_timestamp=assessment_start,
                 assessment_end_timestamp=assessment_end,
             )
-            processed_tables.append(table)
+            processed_tables.append(used_table)
         self._used_tables_crawler.dump_all(processed_tables)
 
     def _lint_dashboards(self, context: _ReportingContext) -> None:

tests/integration/source_code/test_directfs_access.py

@@ -13,12 +13,14 @@ def test_query_dfsa_ownership(runtime_ctx, make_query, make_dashboard, inventory
     # Produce a DFSA record for the query.
     linter = QueryLinter(
         runtime_ctx.workspace_client,
+        sql_backend,
+        inventory_schema,
         TableMigrationIndex([]),
         runtime_ctx.directfs_access_crawler_for_queries,
         runtime_ctx.used_tables_crawler_for_queries,
         include_dashboard_ids=[dashboard.id],
     )
-    linter.refresh_report(sql_backend, inventory_schema)
+    linter.refresh_report()
 
     # Find a record for the query.
     records = runtime_ctx.directfs_access_crawler_for_queries.snapshot()

tests/integration/source_code/test_queries.py

@@ -13,12 +13,14 @@ def test_query_linter_lints_queries_and_stores_dfsas_and_tables(
     dashboards.append(make_dashboard(query=queries[1]))
     linter = QueryLinter(
         ws,
+        sql_backend,
+        simple_ctx.inventory_database,
         TableMigrationIndex([]),
         simple_ctx.directfs_access_crawler_for_queries,
         simple_ctx.used_tables_crawler_for_queries,
         None,
     )
-    linter.refresh_report(sql_backend, simple_ctx.inventory_database)
+    linter.refresh_report()
     all_problems = sql_backend.fetch("SELECT * FROM query_problems", schema=simple_ctx.inventory_database)
     problems = [row for row in all_problems if row["query_name"] == queries[0].name]
     assert len(problems) == 1

tests/unit/progress/test_workflows.py

@@ -77,8 +77,8 @@ def test_linter_runtime_refresh(run_workflow, task, linter) -> None:
     linter_class = get_type_hints(linter.func)["return"]
     mock_linter = create_autospec(linter_class)
     linter_name = linter.attrname
-    ctx = run_workflow(task, **{linter_name: mock_linter})
-    mock_linter.refresh_report.assert_called_once_with(ctx.sql_backend, ctx.inventory_database)
+    run_workflow(task, **{linter_name: mock_linter})
+    mock_linter.refresh_report.assert_called_once()
 
 
 def test_migration_progress_with_valid_prerequisites(run_workflow) -> None:

tests/unit/source_code/test_queries.py

@@ -24,12 +24,14 @@
         ),
     ],
 )
-def test_query_linter_collects_dfsas_from_queries(name, query, dfsa_paths, is_read, is_write, migration_index) -> None:
+def test_query_linter_collects_dfsas_from_queries(
+    name, query, dfsa_paths, is_read, is_write, migration_index, mock_backend
+) -> None:
     ws = create_autospec(WorkspaceClient)
     dfsa_crawler = create_autospec(DirectFsAccessCrawler)
     used_tables_crawler = create_autospec(UsedTablesCrawler)
     query = LegacyQuery.from_dict({"parent": "workspace", "name": name, "query": query})
-    linter = QueryLinter(ws, migration_index, dfsa_crawler, used_tables_crawler, None)
+    linter = QueryLinter(ws, mock_backend, "test", migration_index, dfsa_crawler, used_tables_crawler, None)
     dfsas = linter.collect_dfsas_from_query("no-dashboard-id", query)
     ws.assert_not_called()
     dfsa_crawler.assert_not_called()
@@ -43,11 +45,11 @@ def test_query_linter_refresh_report_writes_query_problems(migration_index, mock
     ws = create_autospec(WorkspaceClient)
     dfsa_crawler = create_autospec(DirectFsAccessCrawler)
     used_tables_crawler = create_autospec(UsedTablesCrawler)
-    linter = QueryLinter(ws, migration_index, dfsa_crawler, used_tables_crawler, None)
+    linter = QueryLinter(ws, mock_backend, "test", migration_index, dfsa_crawler, used_tables_crawler, None)
 
-    linter.refresh_report(mock_backend, inventory_database="test")
+    linter.refresh_report()
 
-    assert mock_backend.has_rows_written_for("`test`.query_problems")
+    assert mock_backend.has_rows_written_for("`hive_metastore`.`test`.`query_problems`")
     ws.dashboards.list.assert_called_once()
     dfsa_crawler.assert_not_called()
     used_tables_crawler.assert_not_called()
@@ -60,10 +62,10 @@ def test_lints_queries(migration_index, mock_backend) -> None:
         ws = create_autospec(WorkspaceClient)
         dfsa_crawler = create_autospec(DirectFsAccessCrawler)
         used_tables_crawler = create_autospec(UsedTablesCrawler)
-        linter = QueryLinter(ws, migration_index, dfsa_crawler, used_tables_crawler, ["1"])
-        linter.refresh_report(mock_backend, inventory_database="test")
+        linter = QueryLinter(ws, mock_backend, "test", migration_index, dfsa_crawler, used_tables_crawler, ["1"])
+        linter.refresh_report()
 
-        assert mock_backend.has_rows_written_for("`test`.query_problems")
+        assert mock_backend.has_rows_written_for("`hive_metastore`.`test`.`query_problems`")
         ws.dashboards.list.assert_not_called()
         dfsa_crawler.assert_not_called()
         used_tables_crawler.assert_not_called()