Merge from aws/aws-sam-cli/develop

samcli/__init__.py

@@ -2,4 +2,4 @@
 SAM CLI version
 """
 
-__version__ = "1.57.0"
+__version__ = "1.58.0"

samcli/local/docker/container.py

@@ -396,13 +396,17 @@ def _write_container_output(output_itr, stdout=None, stderr=None):
             Stream writer to write stderr data from the Container into
         """
 
-        # Iterator returns a tuple of (stdout, stderr)
-        for stdout_data, stderr_data in output_itr:
-            if stdout_data and stdout:
-                stdout.write(stdout_data)
-
-            if stderr_data and stderr:
-                stderr.write(stderr_data)
+        # following iterator might throw an exception (see: https://github.com/aws/aws-sam-cli/issues/4222)
+        try:
+            # Iterator returns a tuple of (stdout, stderr)
+            for stdout_data, stderr_data in output_itr:
+                if stdout_data and stdout:
+                    stdout.write(stdout_data)
+
+                if stderr_data and stderr:
+                    stderr.write(stderr_data)
+        except Exception as ex:
+            LOG.debug("Failed to get the logs from the container", exc_info=ex)
 
     @property
     def network_id(self):

tests/integration/local/common_utils.py

@@ -0,0 +1,33 @@
+# Common utils between local tests
+import logging
+import random
+import time
+
+LOG = logging.getLogger(__name__)
+
+START_WAIT_TIME_SECONDS = 60
+
+
+class InvalidAddressException(Exception):
+    pass
+
+
+def wait_for_local_process(process, port):
+    start_time = time.time()
+    while True:
+        if time.time() - start_time > START_WAIT_TIME_SECONDS:
+            # Couldn't match any output string during the max allowed wait time
+            raise ValueError("Ran out of time attempting to start api/lambda process")
+        line = process.stderr.readline()
+        line_as_str = str(line.decode("utf-8")).strip()
+        if line_as_str:
+            LOG.info(f"{line_as_str}")
+        if "Address already in use" in line_as_str:
+            LOG.info(f"Attempted to start port on {port} but it is already in use, restarting on a new port.")
+            raise InvalidAddressException()
+        if "(Press CTRL+C to quit)" in line_as_str:
+            break
+
+
+def random_port():
+    return random.randint(30000, 40000)

tests/integration/local/start_api/start_api_integ_base.py

@@ -6,12 +6,12 @@
 from subprocess import Popen, PIPE
 import os
 import logging
-import random
 from pathlib import Path
 
 import docker
 from docker.errors import APIError
 
+from tests.integration.local.common_utils import InvalidAddressException, random_port, wait_for_local_process
 from tests.testing_utils import kill_process
 from tests.testing_utils import SKIP_DOCKER_MESSAGE, SKIP_DOCKER_TESTS, run_command
 
@@ -43,15 +43,13 @@ def setUpClass(cls):
         if cls.build_before_invoke:
             cls.build()
 
-        cls.port = str(StartApiIntegBaseClass.random_port())
-
         cls.docker_client = docker.from_env()
         for container in cls.docker_client.api.containers():
             try:
                 cls.docker_client.api.remove_container(container, force=True)
             except APIError as ex:
                 LOG.error("Failed to remove container %s", container, exc_info=ex)
-        cls.start_api()
+        cls.start_api_with_retry()
 
     @classmethod
     def build(cls):
@@ -67,6 +65,21 @@ def build(cls):
         working_dir = str(Path(cls.template).resolve().parents[0])
         run_command(command_list, cwd=working_dir)
 
+    @classmethod
+    def start_api_with_retry(cls, retries=3):
+        retry_count = 0
+        while retry_count < retries:
+            cls.port = str(random_port())
+            try:
+                cls.start_api()
+            except InvalidAddressException:
+                retry_count += 1
+                continue
+            break
+
+        if retry_count == retries:
+            raise ValueError("Ran out of retries attempting to start api")
+
     @classmethod
     def start_api(cls):
         command = "sam"
