fix: unrelated test suite pollution, and update spark base class

Author: Brock Wade

src/sagemaker/spark/processing.py

@@ -279,6 +279,12 @@ def run(
     def _extend_processing_args(self, inputs, outputs, **kwargs):
         """Extends processing job args such as inputs."""
 
+        # make a copy of user outputs
+        outputs = outputs or []
+        extended_outputs = []
+        for user_output in outputs:
+            extended_outputs.append(user_output)
+
         if kwargs.get("spark_event_logs_s3_uri"):
             spark_event_logs_s3_uri = kwargs.get("spark_event_logs_s3_uri")
             self._validate_s3_uri(spark_event_logs_s3_uri)
@@ -297,16 +303,20 @@ def _extend_processing_args(self, inputs, outputs, **kwargs):
                 s3_upload_mode="Continuous",
             )
 
-            outputs = outputs or []
-            outputs.append(output)
+            extended_outputs.append(output)
+
+        # make a copy of user inputs
+        inputs = inputs or []
+        extended_inputs = []
+        for user_input in inputs:
+            extended_inputs.append(user_input)
 
         if kwargs.get("configuration"):
             configuration = kwargs.get("configuration")
             self._validate_configuration(configuration)
-            inputs = inputs or []
-            inputs.append(self._stage_configuration(configuration))
+            extended_inputs.append(self._stage_configuration(configuration))
 
-        return inputs, outputs
+        return extended_inputs, extended_outputs
 
     def start_history_server(self, spark_event_logs_s3_uri=None):
         """Starts a Spark history server.

tests/unit/sagemaker/workflow/test_pipeline.py

@@ -17,7 +17,7 @@
 
 import pytest
 
-from mock import Mock
+from mock import Mock, patch
 
 from sagemaker import s3
 from sagemaker.workflow.condition_step import ConditionStep
@@ -78,6 +78,7 @@ def test_pipeline_create_with_parallelism_config(sagemaker_session_mock, role_ar
     )
 
 
+@patch("sagemaker.s3.S3Uploader.upload_string_as_file_body")
 def test_large_pipeline_create(sagemaker_session_mock, role_arn):
     parameter = ParameterString("MyStr")
     pipeline = Pipeline(
@@ -87,8 +88,6 @@ def test_large_pipeline_create(sagemaker_session_mock, role_arn):
         sagemaker_session=sagemaker_session_mock,
     )
 
-    s3.S3Uploader.upload_string_as_file_body = Mock()
-
     pipeline.create(role_arn=role_arn)
 
     assert s3.S3Uploader.upload_string_as_file_body.called_with(
@@ -151,6 +150,7 @@ def test_pipeline_update_with_parallelism_config(sagemaker_session_mock, role_ar
     )
 
 
+@patch("sagemaker.s3.S3Uploader.upload_string_as_file_body")
 def test_large_pipeline_update(sagemaker_session_mock, role_arn):
     parameter = ParameterString("MyStr")
     pipeline = Pipeline(
@@ -160,8 +160,6 @@ def test_large_pipeline_update(sagemaker_session_mock, role_arn):
         sagemaker_session=sagemaker_session_mock,
     )
 
-    s3.S3Uploader.upload_string_as_file_body = Mock()
-
     pipeline.create(role_arn=role_arn)
 
     assert s3.S3Uploader.upload_string_as_file_body.called_with(

tests/unit/sagemaker/workflow/test_processing_step.py

@@ -992,6 +992,10 @@ def test_spark_processor(spark_processor, processing_input, pipeline_session):
                     SPARK_SUBMIT_FILE2,
                 ],
                 "spark_event_logs_s3_uri": ParameterString("MySparkEventLogS3Uri"),
+                "configuration": {
+                    "Classification": "core-site",
+                    "Properties": {"hadoop.security.groups.cache.secs": "250"},
+                },
             },
         ),
         (
@@ -1016,6 +1020,10 @@ def test_spark_processor(spark_processor, processing_input, pipeline_session):
                 "submit_jars": [SPARK_DEP_JAR],
                 "submit_files": [SPARK_SUBMIT_FILE1, SPARK_SUBMIT_FILE2],
                 "spark_event_logs_s3_uri": ParameterString("MySparkEventLogS3Uri"),
+                "configuration": {
+                    "Classification": "core-site",
+                    "Properties": {"hadoop.security.groups.cache.secs": "250"},
+                },
             },
         ),
     ],