fix: address review comments (iteration #1)

Author: aviruthen

sagemaker-core/src/sagemaker/core/processing.py

@@ -483,10 +483,16 @@ def _normalize_outputs(self, outputs=None):
                 # Generate a name for the ProcessingOutput if it doesn't have one.
                 if output.output_name is None:
                     output.output_name = "output-{}".format(count)
-                if output.s3_output and output.s3_output.s3_uri is not None and is_pipeline_variable(output.s3_output.s3_uri):
+                if (
+                    output.s3_output
+                    and output.s3_output.s3_uri is not None
+                    and is_pipeline_variable(output.s3_output.s3_uri)
+                ):
                     normalized_outputs.append(output)
                     continue
-                # If s3_output is None or s3_uri is None, auto-generate an S3 URI
+                # If s3_output is None or s3_uri is None, auto-generate
+                # an S3 URI (V2 parity: destination=None delegates to
+                # SageMaker).
                 if not output.s3_output or output.s3_output.s3_uri is None:
                     if _pipeline_config:
                         s3_uri = Join(
@@ -495,7 +501,6 @@ def _normalize_outputs(self, outputs=None):
                                 "s3:/",
                                 self.sagemaker_session.default_bucket(),
                                 *(
-                                    # don't include default_bucket_prefix if it is None or ""
                                     [self.sagemaker_session.default_bucket_prefix]
                                     if self.sagemaker_session.default_bucket_prefix
                                     else []
@@ -517,20 +522,32 @@ def _normalize_outputs(self, outputs=None):
                             output.output_name,
                         )
                     if output.s3_output:
+                        # s3_output exists but s3_uri is None
                         output.s3_output.s3_uri = s3_uri
                     else:
-                        from sagemaker.core.shapes import ProcessingS3Output as _ProcessingS3Output
+                        # s3_output is None — create a new one with
+                        # sensible defaults.
+                        # Import here to avoid circular import with
+                        # shapes module.
+                        from sagemaker.core.shapes import (
+                            ProcessingS3Output as _ProcessingS3Output,
+                        )
                         output.s3_output = _ProcessingS3Output(
                             s3_uri=s3_uri,
-                            local_path=output.s3_output.local_path if output.s3_output else "/opt/ml/processing/output",
+                            local_path="/opt/ml/processing/output",
                             s3_upload_mode="EndOfJob",
                         )
                     normalized_outputs.append(output)
                     continue
                 # If the output's s3_uri is not an s3_uri, create one.
                 parse_result = urlparse(output.s3_output.s3_uri)
                 if parse_result.scheme != "s3":
-                    if getattr(self.sagemaker_session, "local_mode", False) and parse_result.scheme == "file":
+                    if (
+                        getattr(
+                            self.sagemaker_session, "local_mode", False
+                        )
+                        and parse_result.scheme == "file"
+                    ):
                         normalized_outputs.append(output)
                         continue
                     if _pipeline_config:

sagemaker-core/tests/unit/test_processing.py

@@ -1436,6 +1436,8 @@ def test_normalize_outputs_with_none_s3_uri_generates_s3_path(self, mock_session
 
     def test_normalize_outputs_with_none_s3_uri_and_pipeline_config(self, mock_session):
         """When s3_uri is None and pipeline_config is set, use pipeline-based path."""
+        from sagemaker.core.workflow.functions import Join
+
         processor = Processor(
             role="arn:aws:iam::123456789012:role/SageMakerRole",
             image_uri="test-image:latest",
@@ -1459,7 +1461,11 @@ def test_normalize_outputs_with_none_s3_uri_and_pipeline_config(self, mock_sessi
 
         assert len(result) == 1
         # The result should be a Join object (pipeline variable) when pipeline_config is set
-        assert result[0].s3_output.s3_uri is not None
+        assert isinstance(result[0].s3_output.s3_uri, Join)
+        # Verify the Join contains expected pipeline-related values
+        join_obj = result[0].s3_output.s3_uri
+        assert join_obj.on == "/"
+        assert "test-pipeline" in join_obj.values
 
     def test_normalize_outputs_with_none_s3_uri_auto_generates_name(self, mock_session):
         """When output_name is None and s3_uri is None, both should be auto-generated."""
@@ -1488,6 +1494,29 @@ def test_normalize_outputs_with_none_s3_uri_auto_generates_name(self, mock_sessi
         assert generated_uri.startswith("s3://")
         assert "output-1" in generated_uri
 
+    def test_normalize_outputs_with_no_s3_output_at_all(self, mock_session):
+        """When s3_output is None entirely, a new ProcessingS3Output is created."""
+        processor = Processor(
+            role="arn:aws:iam::123456789012:role/SageMakerRole",
+            image_uri="test-image:latest",
+            instance_count=1,
+            instance_type="ml.m5.xlarge",
+            sagemaker_session=mock_session,
+        )
+        processor._current_job_name = "test-job"
+
+        outputs = [ProcessingOutput(output_name="my-output")]
+
+        with patch("sagemaker.core.workflow.utilities._pipeline_config", None):
+            result = processor._normalize_outputs(outputs)
+
+        assert len(result) == 1
+        assert result[0].s3_output is not None
+        assert result[0].s3_output.s3_uri.startswith("s3://")
+        assert result[0].s3_output.local_path == "/opt/ml/processing/output"
+        assert result[0].s3_output.s3_upload_mode == "EndOfJob"
+        assert "my-output" in result[0].s3_output.s3_uri
+
     def test_processing_output_to_request_dict_omits_s3_uri_when_none(self):
         """Verify _processing_output_to_request_dict omits S3Uri when s3_uri is None."""
         s3_output = ProcessingS3Output(