Added error handling and tests for Workflow linter during pipeline fetch (#3819)

## Changes
Added error handling while fetching pipelines in workflow linting.

### Linked issues
Resolves #3815

### Functionality

- [ ] added error handling

### Tests
<!-- How is this tested? Please see the checklist below and also
describe any other relevant tests -->

- [ ] manually tested
- [x] added unit tests
- [x] added integration tests
- [ ] verified on staging environment (screenshot attached)

Author: pritishpai

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

@@ -11,7 +11,7 @@
 from databricks.labs.blueprint.paths import DBFSPath
 from databricks.sdk import WorkspaceClient
 from databricks.sdk.errors import ResourceDoesNotExist, BadRequest, InvalidParameterValue, DatabricksError
-from databricks.sdk.service import compute, jobs
+from databricks.sdk.service import compute, jobs, pipelines
 from databricks.sdk.service.compute import DataSecurityMode
 from databricks.sdk.service.jobs import Source
 
@@ -286,24 +286,31 @@ def _register_pipeline_task(self, graph: DependencyGraph) -> Iterable[Dependency
         if not self._task.pipeline_task:
             return
 
-        pipeline = self._ws.pipelines.get(self._task.pipeline_task.pipeline_id)
-        if not pipeline.spec:
+        try:
+            pipeline = self._ws.pipelines.get(self._task.pipeline_task.pipeline_id)
+        except ResourceDoesNotExist:
+            yield DependencyProblem(
+                'pipeline-not-found', f'Could not find pipeline: {self._task.pipeline_task.pipeline_id}'
+            )
             return
-        if not pipeline.spec.libraries:
+
+        if not pipeline.spec or not pipeline.spec.libraries:
             return
 
-        pipeline_libraries = pipeline.spec.libraries
-        for library in pipeline_libraries:
-            if not library.notebook:
-                return
-            if library.notebook.path:
-                yield from self._register_notebook_path(graph, library.notebook.path)
-            if library.jar:
-                yield from self._register_library(graph, compute.Library(jar=library.jar))
-            if library.maven:
-                yield DependencyProblem('not-yet-implemented', 'Maven library is not yet implemented')
-            if library.file:
-                yield DependencyProblem('not-yet-implemented', 'File library is not yet implemented')
+        for library in pipeline.spec.libraries:
+            yield from self._register_pipeline_library(graph, library)
+
+    def _register_pipeline_library(
+        self, graph: DependencyGraph, library: pipelines.PipelineLibrary
+    ) -> Iterable[DependencyProblem]:
+        if library.notebook and library.notebook.path:
+            yield from self._register_notebook_path(graph, library.notebook.path)
+        if library.jar:
+            yield from self._register_library(graph, compute.Library(jar=library.jar))
+        if library.maven:
+            yield DependencyProblem('not-yet-implemented', 'Maven library is not yet implemented')
+        if library.file:
+            yield DependencyProblem('not-yet-implemented', 'File library is not yet implemented')
 
     def _register_notebook_path(self, graph: DependencyGraph, notebook_path: str) -> Iterable[DependencyProblem]:
         try:

tests/integration/source_code/test_jobs.py

@@ -519,6 +519,16 @@ def test_job_dlt_task_linter_unhappy_path(
         len([problem for problem in problems if problem.message == "Could not locate import: library_not_found"]) == 1
     )
 
+    # Pipeline id does not exist
+    task = jobs.Task(
+        task_key=make_random(4),
+        pipeline_task=jobs.PipelineTask(pipeline_id="1234"),
+    )
+    j = make_job(tasks=[task])
+
+    problems, *_ = simple_ctx.workflow_linter.lint_job(j.job_id)
+    assert len([problem for problem in problems if problem.message == "Could not find pipeline: 1234"]) == 1
+
 
 def test_job_dlt_task_linter_happy_path(
     simple_ctx,

tests/unit/source_code/test_jobs.py

@@ -34,6 +34,7 @@
 from databricks.labs.ucx.source_code.notebooks.loaders import NotebookLoader, NotebookResolver
 from databricks.labs.ucx.source_code.python_libraries import PythonLibraryResolver
 from databricks.labs.ucx.source_code.used_table import UsedTablesCrawler
+from databricks.sdk.errors.platform import ResourceDoesNotExist
 
 
 def test_job_problem_as_message() -> None:
@@ -486,6 +487,14 @@ def test_workflow_linter_dlt_pipeline_task(graph) -> None:
     assert len(problems) == 4
     ws.assert_not_called()
 
+    task = jobs.Task(task_key="test", pipeline_task=jobs.PipelineTask(pipeline_id='1234'))
+
+    ws.pipelines.get.side_effect = ResourceDoesNotExist("Could not find pipeline: 1234")
+
+    workflow_task_container = WorkflowTaskContainer(ws, task, Job())
+    problems = workflow_task_container.build_dependency_graph(graph)
+    assert len(problems) == 1
+
 
 def test_xxx(graph) -> None:
     ws = create_autospec(WorkspaceClient)