Support download output of pipeline containing settings attribute (Azure#28962)

* add name and version for output

* add asset_name, asset_version, merge 2022-12-01 pr

* resolve comments

* add test and modify related codes

* refine code

* fix failed test

* fix comments

* fix condition._port_name

* fix test and some comments

* revert the modification of compute

* fix comments

* fix pylint error, add tests for CommandJob, Spark, Parallel, Sweep

* delete one test, will add in another pr

* modify error message

* fix settings, add ut

* run black

* add info in changelog

---------

Co-authored-by: chenyangliao <[email protected]>

Author: lllilithyang

sdk/ml/azure-ai-ml/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features Added
 - Added support for `tags` on Compute Resources.
 - Added support for promoting data asset from a workspace to a registry
+- Added support for registering named asset from job output by specifying named and version settings.
 
 ### Bugs Fixed
 

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

@@ -97,7 +97,7 @@ def settings(self) -> PipelineJobSettings:
 
     @settings.setter
     def settings(self, value):
-        if value is not None and not isinstance(value, PipelineJobSettings):
+        if value and not isinstance(value, PipelineJobSettings):
             raise TypeError("settings must be PipelineJobSettings or dict but got {}".format(type(value)))
         self._settings = value
 

sdk/ml/azure-ai-ml/tests/job_common/unittests/test_job_operations.py

@@ -147,6 +147,32 @@ def test_get(self, mock_method, mock_job_operation: JobOperations) -> None:
         mock_job_operation.get("randon_name")
         mock_job_operation._operation_2022_12_preview.get.assert_called_once()
 
+    @patch.object(JobOperations, "_get_job")
+    def test_get_job(self, mock_method, mock_job_operation: JobOperations) -> None:
+        from azure.ai.ml import Input, dsl, load_component
+
+        component = load_component(source="./tests/test_configs/components/helloworld_component.yml")
+        component_input = Input(type="uri_file", path="https://dprepdata.blob.core.windows.net/demo/Titanic.csv")
+
+        @dsl.pipeline()
+        def sub_pipeline():
+            node = component(component_in_path=component_input)
+
+        @dsl.pipeline()
+        def register_both_output():
+            sub_node = sub_pipeline()
+
+        pipeline = register_both_output()
+        pipeline.settings.default_compute = "cpu-cluster"
+        pipeline.jobs["sub_node"]._component = "fake_component"
+
+        # add settings for subgraph node to simulate the result of getting pipeline that submitted with previous sdk
+        pipeline.jobs["sub_node"]["settings"] = {}
+
+        pipeline_job_base = pipeline._to_rest_object()
+        mock_method.return_value = pipeline_job_base
+        mock_job_operation.get(name="random_name")
+
     @patch.object(Job, "_from_rest_object")
     @patch.dict(os.environ, {AZUREML_PRIVATE_FEATURES_ENV_VAR: "True"})
     def test_get_private_preview_flag_returns_latest(self, mock_method, mock_job_operation: JobOperations) -> None: