Check 1p component support register NodeOutput (Azure#28807)

* 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

* add e2e test

* update recordings

* update recording test_case2

* test reuse when don't write version for output

* tmp commit

* revert test_register_output_sdk_succeed, add modification in another pr

* add new line

* run black with 22.3.0

---------

Co-authored-by: chenyangliao <chenyangliao@microsoft.com>

Author: lllilithyang

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

@@ -27,6 +27,21 @@
 
 _dependent_datasets = {}
 
+PARAMETERS_TO_TEST_WITH_OUTPUT_OUTPUT_RELATED = [
+    ("output_path", "Distributed_output", "1"),
+    ("scored_dataset", "Parallel_output", "1"),
+    ("SSPath", "Scope_output", "1"),
+    ("output_path", "HDInsight_output", "1"),
+    ("output1", "Hemera_output", "1"),
+    ("destination_data", "DataTransfer_output", "1"),
+    ("CosmosPath", "Starlite_output", "1"),
+    ("outputfolderEnc", "Ae365expool_output", "1"),
+]
+PARAMETERS_TO_TEST_WITH_OUTPUT = []
+# don't use PARAMETERS_TO_TEST[0] because this component doesn't have output
+for ori_tuple, output_tuple in zip(PARAMETERS_TO_TEST[1:], PARAMETERS_TO_TEST_WITH_OUTPUT_OUTPUT_RELATED):
+    PARAMETERS_TO_TEST_WITH_OUTPUT.append(ori_tuple + output_tuple)
+
 
 @pytest.fixture
 def create_internal_sample_dependent_datasets(client: MLClient):
@@ -97,6 +112,47 @@ def pipeline_func():
             json.dumps(node_rest_dict, indent=2), json.dumps(mismatched_runsettings, indent=2)
         )
 
+    @pytest.mark.parametrize(
+        "test_case",
+        PARAMETERS_TO_TEST_WITH_OUTPUT,
+    )
+    @pytest.mark.disable_mock_code_hash
+    def test_register_output_for_anonymous_internal_component(
+        self,
+        client: MLClient,
+        test_case: tuple,
+    ):
+        (
+            yaml_path,
+            inputs,
+            runsettings_dict,
+            pipeline_runsettings_dict,
+            output_port_name,
+            output_name,
+            output_version,
+        ) = test_case
+        node_func: InternalComponent = load_component(yaml_path)
+
+        @pipeline()
+        def pipeline_func():
+            node = node_func(**inputs)
+            node.outputs[output_port_name].type = "uri_file"  # use this, in case that the type is path
+            node.outputs[output_port_name].name = output_name
+            node.outputs[output_port_name].version = output_version
+            set_run_settings(node, runsettings_dict)
+
+        dsl_pipeline: PipelineJob = pipeline_func()
+        set_run_settings(dsl_pipeline.settings, pipeline_runsettings_dict)
+        dsl_pipeline.settings.default_compute = "cpu-cluster"
+
+        result = dsl_pipeline._validate()
+        assert result._to_dict() == {"result": "Succeeded"}
+
+        pipeline_job = assert_job_cancel(dsl_pipeline, client)
+
+        assert pipeline_job.jobs["node"].outputs[output_port_name].name == output_name
+        assert pipeline_job.jobs["node"].outputs[output_port_name].version == output_version
+
     @pytest.mark.parametrize(
         "test_case_i,test_case_name",
         TEST_CASE_NAME_ENUMERATE,