fix: address review comments (iteration #1)

Author: aviruthen

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

@@ -21,7 +21,7 @@
 normalization. For a convenient way to create ``ProcessingInput`` objects from
 local sources, use the :func:`processing_input_from_local` helper function.
 """
-from __future__ import absolute_import
+from __future__ import annotations
 
 import json
 import logging
@@ -93,11 +93,11 @@
 def processing_input_from_local(
     source: str,
     destination: str,
-    input_name: Optional[str] = None,
+    input_name: str | None = None,
     s3_data_type: str = "S3Prefix",
     s3_input_mode: str = "File",
-    s3_data_distribution_type: Optional[str] = None,
-    s3_compression_type: Optional[str] = None,
+    s3_data_distribution_type: str | None = None,
+    s3_compression_type: str | None = None,
 ) -> ProcessingInput:
     """Creates a ProcessingInput from a local file/directory path or S3 URI.
 
@@ -131,7 +131,7 @@ def processing_input_from_local(
 
     Raises:
         ValueError: If ``source`` is a local path that does not exist.
-        ValueError: If ``source`` is empty or None.
+        TypeError: If ``source`` is not a string.
 
     Examples:
         Create an input from a local directory::
@@ -152,12 +152,15 @@ def processing_input_from_local(
                 destination="/opt/ml/processing/input/data",
             )
     """
-    if not source:
+    if not isinstance(source, str) or not source:
         raise ValueError(
-            f"source must be a valid local path or S3 URI, got: {source!r}"
+            f"source must be a non-empty string containing a valid local path "
+            f"or S3 URI, got: {source!r}"
         )
 
-    # Check if source is a local path (not an S3 URI)
+    # Check if source is a local path (not a remote URI).
+    # Note: On Windows, absolute paths like C:\data\file.csv will have
+    # parse_result.scheme == 'c', which correctly falls into the local path branch.
     parse_result = urlparse(source)
     if parse_result.scheme not in ("s3", "http", "https"):
         # Treat as local path - validate existence
@@ -172,6 +175,8 @@ def processing_input_from_local(
 
     s3_input_kwargs = {
         "s3_uri": source,
+        # local_path in ProcessingS3Input maps to the container destination path
+        # where the input data will be made available inside the processing container.
         "local_path": destination,
         "s3_data_type": s3_data_type,
         "s3_input_mode": s3_input_mode,
@@ -534,9 +539,10 @@ def _normalize_inputs(self, inputs=None, kms_key=None):
                         local_source = url2pathname(parse_result.path)
                     if not os.path.exists(local_source):
                         raise ValueError(
-                            f"Input source path does not exist: {file_input.s3_input.s3_uri!r}. "
-                            f"Please provide a valid local path or S3 URI for input "
-                            f"'{file_input.input_name}'."
+                            f"Input source path does not exist: "
+                            f"{file_input.s3_input.s3_uri!r}. "
+                            f"Please provide a valid local path or S3 URI "
+                            f"for input '{file_input.input_name}'."
                         )
                     if _pipeline_config:
                         desired_s3_uri = s3.s3_path_join(
@@ -557,6 +563,7 @@ def _normalize_inputs(self, inputs=None, kms_key=None):
                             "input",
                             file_input.input_name,
                         )
+                    # Log after both pipeline/non-pipeline branches have set desired_s3_uri
                     logger.info(
                         "Uploading local input '%s' from %s to %s",
                         file_input.input_name,

sagemaker-core/tests/unit/test_processing.py

@@ -50,37 +50,33 @@ def mock_session():
 class TestProcessingInputFromLocal:
     """Tests for the processing_input_from_local() factory function."""
 
-    def test_processing_input_from_local_with_file_path_creates_valid_input(self):
+    def test_processing_input_from_local_with_file_path_creates_valid_input(self, tmp_path):
         """A local file path should produce a valid ProcessingInput."""
-        with tempfile.NamedTemporaryFile(mode="w", delete=False, suffix=".csv") as f:
-            f.write("col1,col2\n1,2\n")
-            temp_file = f.name
-
-        try:
-            result = processing_input_from_local(
-                source=temp_file,
-                destination="/opt/ml/processing/input/data",
-                input_name="my-data",
-            )
-            assert isinstance(result, ProcessingInput)
-            assert result.input_name == "my-data"
-            assert result.s3_input.s3_uri == temp_file
-            assert result.s3_input.local_path == "/opt/ml/processing/input/data"
-            assert result.s3_input.s3_data_type == "S3Prefix"
-            assert result.s3_input.s3_input_mode == "File"
-        finally:
-            os.unlink(temp_file)
+        temp_file = tmp_path / "data.csv"
+        temp_file.write_text("col1,col2\n1,2\n")
 
-    def test_processing_input_from_local_with_directory_path_creates_valid_input(self):
+        result = processing_input_from_local(
+            source=str(temp_file),
+            destination="/opt/ml/processing/input/data",
+            input_name="my-data",
+        )
+        assert isinstance(result, ProcessingInput)
+        assert result.input_name == "my-data"
+        assert result.s3_input.s3_uri == str(temp_file)
+        # local_path in ProcessingS3Input maps to the container destination path
+        assert result.s3_input.local_path == "/opt/ml/processing/input/data"
+        assert result.s3_input.s3_data_type == "S3Prefix"
+        assert result.s3_input.s3_input_mode == "File"
+
+    def test_processing_input_from_local_with_directory_path_creates_valid_input(self, tmp_path):
         """A local directory path should produce a valid ProcessingInput."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            result = processing_input_from_local(
-                source=tmpdir,
-                destination="/opt/ml/processing/input/data",
-                input_name="dir-data",
-            )
-            assert isinstance(result, ProcessingInput)
-            assert result.s3_input.s3_uri == tmpdir
+        result = processing_input_from_local(
+            source=str(tmp_path),
+            destination="/opt/ml/processing/input/data",
+            input_name="dir-data",
+        )
+        assert isinstance(result, ProcessingInput)
+        assert result.s3_input.s3_uri == str(tmp_path)
 
     def test_processing_input_from_local_with_s3_uri_passes_through(self):
         """An S3 URI should pass through without local path validation."""
@@ -100,56 +96,53 @@ def test_processing_input_from_local_with_nonexistent_path_raises_value_error(se
                 destination="/opt/ml/processing/input/data",
             )
 
-    def test_processing_input_from_local_with_custom_input_name(self):
+    def test_processing_input_from_local_with_custom_input_name(self, tmp_path):
         """Custom input_name should be set on the ProcessingInput."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            result = processing_input_from_local(
-                source=tmpdir,
-                destination="/opt/ml/processing/input/data",
-                input_name="custom-name",
-            )
-            assert result.input_name == "custom-name"
+        result = processing_input_from_local(
+            source=str(tmp_path),
+            destination="/opt/ml/processing/input/data",
+            input_name="custom-name",
+        )
+        assert result.input_name == "custom-name"
 
-    def test_processing_input_from_local_default_parameters(self):
+    def test_processing_input_from_local_default_parameters(self, tmp_path):
         """Default parameters should be applied correctly."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            result = processing_input_from_local(
-                source=tmpdir,
-                destination="/opt/ml/processing/input/data",
-            )
-            assert result.input_name is None
-            assert result.s3_input.s3_data_type == "S3Prefix"
-            assert result.s3_input.s3_input_mode == "File"
+        result = processing_input_from_local(
+            source=str(tmp_path),
+            destination="/opt/ml/processing/input/data",
+        )
+        assert result.input_name is None
+        assert result.s3_input.s3_data_type == "S3Prefix"
+        assert result.s3_input.s3_input_mode == "File"
 
     def test_processing_input_from_local_with_empty_source_raises_value_error(self):
         """Empty source should raise ValueError."""
-        with pytest.raises(ValueError, match="source must be a valid local path or S3 URI"):
+        with pytest.raises(ValueError, match="source must be a non-empty string"):
             processing_input_from_local(
                 source="",
                 destination="/opt/ml/processing/input/data",
             )
 
     def test_processing_input_from_local_with_none_source_raises_value_error(self):
         """None source should raise ValueError."""
-        with pytest.raises(ValueError, match="source must be a valid local path or S3 URI"):
+        with pytest.raises(ValueError, match="source must be a non-empty string"):
             processing_input_from_local(
                 source=None,
                 destination="/opt/ml/processing/input/data",
             )
 
-    def test_processing_input_from_local_with_optional_s3_params(self):
+    def test_processing_input_from_local_with_optional_s3_params(self, tmp_path):
         """Optional S3 parameters should be passed through."""
-        with tempfile.TemporaryDirectory() as tmpdir:
-            result = processing_input_from_local(
-                source=tmpdir,
-                destination="/opt/ml/processing/input/data",
-                s3_data_distribution_type="FullyReplicated",
-                s3_compression_type="Gzip",
-            )
-            assert result.s3_input.s3_data_distribution_type == "FullyReplicated"
-            assert result.s3_input.s3_compression_type == "Gzip"
+        result = processing_input_from_local(
+            source=str(tmp_path),
+            destination="/opt/ml/processing/input/data",
+            s3_data_distribution_type="FullyReplicated",
+            s3_compression_type="Gzip",
+        )
+        assert result.s3_input.s3_data_distribution_type == "FullyReplicated"
+        assert result.s3_input.s3_compression_type == "Gzip"
 
