Fix fmt

Author: pritishpai

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

@@ -1,5 +1,4 @@
 import logging
-from functools import cached_property
 
 from databricks.sdk.service.jobs import JobParameterDefinition
 
@@ -19,12 +18,14 @@ def _get_force_refresh(ctx: RuntimeContext) -> bool:
         """Extracts the force_refresh parameter from the job run parameters."""
         force_refresh = False
         job_id = ctx.install_state.jobs["assessment"]
-        job_runs = [job_run for job_run in ctx.workspace_client.jobs.list_runs(active_only=True, job_id=job_id)]
+        job_runs = list(ctx.workspace_client.jobs.list_runs(active_only=True, job_id=job_id))
         job_parameters = job_runs[0].job_parameters if job_runs else []
+        if not job_parameters:
+            return False
         for job_parameter in job_parameters:
             if job_parameter.name == "force_refresh" and job_parameter.value is not None:
                 force_refresh = job_parameter.value.lower() in {"true", "1"}
-                logger.info("Found force_refresh parameter in job run: %s", force_refresh)
+                logger.info(f"Found force_refresh parameter in job run: {force_refresh}")
                 break
         return force_refresh
 

src/databricks/labs/ucx/cli.py

@@ -219,7 +219,9 @@ def validate_external_locations(
 
 
 @ucx.command
-def ensure_assessment_run(w: WorkspaceClient, run_as_collection: bool = False, a: AccountClient | None = None, **named_parameters):
+def ensure_assessment_run(
+    w: WorkspaceClient, run_as_collection: bool = False, a: AccountClient | None = None, **named_parameters
+):
     """ensure the assessment job was run on a workspace"""
     workspace_contexts = _get_workspace_contexts(w, a, run_as_collection)
     force_refresh = named_parameters.get("force_refresh", "false").lower() == "true"
@@ -232,11 +234,14 @@ def ensure_assessment_run(w: WorkspaceClient, run_as_collection: bool = False, a
             logger.info(f"The assessment workflow has successfully completed in workspace: {workspace_id}")
         elif force_refresh:
             logger.info(f"Re-running assessment workflow in workspace: {workspace_id}")
-            deployed_workflows.run_workflow("assessment", skip_job_wait=run_as_collection, named_parameters={"force_refresh": "true"})
+            deployed_workflows.run_workflow(
+                "assessment", skip_job_wait=run_as_collection, named_parameters={"force_refresh": "true"}
+            )
         else:
             logger.info(f"Starting assessment workflow in workspace: {workspace_id}")
             # If running for a collection, don't wait for each assessment job to finish as that will take a long time.
 
+
 @ucx.command
 def update_migration_progress(
     w: WorkspaceClient,

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

@@ -249,8 +249,13 @@ def __init__(self, ws: WorkspaceClient, install_state: InstallState):
         self._ws = ws
         self._install_state = install_state
 
-    def run_workflow(self, step: str, skip_job_wait: bool = False, max_wait: timedelta = timedelta(minutes=20),
-                     named_parameters: dict[str, str] | None = None ) -> int:
+    def run_workflow(
+        self,
+        step: str,
+        skip_job_wait: bool = False,
+        max_wait: timedelta = timedelta(minutes=20),
+        named_parameters: dict[str, str] | None = None,
+    ) -> int:
         # this dunder variable is hiding this method from tracebacks, making it cleaner
         # for the user to see the actual error without too much noise.
         __tracebackhide__ = True  # pylint: disable=unused-variable

tests/integration/assessment/test_workflows.py

@@ -51,9 +51,18 @@ def test_running_real_assessment_job(
     assert not workflow_problems_without_path
 
     post_assessment_table = installation_ctx.make_table(schema_name=source_schema.name, ctas="SELECT 1 AS one")
-    installation_ctx.deployed_workflows.run_workflow(workflow, skip_job_wait=False, named_parameters={"force_refresh": "true"})
+    installation_ctx.deployed_workflows.run_workflow(
+        workflow, skip_job_wait=False, named_parameters={"force_refresh": "true"}
+    )
     assert installation_ctx.deployed_workflows.validate_step(workflow), f"Workflow failed: {workflow}"
-    expected_tables = {managed_table.name, external_table.name, tmp_table.name, view.name, non_delta.name, post_assessment_table.name}
+    expected_tables = {
+        managed_table.name,
+        external_table.name,
+        tmp_table.name,
+        view.name,
+        non_delta.name,
+        post_assessment_table.name,
+    }
 
     query = f"SELECT * FROM {installation_ctx.inventory_database}.tables"
     tables_after_assessment_rerun = [table.name for table in sql_backend.fetch(query)]