[ML][Pipelines]Fix CLI bug when control flow in pipeline component (Azure#28428)

* add tests

* re-record

Author: D-W-

sdk/ml/azure-ai-ml/azure/ai/ml/_schema/pipeline/pipeline_component.py

@@ -189,6 +189,11 @@ class _AnonymousPipelineComponentSchema(AnonymousAssetSchema, PipelineComponentS
     def make(self, data, **kwargs):
         from azure.ai.ml.entities._component.pipeline_component import PipelineComponent
 
+        # pipeline jobs post process is required before init of pipeline component: it converts control node dict
+        # to entity.
+        # however @post_load invocation order is not guaranteed, so we need to call it explicitly here.
+        _post_load_pipeline_jobs(self.context, data)
+
         return PipelineComponent(
             base_path=self.context[BASE_PATH_CONTEXT_KEY],
             **data,

sdk/ml/azure-ai-ml/tests/pipeline_job/e2etests/test_control_flow_pipeline.py

@@ -124,7 +124,7 @@ class TestDoWhile(TestConditionalNodeInPipeline):
     def test_pipeline_with_do_while_node(self, client: MLClient, randstr: Callable[[], str]) -> None:
         params_override = [{"name": randstr('name')}]
         pipeline_job = load_job(
-            "./tests/test_configs/dsl_pipeline/pipeline_with_do_while/pipeline.yml",
+            "./tests/test_configs/pipeline_jobs/control_flow/do_while/pipeline.yml",
             params_override=params_override,
         )
         created_pipeline = assert_job_cancel(pipeline_job, client)
@@ -138,7 +138,7 @@ def test_pipeline_with_do_while_node(self, client: MLClient, randstr: Callable[[
     def test_do_while_pipeline_with_primitive_inputs(self, client: MLClient, randstr: Callable[[], str]) -> None:
         params_override = [{"name": randstr('name')}]
         pipeline_job = load_job(
-            "./tests/test_configs/dsl_pipeline/pipeline_with_do_while/pipeline_with_primitive_inputs.yml",
+            "./tests/test_configs/pipeline_jobs/control_flow/do_while/pipeline_with_primitive_inputs.yml",
             params_override=params_override,
         )
         created_pipeline = assert_job_cancel(pipeline_job, client)
@@ -208,3 +208,42 @@ def test_output_binding_foreach_node(self, client: MLClient, randstr: Callable):
             'type': 'parallel_for'
         }
         assert_foreach(client, randstr("job_name"), source, expected_node)
+
+
+def assert_control_flow_in_pipeline_component(client, component_path, pipeline_path):
+    params_override = [{"component": component_path}]
+    pipeline_job = load_job(
+        pipeline_path,
+        params_override=params_override,
+    )
+    created_pipeline = assert_job_cancel(pipeline_job, client)
+    pipeline_job_dict = created_pipeline._to_rest_object().as_dict()
+
+    pipeline_job_dict = omit_with_wildcard(pipeline_job_dict, *omit_fields)
+    assert pipeline_job_dict["properties"]["jobs"] == {}
+
+
+class TestControlFLowPipelineComponent(TestConditionalNodeInPipeline):
+    def test_if_else(self, client: MLClient, randstr: Callable[[], str]):
+        assert_control_flow_in_pipeline_component(
+            client=client,
+            component_path="./if_else/simple_pipeline.yml",
+            pipeline_path="./tests/test_configs/pipeline_jobs/control_flow/control_flow_with_pipeline_component.yml"
+        )
+
+    @pytest.mark.skip(
+        reason="TODO(2177353): check why recorded tests failure."
+    )
+    def test_do_while(self, client: MLClient, randstr: Callable[[], str]):
+        assert_control_flow_in_pipeline_component(
+            client=client,
+            component_path="./do_while/pipeline_component.yml",
+            pipeline_path="./tests/test_configs/pipeline_jobs/control_flow/control_flow_with_pipeline_component.yml"
+        )
+
+    def test_foreach(self, client: MLClient, randstr: Callable[[], str]):
+        assert_control_flow_in_pipeline_component(
+            client=client,
+            component_path="./parallel_for/simple_pipeline.yml",
+            pipeline_path="./tests/test_configs/pipeline_jobs/control_flow/control_flow_with_pipeline_component.yml"
+        )

sdk/ml/azure-ai-ml/tests/pipeline_job/unittests/test_pipeline_job_validate.py

@@ -654,7 +654,7 @@ def pipeline_with_compute_binding(compute_name: str):
     def test_pipeline_with_invalid_do_while_node(self) -> None:
         with pytest.raises(ValidationError) as exception:
             load_job(
-                "./tests/test_configs/dsl_pipeline/pipeline_with_do_while/invalid_pipeline.yml",
+                "./tests/test_configs/pipeline_jobs/control_flow/do_while/invalid_pipeline.yml",
             )
         error_message_str = re.findall(r"(\{.*\})", exception.value.args[0].replace("\n", ""))[0]
         error_messages = json.loads(error_message_str)