-    def test_processing_input_from_local_used_in_normalize_inputs(self, mock_session):
+    def test_processing_input_from_local_used_in_normalize_inputs(self, mock_session, tmp_path):
         """ProcessingInput from processing_input_from_local should work with _normalize_inputs."""
         processor = Processor(
             role="arn:aws:iam::123456789012:role/SageMakerRole",
@@ -160,18 +153,18 @@ def test_processing_input_from_local_used_in_normalize_inputs(self, mock_session
         )
         processor._current_job_name = "test-job"
 
-        with tempfile.TemporaryDirectory() as tmpdir:
-            inp = processing_input_from_local(
-                source=tmpdir,
-                destination="/opt/ml/processing/input/data",
-                input_name="local-data",
-            )
-            with patch(
-                "sagemaker.core.s3.S3Uploader.upload", return_value="s3://bucket/uploaded"
-            ):
-                result = processor._normalize_inputs([inp])
-                assert len(result) == 1
-                assert result[0].s3_input.s3_uri == "s3://bucket/uploaded"
+        inp = processing_input_from_local(
+            source=str(tmp_path),
+            destination="/opt/ml/processing/input/data",
+            input_name="local-data",
+        )
+        with patch(
+            "sagemaker.core.processing.s3.S3Uploader.upload",
+            return_value="s3://bucket/uploaded",
+        ):
+            result = processor._normalize_inputs([inp])
+            assert len(result) == 1
+            assert result[0].s3_input.s3_uri == "s3://bucket/uploaded"
 
 
 class TestNormalizeInputsLocalPathValidation:
@@ -199,7 +192,7 @@ def test_normalize_inputs_with_nonexistent_local_path_raises_value_error(self, m
         with pytest.raises(ValueError, match="Input source path does not exist"):
             processor._normalize_inputs(inputs)
 
-    def test_normalize_inputs_with_local_source_uploads_to_s3(self, mock_session):
+    def test_normalize_inputs_with_local_source_uploads_to_s3(self, mock_session, tmp_path):
         """A valid local path should be uploaded to S3."""
         processor = Processor(
             role="arn:aws:iam::123456789012:role/SageMakerRole",
@@ -210,25 +203,24 @@ def test_normalize_inputs_with_local_source_uploads_to_s3(self, mock_session):
         )
         processor._current_job_name = "test-job"
 
-        with tempfile.TemporaryDirectory() as tmpdir:
-            s3_input = ProcessingS3Input(
-                s3_uri=tmpdir,
-                local_path="/opt/ml/processing/input",
-                s3_data_type="S3Prefix",
-                s3_input_mode="File",
-            )
-            inputs = [ProcessingInput(input_name="local-input", s3_input=s3_input)]
+        s3_input = ProcessingS3Input(
+            s3_uri=str(tmp_path),
+            local_path="/opt/ml/processing/input",
+            s3_data_type="S3Prefix",
+            s3_input_mode="File",
+        )
+        inputs = [ProcessingInput(input_name="local-input", s3_input=s3_input)]
 
-            with patch(
-                "sagemaker.core.s3.S3Uploader.upload",
-                return_value="s3://test-bucket/sagemaker/test-job/input/local-input",
-            ) as mock_upload:
-                result = processor._normalize_inputs(inputs)
-                assert len(result) == 1
-                assert result[0].s3_input.s3_uri.startswith("s3://")
-                mock_upload.assert_called_once()
+        with patch(
+            "sagemaker.core.processing.s3.S3Uploader.upload",
+            return_value="s3://test-bucket/sagemaker/test-job/input/local-input",
+        ) as mock_upload:
+            result = processor._normalize_inputs(inputs)
+            assert len(result) == 1
+            assert result[0].s3_input.s3_uri.startswith("s3://")
+            mock_upload.assert_called_once()
 
-    def test_normalize_inputs_local_path_logs_upload_info(self, mock_session):
+    def test_normalize_inputs_local_path_logs_upload_info(self, mock_session, tmp_path):
         """Uploading a local path should log an info message."""
         processor = Processor(
             role="arn:aws:iam::123456789012:role/SageMakerRole",
@@ -239,27 +231,36 @@ def test_normalize_inputs_local_path_logs_upload_info(self, mock_session):
         )
         processor._current_job_name = "test-job"
 
-        with tempfile.TemporaryDirectory() as tmpdir:
-            s3_input = ProcessingS3Input(
-                s3_uri=tmpdir,
-                local_path="/opt/ml/processing/input",
-                s3_data_type="S3Prefix",
-                s3_input_mode="File",
-            )
-            inputs = [ProcessingInput(input_name="local-input", s3_input=s3_input)]
+        s3_input = ProcessingS3Input(
+            s3_uri=str(tmp_path),
+            local_path="/opt/ml/processing/input",
+            s3_data_type="S3Prefix",
+            s3_input_mode="File",
+        )
+        inputs = [ProcessingInput(input_name="local-input", s3_input=s3_input)]
 
-            with patch(
-                "sagemaker.core.s3.S3Uploader.upload",
-                return_value="s3://test-bucket/uploaded",
-            ):
-                with patch("sagemaker.core.processing.logger") as mock_logger:
-                    processor._normalize_inputs(inputs)
-                    mock_logger.info.assert_any_call(
-                        "Uploading local input '%s' from %s to %s",
-                        "local-input",
-                        tmpdir,
-                        f"s3://test-bucket/sagemaker/test-job/input/local-input",
-                    )
+        with patch(
+            "sagemaker.core.processing.s3.S3Uploader.upload",
+            return_value="s3://test-bucket/uploaded",
+        ):
+            with patch("sagemaker.core.processing.logger") as mock_logger:
+                processor._normalize_inputs(inputs)
+                # Verify the upload log message was emitted with the correct format.
+                # The exact S3 path depends on default_bucket/prefix/job_name/input_name.
+                mock_logger.info.assert_called()
+                log_calls = mock_logger.info.call_args_list
+                upload_log_found = any(
+                    len(call.args) >= 4
+                    and call.args[0] == "Uploading local input '%s' from %s to %s"
+                    and call.args[1] == "local-input"
+                    and call.args[2] == str(tmp_path)
+                    and call.args[3].startswith("s3://")
+                    for call in log_calls
+                )
+                assert upload_log_found, (
+                    f"Expected upload log message not found in logger.info calls: "
+                    f"{log_calls}"
+                )
 
 
 class TestProcessorNormalizeArgs: