[ML][Pipelines] Fix breaking tests in live mode (Azure#27468)

* test: fix reuse test by disabling mock code hash and make these tests only run in live mode

* test: sleep 5s for breaking test in live mode

* test: only enable registry test in master region

* Revert "test: only enable registry test in master region"

This reverts commit a447e48.

* test: skip registry test for now

* test: enable `test_mpi_component` in playback mode

* test: add mark "disable_mock_code_hash" for pipeline reuse test

Author: zhengfeiwang

sdk/ml/azure-ai-ml/tests/component/e2etests/test_component.py

@@ -27,7 +27,7 @@
 from ..unittests.test_component_schema import load_component_entity_from_rest_json
 
 
-from devtools_testutils import AzureRecordedTestCase, is_live, set_bodiless_matcher
+from devtools_testutils import AzureRecordedTestCase, is_live
 
 
 def create_component(
@@ -372,7 +372,7 @@ def test_component_update(self, client: MLClient, randstr: Callable[[str], str])
         assert component_resource.display_name == display_name
 
     @pytest.mark.disable_mock_code_hash
-    @pytest.mark.skipif(condition=not is_live(), reason="non-deterministic upload fails in playback on CI")
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_component_create_twice_same_code_arm_id(
         self, client: MLClient, randstr: Callable[[str], str], tmp_path: Path
     ) -> None:
@@ -424,6 +424,7 @@ def test_component_update_code(self, client: MLClient, randstr: Callable[[str],
             client.components.create_or_update(command_component)
 
     @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_component_create_default_code(self, client: MLClient, randstr: Callable[[str], str]) -> None:
         # step2: test component without code
         component_name = randstr("component_name")
@@ -549,6 +550,7 @@ def test_command_component_create_autoincrement(self, client: MLClient, randstr:
         assert next_component_asset._auto_increment_version is False
 
     @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_anonymous_component_reuse(self, client: MLClient, variable_recorder) -> None:
         # component without code
         component_name_1 = variable_recorder.get_or_record("component_name_1", str(uuid.uuid4()))
@@ -838,6 +840,7 @@ def test_create_pipeline_component_from_job(self, client: MLClient, randstr: Cal
         rest_component = client.components.create_or_update(component)
         assert rest_component.name == name
 
+    @pytest.mark.skipif(condition=not is_live(), reason="registry test, target to verify service-side behavior")
     def test_component_with_default_label(
         self,
         client: MLClient,
@@ -848,6 +851,8 @@ def test_component_with_default_label(
 
         create_component(client, component_name, path=yaml_path)
 
+        sleep_if_live(5)  # sleep 5 seconds to wait for index service update
+
         target_component = client.components.get(component_name, label="latest")
 
         for default_component in [

sdk/ml/azure-ai-ml/tests/conftest.py

@@ -427,7 +427,7 @@ def mock_code_hash(request, mocker: MockFixture) -> None:
     def generate_hash(*args, **kwargs):
         return str(uuid.uuid4())
 
-    if is_live_and_not_recording():
+    if "disable_mock_code_hash" not in request.keywords and is_live_and_not_recording():
         mocker.patch("azure.ai.ml._artifacts._artifact_utilities.get_object_hash", side_effect=generate_hash)
     elif not is_live():
         mocker.patch(

sdk/ml/azure-ai-ml/tests/dsl/e2etests/test_dsl_pipeline.py

@@ -1285,6 +1285,8 @@ def pipeline_with_default_optional_parameters(
         # assert "optional_param_with_default" not in pipeline_job.inputs
         # assert "optional_param_with_default" not in next(pipeline_job.jobs.values().__iter__()).inputs
 
+    @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_component_reuse(self, client: MLClient) -> None:
         path = "./tests/test_configs/components/helloworld_component.yml"
         component_func1 = load_component(source=path)
@@ -1312,10 +1314,8 @@ def pipeline(component_in_number, component_in_path):
 
         assert len(component_ids) == 1, f"Got multiple component id: {component_ids} for same anon component."
 
-    @pytest.mark.skipif(
-        condition=not is_live(),
-        reason="this test targets to verify service-side behaviour, only enable it in live test.",
-    )
+    @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_pipeline_reuse(self, client: MLClient, randstr: Callable[[str], str], randint: Callable) -> None:
         component_yaml = components_dir / "helloworld_component.yml"
         component_func1 = load_component(source=component_yaml, params_override=[{"name": randstr("component_name")}])

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

@@ -93,6 +93,7 @@ def test_pipeline_job_create(
         assert new_tag_name in updated_job.tags
         assert updated_job.tags[new_tag_name] == new_tag_value
 
+    @pytest.mark.skip("skip as registries not work in canary region for now")
     def test_pipeline_job_create_with_registries(
         self,
         client: MLClient,
@@ -660,6 +661,7 @@ def test_pipeline_job_create_with_distribution_component(
             )
 
     @pytest.mark.disable_mock_code_hash
+    @pytest.mark.skipif(condition=not is_live(), reason="reuse test, target to verify service-side behavior")
     def test_pipeline_job_anonymous_component_reuse(
         self,
         client: MLClient,