[Do_while] Fix pipeline component with do-while operator cannot be submitted. (Azure#27063)

* remove empty mapping check

* fix error

* fix test case

* fix comment

Author: lalala123123

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_builders/do_while.py

@@ -101,7 +101,7 @@ def get_port_obj(body, port_name, is_input=True, validate_port=True):
                 port = body.inputs.get(port_name, None)
             else:
                 port = body.outputs.get(port_name, None)
-            if not port:
+            if port is None:
                 if validate_port:
                     raise ValidationError(
                         message=f"Cannot find {port_name} in do_while loop body {'inputs' if is_input else 'outputs'}.",
@@ -229,7 +229,7 @@ def _validate_loop_condition(self, raise_error=True):
             if validation_result.passed:
                 # Check condition is a control output.
                 condition_name = self.condition if isinstance(self.condition, str) else self.condition._name
-                if not self.body.component.outputs[condition_name].is_control:
+                if not self.body._outputs[condition_name].is_control:
                     validation_result.append_error(
                         yaml_path="condition",
                         message=(
@@ -251,18 +251,14 @@ def _validate_do_while_limit(self, raise_error=True):
         elif self.limits.max_iteration_count > DO_WHILE_MAX_ITERATION or self.limits.max_iteration_count < 0:
             validation_result.append_error(
                 yaml_path="limit.max_iteration_count",
-                message=f"The max iteration count cannot be less than 0 and larger than {DO_WHILE_MAX_ITERATION}.",
+                message=f"The max iteration count cannot be less than 0 or larger than {DO_WHILE_MAX_ITERATION}.",
             )
         return validation_result.try_raise(self._get_validation_error_target(), raise_error=raise_error)
 
     def _validate_body_output_mapping(self, raise_error=True):
         # pylint disable=protected-access
         validation_result = self._create_empty_validation_result()
-        if not self.mapping:
-            validation_result.append_error(
-                yaml_path="mapping", message="The mapping of body output to input cannot be empty."
-            )
-        elif not isinstance(self.mapping, dict):
+        if not isinstance(self.mapping, dict):
             validation_result.append_error(
                 yaml_path="mapping", message=f"Mapping expects a dict type but passes in a {type(self.mapping)} type."
             )
@@ -277,7 +273,7 @@ def _validate_body_output_mapping(self, raise_error=True):
                     output, self.body.outputs, port_type="output", yaml_path="mapping"
                 )
                 if validate_results.passed:
-                    is_control_output = self.body.component.outputs[output_name].is_control
+                    is_control_output = self.body._outputs[output_name].is_control
                     inputs = inputs if isinstance(inputs, list) else [inputs]
                     for item in inputs:
                         input_validate_results = self._validate_port(
@@ -287,11 +283,12 @@ def _validate_body_output_mapping(self, raise_error=True):
                         # pylint: disable=protected-access
                         input_name = item if isinstance(item, str) else item._name
                         input_output_mapping[input_name] = input_output_mapping.get(input_name, []) + [output_name]
+                        is_primitive_type = self.body._inputs[input_name]._meta._is_primitive_type
 
                         if (
                             input_validate_results.passed
                             and not is_control_output
-                            and self.body.component.inputs[input_name]._is_primitive_type # pylint: disable=protected-access
+                            and is_primitive_type  # pylint: disable=protected-access
                         ):
                             validate_results.append_error(
                                 yaml_path="mapping",

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

@@ -0,0 +1,61 @@
+import pytest
+from typing import Callable
+from devtools_testutils import AzureRecordedTestCase
+from test_utilities.utils import _PYTEST_TIMEOUT_METHOD
+
+from azure.ai.ml import MLClient, load_job
+from azure.ai.ml._utils.utils import load_yaml
+from azure.ai.ml._schema.pipeline import pipeline_job
+from azure.ai.ml.entities._builders import Command, Pipeline
+from azure.ai.ml.entities._builders.do_while import DoWhile
+
+from .._util import _PIPELINE_JOB_TIMEOUT_SECOND
+from .test_pipeline_job import assert_job_cancel
+
+
+@pytest.fixture()
+def update_pipeline_schema():
+    # Update the job type that the pipeline is supported.
+    schema = pipeline_job.PipelineJobSchema
+    schema._declared_fields['jobs'] = pipeline_job.PipelineJobsField()
+
+
+@pytest.mark.usefixtures(
+    "recorded_test",
+    "mock_code_hash",
+    "enable_pipeline_private_preview_features",
+    "update_pipeline_schema",
+    "mock_asset_name",
+    "mock_component_hash",
+)
+@pytest.mark.timeout(timeout=_PIPELINE_JOB_TIMEOUT_SECOND, method=_PYTEST_TIMEOUT_METHOD)
+@pytest.mark.e2etest
+@pytest.mark.pipeline_test
+class TestConditionalNodeInPipeline(AzureRecordedTestCase):
+    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",
+            params_override=params_override,
+        )
+        created_pipeline = assert_job_cancel(pipeline_job, client)
+        assert len(created_pipeline.jobs) == 5
+        assert isinstance(created_pipeline.jobs["pipeline_body_node"], Pipeline)
+        assert isinstance(created_pipeline.jobs["do_while_job_with_pipeline_job"], DoWhile)
+        assert isinstance(created_pipeline.jobs["do_while_job_with_command_component"], DoWhile)
+        assert isinstance(created_pipeline.jobs["command_component_body_node"], Command)
+        assert isinstance(created_pipeline.jobs["get_do_while_result"], Command)
+
+    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",
+            params_override=params_override,
+        )
+        created_pipeline = assert_job_cancel(pipeline_job, client)
+        assert len(created_pipeline.jobs) == 5
+        assert isinstance(created_pipeline.jobs["pipeline_body_node"], Pipeline)
+        assert isinstance(created_pipeline.jobs["do_while_job_with_pipeline_job"], DoWhile)
+        assert isinstance(created_pipeline.jobs["do_while_job_with_command_component"], DoWhile)
+        assert isinstance(created_pipeline.jobs["command_component_body_node"], Command)
+        assert isinstance(created_pipeline.jobs["get_do_while_result"], Command)

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

@@ -1416,63 +1416,6 @@ def test_pipeline_with_pipeline_component(self, client: MLClient, randstr: Calla
             "_source": "YAML.JOB",
         }
 
-    @pytest.mark.skip(reason="Currently do_while only enable in master region.")
-    def test_pipeline_with_do_while_node(self, client: MLClient, randstr: Callable[[], str]) -> None:
-        params_override = [{"name": randstr()}]
-        pipeline_job = load_job(
-            "./tests/test_configs/dsl_pipeline/pipeline_with_do_while/pipeline.yml",
-            params_override=params_override,
-        )
-        created_pipeline = assert_job_cancel(pipeline_job, client)
-        assert len(created_pipeline.jobs) == 5
-        assert isinstance(created_pipeline.jobs["pipeline_body_node"], Pipeline)
-        assert isinstance(created_pipeline.jobs["do_while_job_with_pipeline_job"], DoWhile)
-        assert isinstance(created_pipeline.jobs["do_while_job_with_command_component"], DoWhile)
-        assert isinstance(created_pipeline.jobs["command_component_body_node"], Command)
-        assert isinstance(created_pipeline.jobs["get_do_while_result"], Command)
-
-    @pytest.mark.skip(reason="Currently not enable submit a pipeline with primitive inputs")
-    def test_do_while_pipeline_with_primitive_inputs(self, client: MLClient, randstr: Callable[[], str]) -> None:
-        params_override = [{"name": randstr()}]
-        pipeline_job = load_job(
-            path="./tests/test_configs/dsl_pipeline/pipeline_with_do_while/pipeline_with_primitive_inputs.yml",
-            params_override=params_override,
-        )
-        created_pipeline = assert_job_cancel(pipeline_job, client)
-        assert len(created_pipeline.jobs) == 5
-        assert isinstance(created_pipeline.jobs["pipeline_body_node"], Pipeline)
-        assert isinstance(created_pipeline.jobs["do_while_job_with_pipeline_job"], DoWhile)
-        assert isinstance(created_pipeline.jobs["do_while_job_with_command_component"], DoWhile)
-        assert isinstance(created_pipeline.jobs["command_component_body_node"], Command)
-        assert isinstance(created_pipeline.jobs["get_do_while_result"], Command)
-
-    @pytest.mark.skip(reason="Currently do_while only enable in master region.")
-    def test_pipeline_with_invalid_do_while_node(self, client: MLClient, randstr: Callable[[], str]) -> None:
-        params_override = [{"name": randstr()}]
-        with pytest.raises(ValidationError) as exception:
-            load_job(
-                "./tests/test_configs/dsl_pipeline/pipeline_with_do_while/invalid_pipeline.yml",
-                params_override=params_override,
-            )
-        error_message_str = re.findall(r"(\{.*\})", exception.value.args[0].replace("\n", ""))[0]
-        error_messages = json.loads(error_message_str.replace("\\", "\\\\"))
-
-        def assert_error_message(path, except_message, error_messages):
-            msgs = next(filter(lambda item: item["path"] == path, error_messages))
-            assert except_message == msgs["message"]
-
-        assert_error_message("jobs.empty_mapping.mapping", "Missing data for required field.", error_messages["errors"])
-        assert_error_message(
-            "jobs.out_of_range_max_iteration_count.limits.max_iteration_count",
-            "Must be greater than or equal to 1 and less than or equal to 1000.",
-            error_messages["errors"],
-        )
-        assert_error_message(
-            "jobs.invalid_max_iteration_count.limits.max_iteration_count",
-            "Not a valid integer.",
-            error_messages["errors"],
-        )
-
     def test_pipeline_component_job(self, client: MLClient):
         test_path = "./tests/test_configs/pipeline_jobs/pipeline_component_job.yml"
         job: PipelineJob = load_job(source=test_path)

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

@@ -1,3 +1,5 @@
+import re
+import json
 from pathlib import Path
 from unittest.mock import patch
 
@@ -12,6 +14,7 @@
 from azure.ai.ml.exceptions import ValidationException
 
 from .._util import _PIPELINE_JOB_TIMEOUT_SECOND
+from ..e2etests.test_control_flow_node_in_pipeline_job import update_pipeline_schema
 
 
 def assert_the_same_path(actual_path, expected_path):
@@ -637,3 +640,31 @@ def pipeline_with_compute_binding(compute_name: str):
         pipeline_job = pipeline_with_compute_binding('cpu-cluster')
         # Assert compute binding validate not raise error when validate
         assert pipeline_job._validate().passed
+
+    @pytest.mark.usefixtures(
+        "enable_pipeline_private_preview_features",
+        "update_pipeline_schema"
+    )
+    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",
+            )
+        error_message_str = re.findall(r"(\{.*\})", exception.value.args[0].replace("\n", ""))[0]
+        error_messages = json.loads(error_message_str.replace("\\", "\\\\"))
+
+        def assert_error_message(path, except_message, error_messages):
+            msgs = next(filter(lambda item: item["path"] == path, error_messages))
+            assert except_message == msgs["message"]
+
+        assert_error_message("jobs.empty_mapping.mapping", "Missing data for required field.", error_messages["errors"])
+        assert_error_message(
+            "jobs.out_of_range_max_iteration_count.limits.max_iteration_count",
+            "Must be greater than or equal to 1 and less than or equal to 1000.",
+            error_messages["errors"],
+        )
+        assert_error_message(
+            "jobs.invalid_max_iteration_count.limits.max_iteration_count",
+            "Not a valid integer.",
+            error_messages["errors"],
+        )