fix: Prevent passing pipeline variables as code arg in ProcessingStep in compile time

Author: Dewen Qi

src/sagemaker/processing.py

@@ -34,6 +34,7 @@
 from sagemaker.local import LocalSession
 from sagemaker.utils import base_name_from_image, get_config_value, name_from_base
 from sagemaker.session import Session
+from sagemaker.workflow import is_pipeline_variable
 from sagemaker.workflow.properties import Properties
 from sagemaker.workflow.parameters import Parameter
 from sagemaker.workflow.entities import Expression
@@ -233,6 +234,12 @@ def _normalize_args(
             kms_key (str): The ARN of the KMS key that is used to encrypt the
                 user code file (default: None).
         """
+        if code and is_pipeline_variable(code):
+            raise ValueError(
+                f"code argument {code} has to be a valid S3 URI or local file path "
+                + "rather than a pipeline variable"
+            )
+
         self._current_job_name = self._generate_current_job_name(job_name=job_name)
 
         inputs_with_code = self._include_code_in_inputs(inputs, code, kms_key)

tests/unit/sagemaker/workflow/test_steps.py

@@ -666,6 +666,44 @@ def test_processing_step_normalizes_args_with_local_code(mock_normalize_args, sc
     )
 
 
+def test_processing_step_normalizes_args_with_param_str_local_code(
+    sagemaker_session, script_processor
+):
+    cache_config = CacheConfig(enable_caching=True, expire_after="PT1H")
+    code_param = ParameterString(name="Script", default_value="S3://my-bucket/file_name.py")
+    inputs = [
+        ProcessingInput(
+            source=f"s3://{BUCKET}/processing_manifest",
+            destination="processing_manifest",
+        )
+    ]
+    outputs = [
+        ProcessingOutput(
+            source=f"s3://{BUCKET}/processing_manifest",
+            destination="processing_manifest",
+        )
+    ]
+    step = ProcessingStep(
+        name="MyProcessingStep",
+        processor=script_processor,
+        code=code_param,
+        inputs=inputs,
+        outputs=outputs,
+        job_arguments=["arg1", "arg2"],
+        cache_config=cache_config,
+    )
+    pipeline = Pipeline(
+        name="MyPipeline",
+        parameters=[code_param],
+        steps=[step],
+        sagemaker_session=sagemaker_session,
+    )
+    with pytest.raises(ValueError) as error:
+        pipeline.definition()
+
+    assert "has to be a valid S3 URI or local file path" in str(error.value)
+
+
 @patch("sagemaker.processing.ScriptProcessor._normalize_args")
 def test_processing_step_normalizes_args_with_s3_code(mock_normalize_args, script_processor):
     cache_config = CacheConfig(enable_caching=True, expire_after="PT1H")