@@ -90,13 +103,7 @@ def start_api(cls):
 
         cls.start_api_process = Popen(command_list, stderr=PIPE)
 
-        while True:
-            line = cls.start_api_process.stderr.readline()
-            line_as_str = str(line.decode("utf-8")).strip()
-            if line_as_str:
-                LOG.info(f"{line_as_str}")
-            if "(Press CTRL+C to quit)" in line_as_str:
-                break
+        wait_for_local_process(cls.start_api_process, cls.port)
 
         cls.stop_reading_thread = False
 
@@ -117,10 +124,6 @@ def tearDownClass(cls):
         cls.stop_reading_thread = True
         kill_process(cls.start_api_process)
 
-    @staticmethod
-    def random_port():
-        return random.randint(30000, 40000)
-
     @staticmethod
     def get_binary_data(filename):
         if not filename:

tests/integration/local/start_lambda/start_lambda_api_integ_base.py

@@ -5,13 +5,13 @@
 import threading
 from subprocess import Popen, PIPE
 import os
-import random
 import logging
 from pathlib import Path
 
 import docker
 from docker.errors import APIError
 
+from tests.integration.local.common_utils import random_port, InvalidAddressException, wait_for_local_process
 from tests.testing_utils import (
     SKIP_DOCKER_TESTS,
     SKIP_DOCKER_MESSAGE,
@@ -41,7 +41,6 @@ def setUpClass(cls):
         # This is the directory for tests/integration which will be used to file the testdata
         # files for integ tests
         cls.template = cls.integration_dir + cls.template_path
-        cls.port = str(StartLambdaIntegBaseClass.random_port())
         cls.env_var_path = cls.integration_dir + "/testdata/invoke/vars.json"
 
         if cls.build_before_invoke:
@@ -55,7 +54,7 @@ def setUpClass(cls):
             except APIError as ex:
                 LOG.error("Failed to remove container %s", container, exc_info=ex)
 
-        cls.start_lambda()
+        cls.start_lambda_with_retry()
 
     @classmethod
     def build(cls):
@@ -72,7 +71,22 @@ def build(cls):
         run_command(command_list, cwd=working_dir)
 
     @classmethod
-    def start_lambda(cls, wait_time=5):
+    def start_lambda_with_retry(cls, retries=3):
+        retry_count = 0
+        while retry_count < retries:
+            cls.port = str(random_port())
+            try:
+                cls.start_lambda()
+            except InvalidAddressException:
+                retry_count += 1
+                continue
+            break
+
+        if retry_count == retries:
+            raise ValueError("Ran out of retries attempting to start lambda")
+
+    @classmethod
+    def start_lambda(cls):
         command = "sam"
         if os.getenv("SAM_CLI_DEV"):
             command = "samdev"
@@ -100,10 +114,7 @@ def start_lambda(cls, wait_time=5):
 
         cls.start_lambda_process = Popen(command_list, stderr=PIPE)
 
-        while True:
-            line = cls.start_lambda_process.stderr.readline()
-            if "(Press CTRL+C to quit)" in str(line):
-                break
+        wait_for_local_process(cls.start_lambda_process, cls.port)
 
         cls.stop_reading_thread = False
 
@@ -124,10 +135,6 @@ def tearDownClass(cls):
         cls.stop_reading_thread = True
         kill_process(cls.start_lambda_process)
 
-    @staticmethod
-    def random_port():
-        return random.randint(30000, 40000)
-
 
 class WatchWarmContainersIntegBaseClass(StartLambdaIntegBaseClass):
     temp_path: Optional[str] = None

tests/unit/local/docker/test_container.py

@@ -682,6 +682,19 @@ 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()
+
+        def _output_iterator():
+            yield "Hello", None
+            yield None, "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")])
+
 
 class TestContainer_wait_for_logs(TestCase):
     def setUp(self):