[Fix] remove validation on number of tags and added integ tests

Author: jiapinw

src/sagemaker/feature_store/feature_processor/feature_scheduler.py

@@ -139,7 +139,8 @@ def to_pipeline(
     """
 
     _validate_input_for_to_pipeline_api(pipeline_name, step)
-    _validate_tags_for_to_pipeline_api(tags)
+    if tags:
+        _validate_tags_for_to_pipeline_api(tags)
 
     _sagemaker_session = sagemaker_session or Session()
 
@@ -529,13 +530,8 @@ def _validate_tags_for_to_pipeline_api(tags: List[Tuple[str, str]]) -> None:
         tags (List[Tuple[str, str]]): A list of tags attached to the pipeline.
 
     Raises (ValueError): raises ValueError when any of the following scenario happen:
-           1. more than 47 tags are provided to API.
-           2. reserved tag keys are provided to API.
+           1. reserved tag keys are provided to API.
     """
-    if len(tags) > 48:
-        raise ValueError(
-            "to_pipeline can only accept up to 47 tags. Please reduce the number of tags provided."
-        )
     provided_tag_keys = [tag_key_value_pair[0] for tag_key_value_pair in tags]
     for reserved_tag_key in TO_PIPELINE_RESERVED_TAG_KEYS:
         if reserved_tag_key in provided_tag_keys:

tests/integ/sagemaker/feature_store/feature_processor/test_feature_processor.py

@@ -550,11 +550,18 @@ def transform(raw_s3_data_as_df):
             step=transform,
             role=get_execution_role(sagemaker_session),
             max_retries=2,
+            tags=[("integ_test_tag_key_1", "integ_test_tag_key_2")],
             sagemaker_session=sagemaker_session,
         )
+        _sagemaker_client = get_sagemaker_client(sagemaker_session=sagemaker_session)
 
         assert pipeline_arn is not None
 
+        tags = _sagemaker_client.list_tags(ResourceArn=pipeline_arn)["Tags"]
+
+        tag_keys = [tag["Key"] for tag in tags]
+        assert "integ_test_tag_key_1" in tag_keys
+
         pipeline_description = Pipeline(name=pipeline_name).describe()
         assert pipeline_arn == pipeline_description["PipelineArn"]
         assert get_execution_role(sagemaker_session) == pipeline_description["RoleArn"]
@@ -570,7 +577,7 @@ def transform(raw_s3_data_as_df):
 
         status = _wait_for_pipeline_execution_to_reach_terminal_state(
             pipeline_execution_arn=pipeline_execution_arn,
-            sagemaker_client=get_sagemaker_client(sagemaker_session=sagemaker_session),
+            sagemaker_client=_sagemaker_client,
         )
         assert status == "Succeeded"
 

tests/unit/sagemaker/feature_store/feature_processor/test_feature_scheduler.py

@@ -532,68 +532,6 @@ def test_to_pipeline_pipeline_name_length_limit_exceeds(
         )
 
 
-@patch("sagemaker.remote_function.job.Session", return_value=mock_session())
-@patch(
-    "sagemaker.remote_function.job._JobSettings._get_default_spark_image",
-    return_value="some_image_uri",
-)
-@patch("sagemaker.remote_function.job.get_execution_role", return_value=EXECUTION_ROLE_ARN)
-def test_to_pipeline_too_many_tags(get_execution_role, mock_spark_image, session):
-    session.sagemaker_config = None
-    session.boto_region_name = TEST_REGION
-    session.expand_role.return_value = EXECUTION_ROLE_ARN
-    spark_config = SparkConfig(submit_files=["file_a", "file_b", "file_c"])
-    job_settings = _JobSettings(
-        spark_config=spark_config,
-        s3_root_uri=S3_URI,
-        role=EXECUTION_ROLE_ARN,
-        include_local_workdir=True,
-        instance_type="ml.m5.large",
-        encrypt_inter_container_traffic=True,
-        sagemaker_session=session,
-    )
-    jobs_container_entrypoint = [
-        "/bin/bash",
-        f"/opt/ml/input/data/{RUNTIME_SCRIPTS_CHANNEL_NAME}/{ENTRYPOINT_SCRIPT_NAME}",
-    ]
-    jobs_container_entrypoint.extend(["--jars", "path_a"])
-    jobs_container_entrypoint.extend(["--py-files", "path_b"])
-    jobs_container_entrypoint.extend(["--files", "path_c"])
-    jobs_container_entrypoint.extend([SPARK_APP_SCRIPT_PATH])
-    container_args = ["--s3_base_uri", f"{S3_URI}/pipeline_name"]
-    container_args.extend(["--region", session.boto_region_name])
-
-    mock_feature_processor_config = Mock(
-        mode=FeatureProcessorMode.PYSPARK, inputs=[tdh.FEATURE_PROCESSOR_INPUTS], output="some_fg"
-    )
-    mock_feature_processor_config.mode.return_value = FeatureProcessorMode.PYSPARK
-
-    wrapped_func = Mock(
-        Callable,
-        feature_processor_config=mock_feature_processor_config,
-        job_settings=job_settings,
-        wrapped_func=job_function,
-    )
-    wrapped_func.feature_processor_config.return_value = mock_feature_processor_config
-    wrapped_func.job_settings.return_value = job_settings
-    wrapped_func.wrapped_func.return_value = job_function
-
-    tags = [("key_" + str(i), "value_" + str(i)) for i in range(50)]
-
-    with pytest.raises(
-        ValueError,
-        match="to_pipeline can only accept up to 47 tags. Please reduce the number of tags provided.",
-    ):
-        to_pipeline(
-            pipeline_name="pipeline_name",
-            step=wrapped_func,
-            role=EXECUTION_ROLE_ARN,
-            max_retries=1,
-            tags=tags,
-            sagemaker_session=session,
-        )
-
-
 @patch("sagemaker.remote_function.job.Session", return_value=mock_session())
 @patch(
     "sagemaker.remote_function.job._JobSettings._get_default_spark_image",