fix(invoke): Write in UTF-8 string instead of bytes (#5642)

* Revert "fix: Revert UTF-8 fixes #5485 and #5427 (#5512)"

This reverts commit 36f8bf9.

* Enforce utf8 on stdout/stderr/logfile

---------

Co-authored-by: Jacob Fuss <[email protected]>

samcli/commands/local/cli_common/invoke_context.py

@@ -7,7 +7,7 @@
 import os
 from enum import Enum
 from pathlib import Path
-from typing import IO, Any, Dict, List, Optional, Tuple, Type, cast
+from typing import Any, Dict, List, Optional, TextIO, Tuple, Type, cast
 
 from samcli.commands._utils.template import TemplateFailedParsingException, TemplateNotFoundException
 from samcli.commands.exceptions import ContainersInitializationException
@@ -196,7 +196,7 @@ def __init__(
         self._stacks: List[Stack] = None  # type: ignore
         self._env_vars_value: Optional[Dict] = None
         self._container_env_vars_value: Optional[Dict] = None
-        self._log_file_handle: Optional[IO] = None
+        self._log_file_handle: Optional[TextIO] = None
         self._debug_context: Optional[DebugContext] = None
         self._layers_downloader: Optional[LayerDownloader] = None
         self._container_manager: Optional[ContainerManager] = None
@@ -490,7 +490,7 @@ def _get_env_vars_value(filename: Optional[str]) -> Optional[Dict]:
             ) from ex
 
     @staticmethod
-    def _setup_log_file(log_file: Optional[str]) -> Optional[IO]:
+    def _setup_log_file(log_file: Optional[str]) -> Optional[TextIO]:
         """
         Open a log file if necessary and return the file handle. This will create a file if it does not exist
 
@@ -500,7 +500,7 @@ def _setup_log_file(log_file: Optional[str]) -> Optional[IO]:
         if not log_file:
             return None
 
-        return open(log_file, "wb")
+        return open(log_file, "w", encoding="utf8")
 
     @staticmethod
     def _get_debug_context(

samcli/commands/remote/remote_invoke_context.py

@@ -242,7 +242,7 @@ class DefaultRemoteInvokeResponseConsumer(RemoteInvokeConsumer[RemoteInvokeRespo
     _stream_writer: StreamWriter
 
     def consume(self, remote_invoke_response: RemoteInvokeResponse) -> None:
-        self._stream_writer.write(cast(str, remote_invoke_response.response).encode())
+        self._stream_writer.write_bytes(cast(str, remote_invoke_response.response).encode())
 
 
 @dataclass
@@ -254,4 +254,4 @@ class DefaultRemoteInvokeLogConsumer(RemoteInvokeConsumer[RemoteInvokeLogOutput]
     _stream_writer: StreamWriter
 
     def consume(self, remote_invoke_response: RemoteInvokeLogOutput) -> None:
-        self._stream_writer.write(remote_invoke_response.log_output.encode())
+        self._stream_writer.write_bytes(remote_invoke_response.log_output.encode())

samcli/lib/docker/log_streamer.py

@@ -47,23 +47,21 @@ def stream_progress(self, logs: docker.APIClient.logs):
                 else:
                     curr_log_line_id = ids[_id]
                     change_cursor_count = len(ids) - curr_log_line_id
-                    self._stream.write(
+                    self._stream.write_str(
                         self._cursor_up_formatter.cursor_format(change_cursor_count)
-                        + self._cursor_left_formatter.cursor_format(),
-                        encode=True,
+                        + self._cursor_left_formatter.cursor_format()
                     )
 
             self._stream_write(_id, status, stream, progress, error)
 
             if _id:
-                self._stream.write(
+                self._stream.write_str(
                     self._cursor_down_formatter.cursor_format(change_cursor_count)
-                    + self._cursor_left_formatter.cursor_format(),
-                    encode=True,
+                    + self._cursor_left_formatter.cursor_format()
                 )
-        self._stream.write(os.linesep, encode=True)
+        self._stream.write_str(os.linesep)
 
-    def _stream_write(self, _id: str, status: str, stream: bytes, progress: str, error: str):
+    def _stream_write(self, _id: str, status: str, stream: str, progress: str, error: str):
         """
         Write stream information to stderr, if the stream information contains a log id,
         use the carriage return character to rewrite that particular line.
@@ -80,14 +78,14 @@ def _stream_write(self, _id: str, status: str, stream: bytes, progress: str, err
 
         # NOTE(sriram-mv): Required for the purposes of when the cursor overflows existing terminal buffer.
         if not stream:
-            self._stream.write(os.linesep, encode=True)
-            self._stream.write(
-                self._cursor_up_formatter.cursor_format() + self._cursor_left_formatter.cursor_format(), encode=True
+            self._stream.write_str(os.linesep)
+            self._stream.write_str(
+                self._cursor_up_formatter.cursor_format() + self._cursor_left_formatter.cursor_format()
             )
-            self._stream.write(self._cursor_clear_formatter.cursor_format(), encode=True)
+            self._stream.write_str(self._cursor_clear_formatter.cursor_format())
 
         if not _id:
-            self._stream.write(stream, encode=True)
-            self._stream.write(status, encode=True)
+            self._stream.write_str(stream)
+            self._stream.write_str(status)
         else:
-            self._stream.write(f"\r{_id}: {status} {progress}", encode=True)
+            self._stream.write_str(f"\r{_id}: {status} {progress}")

samcli/lib/package/ecr_uploader.py

@@ -2,8 +2,8 @@
 Client for uploading packaged artifacts to ecr
 """
 import base64
-import io
 import logging
+from io import StringIO
 from typing import Dict
 
 import botocore
@@ -94,7 +94,7 @@ def upload(self, image, resource_name):
             else:
                 # we need to wait till the image got pushed to ecr, without this workaround sam sync for template
                 # contains image always fail, because the provided ecr uri is not exist.
-                _log_streamer = LogStreamer(stream=StreamWriter(stream=io.BytesIO(), auto_flush=True))
+                _log_streamer = LogStreamer(stream=StreamWriter(stream=StringIO(), auto_flush=True))
                 _log_streamer.stream_progress(push_logs)
 
         except (BuildError, APIError, LogStreamError) as ex:

samcli/lib/package/s3_uploader.py

@@ -265,4 +265,4 @@ def on_progress(self, bytes_transferred, **kwargs):
             )
             sys.stderr.flush()
             if int(percentage) == 100:  # noqa: PLR2004
-                sys.stderr.write("\n")
+                sys.stderr.write(os.linesep)

samcli/lib/utils/osutils.py

@@ -1,6 +1,7 @@
 """
 Common OS utilities
 """
+import io
 import logging
 import os
 import shutil
@@ -78,7 +79,7 @@ def rmtree_if_exists(path: Union[str, Path]):
         shutil.rmtree(path_obj)
 
 
-def stdout():
+def stdout() -> io.TextIOWrapper:
     """
     Returns the stdout as a byte stream in a Py2/PY3 compatible manner
 
@@ -87,10 +88,15 @@ def stdout():
     io.BytesIO
         Byte stream of Stdout
     """
-    return sys.stdout.buffer
+    # ensure stdout is utf8
+    sys.stdout.reconfigure(encoding="utf-8")  # type:ignore[attr-defined]
 
+    # Note(jfuss): sys.stdout is a type typing.TextIO but are initialized to
+    # io.TextIOWrapper. To make mypy and typing play well, tell mypy to ignore.
+    return sys.stdout  # type:ignore[return-value]
 
-def stderr():
+
+def stderr() -> io.TextIOWrapper:
     """
     Returns the stderr as a byte stream in a Py2/PY3 compatible manner
 
@@ -99,7 +105,12 @@ def stderr():
     io.BytesIO
         Byte stream of stderr
     """
-    return sys.stderr.buffer
+    # ensure stderr is utf8
+    sys.stderr.reconfigure(encoding="utf-8")  # type:ignore[attr-defined]
+
+    # Note(jfuss): sys.stderr is a type typing.TextIO but are initialized to
+    # io.TextIOWrapper. To make mypy and typing play well, tell mypy to ignore.
+    return sys.stderr  # type:ignore[return-value]
 
 
 def remove(path):

samcli/lib/utils/stream_writer.py

@@ -1,10 +1,11 @@
 """
 This class acts like a wrapper around output streams to provide any flexibility with output we need
 """
+from typing import TextIO, Union
 
 
 class StreamWriter:
-    def __init__(self, stream, auto_flush=False):
+    def __init__(self, stream: TextIO, auto_flush: bool = False):
         """
         Instatiates new StreamWriter to the specified stream
 
@@ -19,19 +20,33 @@ def __init__(self, stream, auto_flush=False):
         self._auto_flush = auto_flush
 
     @property
-    def stream(self):
+    def stream(self) -> TextIO:
         return self._stream
 
-    def write(self, output, encode=False):
+    def write_bytes(self, output: Union[bytes, bytearray]):
         """
         Writes specified text to the underlying stream
 
         Parameters
         ----------
         output bytes-like object
-            Bytes to write
+            Bytes to write into buffer
         """
-        self._stream.write(output.encode() if encode else output)
+        self._stream.buffer.write(output)
+
+        if self._auto_flush:
+            self._stream.flush()
+
+    def write_str(self, output: str):
+        """
+        Writes specified text to the underlying stream
+
+        Parameters
+        ----------
+        output string object
+            String to write
+        """
+        self._stream.write(output)
 
         if self._auto_flush:
             self._stream.flush()

samcli/lib/utils/subprocess_utils.py

@@ -34,7 +34,7 @@ def default_loading_pattern(stream_writer: Optional[StreamWriter] = None, loadin
         How frequently to generate the pattern
     """
     stream_writer = stream_writer or StreamWriter(sys.stderr)
-    stream_writer.write(".")
+    stream_writer.write_str(".")
     stream_writer.flush()
     sleep(loading_pattern_rate)
 
@@ -96,7 +96,7 @@ def _print_loading_pattern():
                 return_code = process.wait()
                 keep_printing = False
 
-                stream_writer.write(os.linesep)
+                stream_writer.write_str(os.linesep)
                 stream_writer.flush()
                 process_stderr = _check_and_convert_stream_to_string(process.stderr)
 

samcli/local/apigw/local_apigw_service.py

@@ -4,7 +4,7 @@
 import json
 import logging
 from datetime import datetime
-from io import BytesIO
+from io import StringIO
 from time import time
 from typing import Any, Dict, List, Optional
 
@@ -605,7 +605,7 @@ def _invoke_lambda_function(self, lambda_function_name: str, event: dict) -> str
         str
             A string containing the output from the Lambda function
         """
-        with BytesIO() as stdout:
+        with StringIO() as stdout:
             event_str = json.dumps(event, sort_keys=True)
             stdout_writer = StreamWriter(stdout, auto_flush=True)
 

samcli/local/docker/container.py

@@ -1,6 +1,8 @@
 """
 Representation of a generic Docker container
 """
+import io
+import json
 import logging
 import os
 import pathlib
@@ -9,14 +11,15 @@
 import tempfile
 import threading
 import time
-from typing import Optional
+from typing import Iterator, Optional, Tuple, Union
 
 import docker
 import requests
 from docker.errors import NotFound as DockerNetworkNotFound
 
 from samcli.lib.constants import DOCKER_MIN_API_VERSION
 from samcli.lib.utils.retry import retry
+from samcli.lib.utils.stream_writer import StreamWriter
 from samcli.lib.utils.tar import extract_tarfile
 from samcli.local.docker.effective_user import ROOT_USER_ID, EffectiveUser
 from samcli.local.docker.exceptions import ContainerNotStartableException, PortAlreadyInUse
@@ -318,7 +321,7 @@ def start(self, input_data=None):
             raise ex
 
     @retry(exc=requests.exceptions.RequestException, exc_raise=ContainerResponseException)
-    def wait_for_http_response(self, name, event, stdout):
+    def wait_for_http_response(self, name, event, stdout) -> str:
         # TODO(sriram-mv): `aws-lambda-rie` is in a mode where the function_name is always "function"
         # NOTE(sriram-mv): There is a connection timeout set on the http call to `aws-lambda-rie`, however there is not
         # a read time out for the response received from the server.
@@ -328,7 +331,7 @@ def wait_for_http_response(self, name, event, stdout):
             data=event.encode("utf-8"),
             timeout=(self.RAPID_CONNECTION_TIMEOUT, None),
         )
-        stdout.write(resp.content)
+        return json.dumps(json.loads(resp.content), ensure_ascii=False)
 
     def wait_for_result(self, full_path, event, stdout, stderr, start_timer=None):
         # NOTE(sriram-mv): Let logging happen in its own thread, so that a http request can be sent.
@@ -348,11 +351,21 @@ def wait_for_result(self, full_path, event, stdout, stderr, start_timer=None):
         # start the timer for function timeout right before executing the function, as waiting for the socket
         # can take some time
         timer = start_timer() if start_timer else None
-        self.wait_for_http_response(full_path, event, stdout)
+        response = self.wait_for_http_response(full_path, event, stdout)
         if timer:
             timer.cancel()
 
-    def wait_for_logs(self, stdout=None, stderr=None):
+        # NOTE(jfuss): Adding a sleep after we get a response from the contianer but before we
+        # we write the response to ensure the last thing written to stdout is the container response
+        time.sleep(1)
+        stdout.write_str(response)
+        stdout.flush()
+
+    def wait_for_logs(
+        self,
+        stdout: Optional[Union[StreamWriter, io.BytesIO, io.TextIOWrapper]] = None,
+        stderr: Optional[Union[StreamWriter, io.BytesIO, io.TextIOWrapper]] = None,
+    ):
         # Return instantly if we don't have to fetch any logs
         if not stdout and not stderr:
             return
@@ -364,7 +377,6 @@ def wait_for_logs(self, stdout=None, stderr=None):
 
         # Fetch both stdout and stderr streams from Docker as a single iterator.
         logs_itr = real_container.attach(stream=True, logs=True, demux=True)
-
         self._write_container_output(logs_itr, stdout=stdout, stderr=stderr)
 
     def _wait_for_socket_connection(self) -> None:
@@ -415,7 +427,11 @@ def copy(self, from_container_path, to_host_path) -> None:
             extract_tarfile(file_obj=fp, unpack_dir=to_host_path)
 
     @staticmethod
-    def _write_container_output(output_itr, stdout=None, stderr=None):
+    def _write_container_output(
+        output_itr: Iterator[Tuple[bytes, bytes]],
+        stdout: Optional[Union[StreamWriter, io.BytesIO, io.TextIOWrapper]] = None,
+        stderr: Optional[Union[StreamWriter, io.BytesIO, io.TextIOWrapper]] = None,
+    ):
         """
         Based on the data returned from the Container output, via the iterator, write it to the appropriate streams
 
@@ -434,13 +450,26 @@ def _write_container_output(output_itr, stdout=None, stderr=None):
             # Iterator returns a tuple of (stdout, stderr)
             for stdout_data, stderr_data in output_itr:
                 if stdout_data and stdout:
-                    stdout.write(stdout_data)
+                    Container._handle_data_writing(stdout, stdout_data)
 
                 if stderr_data and stderr:
-                    stderr.write(stderr_data)
+                    Container._handle_data_writing(stderr, stderr_data)
+
         except Exception as ex:
             LOG.debug("Failed to get the logs from the container", exc_info=ex)
 
+    @staticmethod
+    def _handle_data_writing(output_stream: Union[StreamWriter, io.BytesIO, io.TextIOWrapper], output_data: bytes):
+        if isinstance(output_stream, StreamWriter):
+            output_stream.write_bytes(output_data)
+            output_stream.flush()
+
+        if isinstance(output_stream, io.BytesIO):
+            output_stream.write(output_data)
+
+        if isinstance(output_stream, io.TextIOWrapper):
+            output_stream.buffer.write(output_data)
+
     @property
     def network_id(self):
         """

samcli/local/docker/lambda_image.py

@@ -3,6 +3,7 @@
 """
 import hashlib
 import logging
+import os
 import platform
 import re
 import sys
@@ -227,7 +228,7 @@ def build(self, runtime, packagetype, image, layers, architecture, stream=None,
             or not runtime
         ):
             stream_writer = stream or StreamWriter(sys.stderr)
-            stream_writer.write("Building image...")
+            stream_writer.write_str("Building image...")
             stream_writer.flush()
             self._build_image(
                 image if image else base_image, rapid_image, downloaded_layers, architecture, stream=stream_writer
@@ -338,15 +339,15 @@ def set_item_permission(tar_info):
                         platform=get_docker_platform(architecture),
                     )
                     for log in resp_stream:
-                        stream_writer.write(".")
+                        stream_writer.write_str(".")
                         stream_writer.flush()
                         if "error" in log:
-                            stream_writer.write("\n")
+                            stream_writer.write_str(os.linesep)
                             LOG.exception("Failed to build Docker Image")
                             raise ImageBuildException("Error building docker image: {}".format(log["error"]))
-                    stream_writer.write("\n")
+                    stream_writer.write_str(os.linesep)
                 except (docker.errors.BuildError, docker.errors.APIError) as ex:
-                    stream_writer.write("\n")
+                    stream_writer.write_str(os.linesep)
                     LOG.exception("Failed to build Docker Image")
                     raise ImageBuildException("Building Image failed.") from ex
         finally:

samcli/local/docker/manager.py

@@ -168,16 +168,16 @@ def pull_image(self, image_name, tag=None, stream=None):
                 raise DockerImagePullFailedException(str(ex)) from ex
 
             # io streams, especially StringIO, work only with unicode strings
-            stream_writer.write("\nFetching {}:{} Docker container image...".format(image_name, tag))
+            stream_writer.write_str("\nFetching {}:{} Docker container image...".format(image_name, tag))
 
             # Each line contains information on progress of the pull. Each line is a JSON string
             for _ in result_itr:
                 # For every line, print a dot to show progress
-                stream_writer.write(".")
+                stream_writer.write_str(".")
                 stream_writer.flush()
 
             # We are done. Go to the next line
-            stream_writer.write("\n")
+            stream_writer.write_str("\n")
 
     def has_image(self, image_name):
         """

samcli/local/lambda_service/local_lambda_invoke_service.py

@@ -162,7 +162,7 @@ def _invoke_request_handler(self, function_name):
 
         request_data = request_data.decode("utf-8")
 
-        stdout_stream = io.BytesIO()
+        stdout_stream = io.StringIO()
         stdout_stream_writer = StreamWriter(stdout_stream, auto_flush=True)
 
         try:

samcli/local/services/base_local_service.py

@@ -82,7 +82,7 @@ def service_response(body, headers, status_code):
 
 class LambdaOutputParser:
     @staticmethod
-    def get_lambda_output(stdout_stream: io.BytesIO) -> Tuple[str, bool]:
+    def get_lambda_output(stdout_stream: io.StringIO) -> Tuple[str, bool]:
         """
         This method will extract read the given stream and return the response from Lambda function separated out
         from any log statements it might have outputted. Logs end up in the stdout stream if the Lambda function
@@ -100,7 +100,7 @@ def get_lambda_output(stdout_stream: io.BytesIO) -> Tuple[str, bool]:
         bool
             If the response is an error/exception from the container
         """
-        lambda_response = stdout_stream.getvalue().decode("utf-8")
+        lambda_response = stdout_stream.getvalue()
 
         # When the Lambda Function returns an Error/Exception, the output is added to the stdout of the container. From
         # our perspective, the container returned some value, which is not always true. Since the output is the only

tests/integration/local/invoke/test_integrations_cli.py

@@ -291,6 +291,27 @@ def test_invoke_returns_expected_result_when_no_event_given(self):
         self.assertEqual(process.returncode, 0)
         self.assertEqual("{}", process_stdout.decode("utf-8"))
 
+    @pytest.mark.flaky(reruns=3)
+    def test_invoke_returns_utf8(self):
+        command_list = InvokeIntegBase.get_command_list(
+            "EchoEventFunction", template_path=self.template_path, event_path=self.event_utf8_path
+        )
+
+        process = Popen(command_list, stdout=PIPE)
+        try:
+            stdout, _ = process.communicate(timeout=TIMEOUT)
+        except TimeoutExpired:
+            process.kill()
+            raise
+
+        process_stdout = stdout.strip()
+
+        with open(self.event_utf8_path) as f:
+            expected_output = json.dumps(json.load(f), ensure_ascii=False)
+
+        self.assertEqual(process.returncode, 0)
+        self.assertEqual(expected_output, process_stdout.decode("utf-8"))
+
     @pytest.mark.flaky(reruns=3)
     def test_invoke_with_env_using_parameters(self):
         command_list = InvokeIntegBase.get_command_list(

tests/unit/commands/local/cli_common/test_invoke_context.py

@@ -1106,7 +1106,7 @@ def test_must_open_file_for_writing(self):
         with patch("samcli.commands.local.cli_common.invoke_context.open", m):
             InvokeContext._setup_log_file(filename)
 
-        m.assert_called_with(filename, "wb")
+        m.assert_called_with(filename, "w", encoding="utf8")
 
 
 class TestInvokeContext_get_debug_context(TestCase):

tests/unit/lib/utils/test_osutils.py

@@ -34,29 +34,21 @@ def test_raises_on_cleanup_failure(self, rmdir_mock):
     @patch("os.rmdir")
     def test_handles_ignore_error_case(self, rmdir_mock):
         rmdir_mock.side_effect = OSError("fail")
-        dir_name = None
         with osutils.mkdir_temp(ignore_errors=True) as tempdir:
-            dir_name = tempdir
             self.assertTrue(os.path.exists(tempdir))
 
 
 class Test_stderr(TestCase):
     def test_must_return_sys_stderr(self):
         expected_stderr = sys.stderr
 
-        if sys.version_info.major > 2:
-            expected_stderr = sys.stderr.buffer
-
         self.assertEqual(expected_stderr, osutils.stderr())
 
 
 class Test_stdout(TestCase):
     def test_must_return_sys_stdout(self):
         expected_stdout = sys.stdout
 
-        if sys.version_info.major > 2:
-            expected_stdout = sys.stdout.buffer
-
         self.assertEqual(expected_stdout, osutils.stdout())
 
 

tests/unit/lib/utils/test_stream_writer.py

@@ -1,6 +1,7 @@
 """
 Tests for StreamWriter
 """
+import io
 
 from unittest import TestCase
 
@@ -11,13 +12,13 @@
 
 class TestStreamWriter(TestCase):
     def test_must_write_to_stream(self):
-        buffer = "something"
+        buffer = b"something"
         stream_mock = Mock()
 
         writer = StreamWriter(stream_mock)
-        writer.write(buffer)
+        writer.write_bytes(buffer)
 
-        stream_mock.write.assert_called_once_with(buffer)
+        stream_mock.buffer.write.assert_called_once_with(buffer)
 
     def test_must_flush_underlying_stream(self):
         stream_mock = Mock()
@@ -31,7 +32,7 @@ def test_auto_flush_must_be_off_by_default(self):
         stream_mock = Mock()
 
         writer = StreamWriter(stream_mock)
-        writer.write("something")
+        writer.write_str("something")
 
         stream_mock.flush.assert_not_called()
 
@@ -46,6 +47,6 @@ def test_when_auto_flush_on_flush_after_each_write(self):
         writer = StreamWriter(stream_mock, True)
 
         for line in lines:
-            writer.write(line)
+            writer.write_str(line)
             flush_mock.assert_called_once_with()
             flush_mock.reset_mock()

tests/unit/lib/utils/test_subprocess_utils.py

@@ -11,6 +11,7 @@
 from parameterized import parameterized
 from unittest.mock import patch, Mock, call, ANY
 
+from samcli.lib.utils.stream_writer import StreamWriter
 from samcli.lib.utils.subprocess_utils import (
     default_loading_pattern,
     invoke_subprocess_with_loading_pattern,
@@ -64,7 +65,7 @@ def test_loader_stream_uses_passed_in_stdout(
     @patch("samcli.lib.utils.subprocess_utils.Popen")
     def test_loader_raises_exception_non_zero_exit_code(self, patched_Popen):
         standard_error = "an error has occurred"
-        mock_stream_writer = Mock()
+        mock_stream_writer = Mock(spec=StreamWriter)
         mock_process = Mock()
         mock_process.returncode = 1
         mock_process.stdout = None
@@ -74,7 +75,7 @@ def test_loader_raises_exception_non_zero_exit_code(self, patched_Popen):
         with self.assertRaises(LoadingPatternError) as ex:
             invoke_subprocess_with_loading_pattern({"args": ["ls"]}, mock_pattern, mock_stream_writer)
         self.assertIn(standard_error, ex.exception.message)
-        mock_stream_writer.write.assert_called_once_with(os.linesep)
+        mock_stream_writer.write_str.assert_called_once_with(os.linesep)
         mock_stream_writer.flush.assert_called_once_with()
 
     @patch("samcli.lib.utils.subprocess_utils.Popen")
@@ -95,19 +96,19 @@ def test_loader_raises_exception_bad_process(self, patched_Popen):
 
     @patch("samcli.lib.utils.subprocess_utils.StreamWriter")
     def test_default_pattern_default_stream_writer(self, patched_stream_writer):
-        stream_writer_mock = Mock()
+        stream_writer_mock = Mock(spec=StreamWriter)
         patched_stream_writer.return_value = stream_writer_mock
         default_loading_pattern(loading_pattern_rate=0.01)
         patched_stream_writer.assert_called_once_with(sys.stderr)
-        stream_writer_mock.write.assert_called_once_with(".")
+        stream_writer_mock.write_str.assert_called_once_with(".")
         stream_writer_mock.flush.assert_called_once_with()
 
     @patch("samcli.lib.utils.subprocess_utils.StreamWriter")
     def test_default_pattern(self, patched_stream_writer):
-        stream_writer_mock = Mock()
+        stream_writer_mock = Mock(spec=StreamWriter)
         default_loading_pattern(stream_writer_mock, 0.01)
         patched_stream_writer.assert_not_called()
-        stream_writer_mock.write.assert_called_once_with(".")
+        stream_writer_mock.write_str.assert_called_once_with(".")
         stream_writer_mock.flush.assert_called_once_with()
 
     @parameterized.expand([("hello".encode("utf-8"), "hello"), ("hello", "hello")])

tests/unit/local/docker/test_container.py

@@ -9,6 +9,7 @@
 from requests import RequestException
 
 from samcli.lib.utils.packagetype import IMAGE
+from samcli.lib.utils.stream_writer import StreamWriter
 from samcli.local.docker.container import (
     Container,
     ContainerResponseException,
@@ -721,17 +722,17 @@ def test_wait_for_result_waits_for_socket_before_post_request(self, patched_time
         self.assertEqual(mock_requests.post.call_count, 0)
 
     def test_write_container_output_successful(self):
-        stdout_mock = Mock()
-        stderr_mock = Mock()
+        stdout_mock = Mock(spec=StreamWriter)
+        stderr_mock = Mock(spec=StreamWriter)
 
         def _output_iterator():
-            yield "Hello", None
-            yield None, "World"
+            yield b"Hello", None
+            yield None, b"World"
             raise ValueError("The pipe has been ended.")
 
         Container._write_container_output(_output_iterator(), stdout_mock, stderr_mock)
-        stdout_mock.assert_has_calls([call.write("Hello")])
-        stderr_mock.assert_has_calls([call.write("World")])
+        stdout_mock.assert_has_calls([call.write_bytes(b"Hello")])
+        stderr_mock.assert_has_calls([call.write_bytes(b"World")])
 
 
 class TestContainer_wait_for_logs(TestCase):
@@ -785,33 +786,33 @@ class TestContainer_write_container_output(TestCase):
     def setUp(self):
         self.output_itr = [(b"stdout1", None), (None, b"stderr1"), (b"stdout2", b"stderr2"), (None, None)]
 
-        self.stdout_mock = Mock()
-        self.stderr_mock = Mock()
+        self.stdout_mock = Mock(spec=StreamWriter)
+        self.stderr_mock = Mock(spec=StreamWriter)
 
     def test_must_write_stdout_and_stderr_data(self):
         # All the invalid frames must be ignored
 
         Container._write_container_output(self.output_itr, stdout=self.stdout_mock, stderr=self.stderr_mock)
 
-        self.stdout_mock.write.assert_has_calls([call(b"stdout1"), call(b"stdout2")])
+        self.stdout_mock.write_bytes.assert_has_calls([call(b"stdout1"), call(b"stdout2")])
 
-        self.stderr_mock.write.assert_has_calls([call(b"stderr1"), call(b"stderr2")])
+        self.stderr_mock.write_bytes.assert_has_calls([call(b"stderr1"), call(b"stderr2")])
 
     def test_must_write_only_stderr(self):
         # All the invalid frames must be ignored
 
         Container._write_container_output(self.output_itr, stdout=None, stderr=self.stderr_mock)
 
-        self.stdout_mock.write.assert_not_called()
+        self.stdout_mock.write_bytes.assert_not_called()
 
-        self.stderr_mock.write.assert_has_calls([call(b"stderr1"), call(b"stderr2")])
+        self.stderr_mock.write_bytes.assert_has_calls([call(b"stderr1"), call(b"stderr2")])
 
     def test_must_write_only_stdout(self):
         Container._write_container_output(self.output_itr, stdout=self.stdout_mock, stderr=None)
 
-        self.stdout_mock.write.assert_has_calls([call(b"stdout1"), call(b"stdout2")])
+        self.stdout_mock.write_bytes.assert_has_calls([call(b"stdout1"), call(b"stdout2")])
 
-        self.stderr_mock.write.assert_not_called()  # stderr must never be called
+        self.stderr_mock.write_bytes.assert_not_called()  # stderr must never be called
 
 
 class TestContainer_wait_for_socket_connection(TestCase):

tests/unit/local/docker/test_lambda_image.py

@@ -1,4 +1,3 @@
-import io
 import tempfile
 
 from unittest import TestCase
@@ -271,7 +270,7 @@ def test_force_building_image_that_doesnt_already_exists(
         docker_client_mock.images.get.side_effect = ImageNotFound("image not found")
         docker_client_mock.images.list.return_value = []
 
-        stream = io.StringIO()
+        stream = Mock()
 
         lambda_image = LambdaImage(layer_downloader_mock, False, True, docker_client=docker_client_mock)
         actual_image_id = lambda_image.build(
@@ -311,7 +310,7 @@ def test_force_building_image_on_daemon_404(
         docker_client_mock.images.get.side_effect = NotFound("image not found")
         docker_client_mock.images.list.return_value = []
 
-        stream = io.StringIO()
+        stream = Mock()
 
         lambda_image = LambdaImage(layer_downloader_mock, False, True, docker_client=docker_client_mock)
         actual_image_id = lambda_image.build(
@@ -351,7 +350,7 @@ def test_docker_distribution_api_error_on_daemon_api_error(
         docker_client_mock.images.get.side_effect = APIError("error from docker daemon")
         docker_client_mock.images.list.return_value = []
 
-        stream = io.StringIO()
+        stream = Mock()
 
         lambda_image = LambdaImage(layer_downloader_mock, False, True, docker_client=docker_client_mock)
         with self.assertRaises(DockerDistributionAPIError):
@@ -377,7 +376,7 @@ def test_not_force_building_image_that_doesnt_already_exists(
         docker_client_mock.images.get.side_effect = ImageNotFound("image not found")
         docker_client_mock.images.list.return_value = []
 
-        stream = io.StringIO()
+        stream = Mock()
 
         lambda_image = LambdaImage(layer_downloader_mock, False, False, docker_client=docker_client_mock)
         actual_image_id = lambda_image.build(

tests/unit/local/docker/test_manager.py

@@ -1,8 +1,6 @@
 """
 Tests container manager
 """
-
-import io
 import importlib
 from unittest import TestCase
 from unittest.mock import Mock, patch, MagicMock, ANY, call
@@ -218,17 +216,29 @@ def setUp(self):
         self.manager = ContainerManager(docker_client=self.mock_docker_client)
 
     def test_must_pull_and_print_progress_dots(self):
-        stream = io.StringIO()
+        stream = Mock()
         pull_result = [1, 2, 3, 4, 5, 6, 7, 8, 9, 0]
         self.mock_docker_client.api.pull.return_value = pull_result
-        expected_stream_output = "\nFetching {}:latest Docker container image...{}\n".format(
-            self.image_name, "." * len(pull_result)  # Progress bar will print one dot per response from pull API
-        )
+        expected_stream_calls = [
+            call(f"\nFetching {self.image_name}:latest Docker container image..."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("."),
+            call("\n"),
+        ]
 
         self.manager.pull_image(self.image_name, stream=stream)
 
         self.mock_docker_client.api.pull.assert_called_with(self.image_name, stream=True, decode=True, tag="latest")
-        self.assertEqual(stream.getvalue(), expected_stream_output)
+
+        stream.write_str.assert_has_calls(expected_stream_calls)
 
     def test_must_raise_if_image_not_found(self):
         msg = "some error"

tests/unit/local/services/test_base_local_service.py

@@ -66,17 +66,17 @@ def test_create_returns_not_implemented(self):
 class TestLambdaOutputParser(TestCase):
     @parameterized.expand(
         [
-            param("with mixed data and json response", b'data\n{"a": "b"}', 'data\n{"a": "b"}'),
-            param("with response as string", b"response", "response"),
-            param("with json response only", b'{"a": "b"}', '{"a": "b"}'),
-            param("with one new line and json", b'\n{"a": "b"}', '\n{"a": "b"}'),
-            param("with response only as string", b"this is the response line", "this is the response line"),
-            param("with whitespaces", b'data\n{"a": "b"}  \n\n\n', 'data\n{"a": "b"}  \n\n\n'),
-            param("with empty data", b"", ""),
-            param("with just new lines", b"\n\n", "\n\n"),
+            param("with mixed data and json response", 'data\n{"a": "b"}', 'data\n{"a": "b"}'),
+            param("with response as string", "response", "response"),
+            param("with json response only", '{"a": "b"}', '{"a": "b"}'),
+            param("with one new line and json", '\n{"a": "b"}', '\n{"a": "b"}'),
+            param("with response only as string", "this is the response line", "this is the response line"),
+            param("with whitespaces", 'data\n{"a": "b"}  \n\n\n', 'data\n{"a": "b"}  \n\n\n'),
+            param("with empty data", "", ""),
+            param("with just new lines", "\n\n", "\n\n"),
             param(
                 "with whitespaces",
-                b"\n   \n   \n",
+                "\n   \n   \n",
                 "\n   \n   \n",
             ),
         ]