[Workflows] Add subpath or workdir to workflow path (#5461)

mlrun/projects/pipelines.py

@@ -522,7 +522,7 @@ def get_state(run_id, project=None):
     @staticmethod
     def _get_handler(workflow_handler, workflow_spec, project, secrets):
         if not (workflow_handler and callable(workflow_handler)):
-            workflow_file = workflow_spec.get_source_file(project.spec.context)
+            workflow_file = workflow_spec.get_source_file(project.spec.get_code_path())
             workflow_handler = create_pipeline(
                 project,
                 workflow_file,
@@ -554,7 +554,7 @@ class _KFPRunner(_PipelineRunner):
     @classmethod
     def save(cls, project, workflow_spec: WorkflowSpec, target, artifact_path=None):
         pipeline_context.set(project, workflow_spec)
-        workflow_file = workflow_spec.get_source_file(project.spec.context)
+        workflow_file = workflow_spec.get_source_file(project.spec.get_code_path())
         functions = FunctionsDict(project)
         pipeline = create_pipeline(
             project,

tests/projects/test_project.py

@@ -2227,6 +2227,29 @@ def test_load_project_dir(project_file_name, expectation):
         shutil.rmtree(project_dir)
 
 
+def test_workflow_path_with_project_workdir():
+    project_name = "project1"
+
+    project = mlrun.new_project(project_name, save=False, context="./context")
+    workflow_path = "workflow.py"
+    workflow_spec = mlrun.projects.pipelines.WorkflowSpec(path=workflow_path)
+    # with_out_workdir
+    path = workflow_spec.get_source_file(project.spec.get_code_path())
+    assert path == "./context/workflow.py"
+
+    # with__subpath
+    project = mlrun.new_project(
+        project_name, save=False, context="./context", subpath="./subpath"
+    )
+    path = workflow_spec.get_source_file(project.spec.get_code_path())
+    assert path == "./context/./subpath/workflow.py"
+
+    # with__workdir
+    project.spec.workdir = "./workdir"
+    path = workflow_spec.get_source_file(project.spec.get_code_path())
+    assert path == "./context/./workdir/workflow.py"
+
+
 class TestModelMonitoring:
     """Test model monitoring project methods"""