[ML][Pipelines] Always cancel submitted pipeline jobs (Azure#27638)

* fix: use another registry for pipelines registry tests

* test: always cancel job for `assert_job_cancel`

* test: remove some fixtures for live only test

Author: zhengfeiwang

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

@@ -325,7 +325,7 @@ def pipelines_registry_client(e2e_ws_scope: OperationScope, auth: ClientSecretCr
     return MLClient(
         credential=auth,
         logging_enable=getenv(E2E_TEST_LOGGING_ENABLED),
-        registry_name="sdk-canary",
+        registry_name="sdk-test",
     )
 
 

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

@@ -1360,27 +1360,20 @@ def test_pipeline_node_with_default_component(self, client: MLClient, randstr: C
         )
 
 
-
-@pytest.mark.usefixtures(
-    "recorded_test",
-    "mock_code_hash",
-    "enable_pipeline_private_preview_features",
-    "mock_asset_name",
-    "mock_component_hash",
-    "enable_environment_id_arm_expansion",
-)
-@pytest.mark.timeout(timeout=_PIPELINE_JOB_LONG_RUNNING_TIMEOUT_SECOND, method=_PYTEST_TIMEOUT_METHOD)
+@pytest.mark.usefixtures("enable_pipeline_private_preview_features")
 @pytest.mark.e2etest
 @pytest.mark.pipeline_test
 @pytest.mark.skipif(condition=not is_live(), reason="no need to run in playback mode")
-class TestPipelineJobLongRunning(AzureRecordedTestCase):
+@pytest.mark.timeout(timeout=_PIPELINE_JOB_LONG_RUNNING_TIMEOUT_SECOND, method=_PYTEST_TIMEOUT_METHOD)
+class TestPipelineJobLongRunning:
     """Long-running tests that require pipeline job completed."""
     def test_pipeline_job_get_child_run(self, client: MLClient, randstr: Callable[[str], str]):
         pipeline_job = load_job(
             source="./tests/test_configs/pipeline_jobs/helloworld_pipeline_job_quick_with_output.yml",
             params_override=[{"name": randstr("name")}],
         )
         job = client.jobs.create_or_update(pipeline_job)
+        print("pipeline job name:", job.name)
         wait_until_done(client, job)
         child_job = next(
             job
@@ -1400,6 +1393,7 @@ def test_pipeline_job_download(
                 params_override=[{"name": randstr("job_name")}],
             )
         )
+        print("pipeline job name:", job.name)
         wait_until_done(client, job)
         client.jobs.download(name=job.name, download_path=tmp_path)
         artifact_dir = tmp_path / "artifacts"
@@ -1415,6 +1409,7 @@ def test_pipeline_job_child_run_download(
                 params_override=[{"name": randstr("job_name")}],
             )
         )
+        print("pipeline job name:", job.name)
         wait_until_done(client, job)
         child_job = next(
             job
@@ -1434,7 +1429,6 @@ def test_pipeline_job_child_run_download(
         assert output_dir.exists()
         assert next(output_dir.iterdir(), None), "No artifacts were downloaded"
 
-    @pytest.mark.disable_mock_code_hash
     def test_reused_pipeline_child_job_download(
         self,
         client: MLClient,
@@ -1445,15 +1439,15 @@ def test_reused_pipeline_child_job_download(
 
         # ensure previous job exists for reuse
         job_name = randstr("job_name")
-        print(f"previous job name: {job_name}")
+        print("previous job name:", job_name)
         previous_job = client.jobs.create_or_update(
             load_job(source=pipeline_spec_path, params_override=[{"name": job_name}])
         )
         wait_until_done(client, previous_job)
 
         # submit a new job that will reuse previous job
         new_job_name = randstr("new_job_name")
-        print(f"new job name: {new_job_name}")
+        print("new job name:", new_job_name)
         new_job = client.jobs.create_or_update(
             load_job(pipeline_spec_path, params_override=[{"name": new_job_name}]),
         )

sdk/ml/azure-ai-ml/tests/test_utilities/utils.py

@@ -292,7 +292,7 @@ def assert_job_cancel(
     created_job = client.jobs.create_or_update(job, experiment_name=experiment_name)
     if check_before_cancelled is not None:
         assert check_before_cancelled(created_job)
-        cancel_job(client, created_job)
+    cancel_job(client, created_job)
     return created_job