[ML][Pipelines] Test: enable pipelines registry consumption tests (Azure#28177)

* test: try to enable test_pipeline_job_create_with_registries

* update recording for test_pipeline_job_create_with_registries

* update test to avoid registry name get sanitized

* update format

* test: enable test_pipeline_job_create_with_registry_model_as_input

* remove unused import

* test: enable test_pipeline_job_create_with_registered_component_on_registry

* update skipif location

* update tests

* add yield in mock_component_hash

Author: zhengfeiwang

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

@@ -1,5 +1,4 @@
 import re
-import tempfile
 import uuid
 from itertools import tee
 from pathlib import Path

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

@@ -570,6 +570,7 @@ def mock_component_hash(mocker: MockFixture, request: FixtureRequest):
     """
     # do nothing if in live mode and not recording
     if is_live_and_not_recording():
+        yield
         return
 
     if is_live():

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

@@ -1,134 +1,97 @@
 from pathlib import Path
 
 import pytest
-from devtools_testutils import AzureRecordedTestCase
-from test_utilities.utils import _PYTEST_TIMEOUT_METHOD, assert_job_cancel
+from azure.core.exceptions import ResourceNotFoundError
 
 from azure.ai.ml import Input, MLClient, load_component, load_model
 from azure.ai.ml.constants import AssetTypes
 from azure.ai.ml.dsl import pipeline
-from azure.core.exceptions import HttpResponseError, ResourceNotFoundError
-
+from devtools_testutils import AzureRecordedTestCase, is_live
+from test_utilities.utils import _PYTEST_TIMEOUT_METHOD, assert_job_cancel
 from .._util import _DSL_TIMEOUT_SECOND
 
 
+def assert_pipeline_job_cancel(client: MLClient, score_func, pipeline_model_input, pipeline_test_data):
+    @pipeline
+    def score_pipeline(model_input, test_data):
+        score = score_func(model_input=model_input, test_data=test_data)  # noqa: F841
+        score_duplicate = score_func(model_input=model_input, test_data=test_data)  # noqa: F841
+
+    pipeline_job = score_pipeline(model_input=pipeline_model_input, test_data=pipeline_test_data)
+    pipeline_job.settings.default_compute = "cpu-cluster"
+    assert_job_cancel(pipeline_job, client)
+
+
+@pytest.mark.skipif(condition=not is_live(), reason="registry test, may fail in playback mode")
 @pytest.mark.usefixtures("enable_pipeline_private_preview_features", "recorded_test")
 @pytest.mark.timeout(timeout=_DSL_TIMEOUT_SECOND, method=_PYTEST_TIMEOUT_METHOD)
 @pytest.mark.e2etest
 @pytest.mark.pipeline_test
 class TestDSLPipelineOnRegistry(AzureRecordedTestCase):
-    @pytest.mark.skip(reason="not able to re-record")
-    def test_pipeline_job_create_with_registered_component_on_registry(
-        self,
-        registry_client: MLClient,
-    ) -> None:
+    test_data = Input(
+        type=AssetTypes.URI_FILE,
+        path="./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/data/sample1.csv"
+    )
+
+    def test_pipeline_job_create_with_registered_component_on_registry(self, pipelines_registry_client: MLClient):
         local_component = load_component("./tests/test_configs/components/basic_component_code_local_path.yml")
         try:
-            created_component = registry_client.components.get(local_component.name, version=local_component.version)
-        except HttpResponseError:
-            created_component = registry_client.components.create_or_update(local_component)
+            created_component = pipelines_registry_client.components.get(
+                local_component.name, version=local_component.version
+            )
+        except ResourceNotFoundError:
+            created_component = pipelines_registry_client.components.create_or_update(local_component)
 
-        @pipeline()
+        @pipeline
         def sample_pipeline():
-            node = created_component()
-            node.compute = "cpu-cluster"
+            created_component()
 
         pipeline_job = sample_pipeline()
-        assert registry_client.jobs.validate(pipeline_job).passed
-        # TODO: add test for pipeline job create with registered component on registry after support is ready on canary
-
-    @pytest.mark.skip(reason="request body still exits when re-record and will raise error "
-                             "'Unable to find a record for the request' in playback mode")
-    def test_pipeline_with_local_component_and_registry_model_as_input(self, registry_client: MLClient, client: MLClient):
-        # get dataset
-        test_data = Input(
-            type=AssetTypes.URI_FILE,
-            path="./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/data/sample1.csv"
-        )
+        pipeline_job.settings.default_compute = "cpu-cluster"
+        assert pipelines_registry_client.jobs.validate(pipeline_job).passed
 
+    # this test will break in playback mode, so include it in live test only
+    def test_pipeline_with_local_component_and_registry_model_as_input(self, client: MLClient):
         # load_component
         score_func = load_component("./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/score.yml")
 
         pipeline_score_model = Input(
-            type='mlflow_model',
-            path='azureml://registries/testFeed/models/iris_model/versions/1'
+            type="custom_model", path="azureml://registries/sdk-test/models/iris_model/versions/1"
         )
 
-        @pipeline()
-        def score_pipeline_with_registry_model(model_input, test_data):
-            score = score_func(model_input=model_input, test_data=test_data)
-            score_duplicate = score_func(model_input=pipeline_score_model, test_data=test_data)
-
-        pipeline_job = score_pipeline_with_registry_model(
-            model_input=pipeline_score_model,
-            test_data=test_data
-        )
-        pipeline_job.settings.default_compute = "cpu-cluster"
-        assert_job_cancel(pipeline_job, client)
+        assert_pipeline_job_cancel(client, score_func, pipeline_score_model, self.test_data)
 
-    @pytest.mark.skip(reason="request body still exits when re-record and will raise error "
-                             "'Unable to find a record for the request' in playback mode")
     def test_pipeline_with_local_component_and_registry_model_as_input_with_model_input(
-            self,
-            registry_client: MLClient,
-            client: MLClient):
-        # get dataset
-        test_data = Input(
-            type=AssetTypes.URI_FILE,
-            path="./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/data/sample1.csv"
-        )
-
+        self, client: MLClient, pipelines_registry_client: MLClient
+    ):
         # load_component
         score_func = load_component("./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/score.yml")
 
         model_path = Path("./tests/test_configs/model/model_iris.yml")
         model_entity = load_model(model_path)
         try:
-            pipeline_score_model = registry_client.models.get(name=model_entity.name, version=model_entity.version)
+            pipeline_score_model = pipelines_registry_client.models.get(
+                name=model_entity.name, version=model_entity.version
+            )
         except ResourceNotFoundError:
-            model_entity = registry_client.models.create_or_update(model_entity)
-            pipeline_score_model = registry_client.models.get(name=model_entity.name, version=model_entity.version)
+            model_entity = pipelines_registry_client.models.create_or_update(model_entity)
+            pipeline_score_model = pipelines_registry_client.models.get(
+                name=model_entity.name, version=model_entity.version
+            )
 
-        @pipeline()
-        def score_pipeline_with_registry_model(model_input, test_data):
-            score = score_func(model_input=model_input, test_data=test_data)
-            score_duplicate = score_func(model_input=pipeline_score_model, test_data=test_data)
-
-        pipeline_job = score_pipeline_with_registry_model(
-            model_input=pipeline_score_model, test_data=test_data
-        )
-        pipeline_job.settings.default_compute = "cpu-cluster"
-        assert_job_cancel(pipeline_job, client)
-
-    @pytest.mark.skip(reason="request body still exits when re-record and will raise error "
-                             "'Unable to find a record for the request' in playback mode")
-    def test_pipeline_with_registry_component_and_model_as_input(self, registry_client: MLClient, client: MLClient):
-        # get dataset
-        test_data = Input(
-            type=AssetTypes.URI_FILE,
-            path="./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/data/sample1.csv"
-        )
+        assert_pipeline_job_cancel(client, score_func, pipeline_score_model, self.test_data)
 
+    def test_pipeline_with_registry_component_and_model_as_input(
+        self, client: MLClient, pipelines_registry_client: MLClient
+    ):
         # load_component
-        score_component_name = "v2_dsl_score_component"
-        component_version = "0.0.8"
-        score_func = registry_client.components.get(
+        score_component_name, component_version = "score_component", "2"
+        score_func = pipelines_registry_client.components.get(
             name=score_component_name, version=component_version
         )
 
         pipeline_score_model = Input(
-            type='mlflow_model',
-            path='azureml://registries/testFeed/models/iris_model/versions/1'
+            type="mlflow_model", path="azureml://registries/sdk-test/models/iris_model/versions/1"
         )
 
-        @pipeline()
-        def score_pipeline_with_registry_model(model_input, test_data):
-            score = score_func(model_input=model_input, test_data=test_data)
-            score_duplicate = score_func(model_input=pipeline_score_model, test_data=test_data)
-
-        pipeline_job = score_pipeline_with_registry_model(
-            model_input=pipeline_score_model,
-            test_data=test_data
-        )
-        pipeline_job.settings.default_compute = "cpu-cluster"
-        assert_job_cancel(pipeline_job, client)
+        assert_pipeline_job_cancel(client, score_func, pipeline_score_model, self.test_data)

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

@@ -76,26 +76,22 @@ 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")
+    @pytest.mark.skipif(condition=not is_live(), reason="registry test, may fail in playback mode")
     def test_pipeline_job_create_with_registries(
-        self,
-        client: MLClient,
-        randstr: Callable[[str], str],
+        self, client: MLClient, randstr: Callable[[str], str]
     ) -> None:
         params_override = [{"name": randstr("name")}]
         pipeline_job = load_job(
             source="./tests/test_configs/pipeline_jobs/hello_pipeline_job_with_registries.yml",
             params_override=params_override,
         )
-        assert (
-            pipeline_job.jobs.get("a").environment
-            == "azureml://registries/testFeed/environments/sklearn-10-ubuntu2004-py38-cpu/versions/19.dev6"
-        )
-        job = client.jobs.create_or_update(pipeline_job)
+        # registry sdk-test may be sanitized as other name, so use two assertions to avoid this issue
+        assert str(pipeline_job.jobs["a"].environment).startswith("azureml://registries/")
+        assert str(pipeline_job.jobs["a"].environment).endswith("/environments/openMPIUbuntu/versions/1")
+        job = assert_job_cancel(pipeline_job, client)
         assert job.name == params_override[0]["name"]
-        assert (
-            job.jobs.get("a").component == "azureml://registries/testFeed/components/my_hello_world_asset_2/versions/1"
-        )
+        assert str(job.jobs["a"].component).startswith("azureml://registries/")
+        assert str(job.jobs["a"].component).endswith("/components/hello_world_asset/versions/1")
 
     @pytest.mark.parametrize(
         "pipeline_job_path",
@@ -980,6 +976,7 @@ def test_pipeline_job_with_automl_image_multiclass_classification(
             "sweep": {
                 "sampling_algorithm": "random",
                 "early_termination": {
+                    "evaluation_interval": 10,
                     "evaluation_interval": 10,
                     "delay_evaluation": 0,
                     "type": "bandit",
@@ -1324,22 +1321,14 @@ def test_remote_pipeline_component_job(self, client: MLClient, randstr: Callable
         # assert pipeline_dict["outputs"] == {"output_path": {"mode": "ReadWriteMount", "job_output_type": "uri_folder"}}
         assert pipeline_dict["settings"] == {"default_compute": "cpu-cluster", "_source": "REMOTE.WORKSPACE.COMPONENT"}
 
-    @pytest.mark.skip(
-        reason="request body still exits when re-record and will raise error "
-        "'Unable to find a record for the request' in playback mode"
-    )
-    def test_pipeline_job_create_with_registry_model_as_input(
-        self,
-        client: MLClient,
-        registry_client: MLClient,
-        randstr: Callable[[str], str],
-    ) -> None:
+    @pytest.mark.skipif(condition=not is_live(), reason="registry test, may fail in playback mode")
+    def test_pipeline_job_create_with_registry_model_as_input(self, client: MLClient, randstr: Callable[[str], str]):
         params_override = [{"name": randstr("name")}]
         pipeline_job = load_job(
             source="./tests/test_configs/pipeline_jobs/job_with_registry_model_as_input/pipeline.yml",
             params_override=params_override,
         )
-        job = client.jobs.create_or_update(pipeline_job)
+        job = assert_job_cancel(pipeline_job, client)
         assert job.name == params_override[0]["name"]
 
     def test_pipeline_node_with_default_component(self, client: MLClient, randstr: Callable[[str], str]):