fix: Update localmode code to decode urllib response as UTF8 (aws#3284)

Author: navinsoni

src/sagemaker/local/entities.py

@@ -314,7 +314,7 @@ def start(self, input_data, output_data, transform_resources, **kwargs):
         endpoint_url = "http://%s:%d/execution-parameters" % (get_docker_host(), serving_port)
         response, code = _perform_request(endpoint_url)
         if code == 200:
-            execution_parameters = json.loads(response.read())
+            execution_parameters = json.loads(response.data.decode("utf-8"))
             # MaxConcurrentTransforms is ignored because we currently only support 1
             for setting in ("BatchStrategy", "MaxPayloadInMB"):
                 if setting not in kwargs and setting in execution_parameters:

src/sagemaker/workflow/conditions.py

@@ -20,13 +20,15 @@
 import abc
 
 from enum import Enum
-from typing import List, Union
+from typing import Dict, List, Union
 
 import attr
 
+from sagemaker.workflow import is_pipeline_variable
 from sagemaker.workflow.entities import (
     DefaultEnumMeta,
     Entity,
+    Expression,
     PrimitiveType,
     RequestType,
 )
@@ -289,3 +291,18 @@ def _referenced_steps(self) -> List[str]:
         for condition in self.conditions:
             steps.extend(condition._referenced_steps)
         return steps
+
+
+def primitive_or_expr(
+    value: Union[ExecutionVariable, Expression, PrimitiveType, Parameter, Properties]
+) -> Union[Dict[str, str], PrimitiveType]:
+    """Provide the expression of the value or return value if it is a primitive.
+
+    Args:
+        value (Union[ConditionValueType, PrimitiveType]): The value to evaluate.
+    Returns:
+        Either the expression of the value or the primitive value.
+    """
+    if is_pipeline_variable(value):
+        return value.expr
+    return value

tests/unit/test_local_entities.py

@@ -106,7 +106,7 @@ def test_start_local_transform_job(_perform_batch_inference, _perform_request, l
 
     response = Mock()
     _perform_request.return_value = (response, 200)
-    response.read.return_value = '{"BatchStrategy": "SingleRecord"}'
+    response.data = '{"BatchStrategy": "SingleRecord"}'.encode("UTF-8")
     local_transform_job.primary_container["ModelDataUrl"] = "file:///some/model"
     local_transform_job.start(input_data, output_data, transform_resources, Environment={})
 
@@ -176,9 +176,9 @@ def test_start_local_transform_job_from_remote_docker_host(
     output_data = {}
     transform_resources = {"InstanceType": "local"}
     m_get_docker_host.return_value = "some_host"
-    perform_request_mock = Mock()
-    m_perform_request.return_value = (perform_request_mock, 200)
-    perform_request_mock.read.return_value = '{"BatchStrategy": "SingleRecord"}'
+    response = Mock()
+    m_perform_request.return_value = (response, 200)
+    response.data = '{"BatchStrategy": "SingleRecord"}'.encode("UTF-8")
     local_transform_job.primary_container["ModelDataUrl"] = "file:///some/model"
     local_transform_job.start(input_data, output_data, transform_resources, Environment={})
     endpoints = [