test: Remote invoke integration tests for response stream configured lambda functions

tests/integration/remote/invoke/remote_invoke_integ_base.py

@@ -0,0 +1,105 @@
+from unittest import TestCase, skipIf
+from pathlib import Path
+from typing import Optional
+
+from tests.testing_utils import (
+    get_sam_command,
+    run_command,
+    RUNNING_ON_CI,
+    RUNNING_TEST_FOR_MASTER_ON_CI,
+    RUN_BY_CANARY,
+)
+from tests.integration.deploy.deploy_integ_base import DeployIntegBase
+
+from samcli.lib.utils.boto_utils import get_boto_resource_provider_with_config, get_boto_client_provider_with_config
+from samcli.lib.utils.cloudformation import get_resource_summaries
+
+SKIP_REMOTE_INVOKE_TESTS = RUNNING_ON_CI and RUNNING_TEST_FOR_MASTER_ON_CI and not RUN_BY_CANARY
+
+
+@skipIf(SKIP_REMOTE_INVOKE_TESTS, "Skip remote invoke tests in CI/CD only")
+class RemoteInvokeIntegBase(TestCase):
+    template: Optional[Path] = None
+
+    @classmethod
+    def setUpClass(cls):
+        cls.cmd = get_sam_command()
+        cls.test_data_path = cls.get_integ_dir().joinpath("testdata")
+        if cls.template:
+            cls.template_path = str(cls.test_data_path.joinpath("remote_invoke", cls.template))
+        cls.events_folder_path = cls.test_data_path.joinpath("remote_invoke", "events")
+
+    @staticmethod
+    def get_integ_dir():
+        return Path(__file__).resolve().parents[2]
+
+    @staticmethod
+    def remote_invoke_deploy_stack(stack_name, template_path):
+
+        deploy_cmd = DeployIntegBase.get_deploy_command_list(
+            stack_name=stack_name,
+            template_file=template_path,
+            resolve_s3=True,
+            capabilities_list=["CAPABILITY_IAM", "CAPABILITY_AUTO_EXPAND"],
+        )
+
+        run_command(deploy_cmd)
+
+    @classmethod
+    def create_resources_and_boto_clients(cls):
+        cls.remote_invoke_deploy_stack(cls.stack_name, cls.template_path)
+        stack_resource_summaries = get_resource_summaries(
+            get_boto_resource_provider_with_config(),
+            get_boto_client_provider_with_config(),
+            cls.stack_name,
+        )
+        cls.stack_resources = {
+            resource_full_path: stack_resource_summary.physical_resource_id
+            for resource_full_path, stack_resource_summary in stack_resource_summaries.items()
+        }
+        cls.cfn_client = get_boto_client_provider_with_config()("cloudformation")
+        cls.lambda_client = get_boto_client_provider_with_config()("lambda")
+
+    @staticmethod
+    def get_command_list(
+        stack_name=None,
+        resource_id=None,
+        event=None,
+        event_file=None,
+        parameter_list=None,
+        output=None,
+        region=None,
+        profile=None,
+        beta_features=None,
+    ):
+        command_list = [get_sam_command(), "remote", "invoke"]
+
+        if stack_name:
+            command_list = command_list + ["--stack-name", stack_name]
+
+        if event:
+            command_list = command_list + ["-e", event]
+
+        if event_file:
+            command_list = command_list + ["--event-file", event_file]
+
+        if profile:
+            command_list = command_list + ["--parameter", parameter]
+
+        if output:
+            command_list = command_list + ["--output", output]
+
+        if parameter_list:
+            for (parameter, value) in parameter_list:
+                command_list = command_list + ["--parameter", f"{parameter}={value}"]
+
+        if region:
+            command_list = command_list + ["--region", region]
+
+        if beta_features is not None:
+            command_list = command_list + ["--beta-features" if beta_features else "--no-beta-features"]
+
+        if resource_id:
+            command_list = command_list + [resource_id]
+
+        return command_list

tests/integration/remote/invoke/test_lambda_invoke_response_stream.py

@@ -0,0 +1,108 @@
+import json
+import uuid
+
+from tests.integration.remote.invoke.remote_invoke_integ_base import RemoteInvokeIntegBase
+from tests.testing_utils import run_command
+
+from pathlib import Path
+import pytest
+
+
+@pytest.mark.xdist_group(name="sam_remote_invoke_lambda_response_streaming")
+class TestInvokeResponseStreamingLambdas(RemoteInvokeIntegBase):
+    template = Path("template-lambda-response-streaming-fns.yaml")
+
+    @classmethod
+    def tearDownClass(cls):
+        # Delete the deployed stack
+        cls.cfn_client.delete_stack(StackName=cls.stack_name)
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.stack_name = f"{TestInvokeResponseStreamingLambdas.__name__}-{uuid.uuid4().hex}"
+        cls.create_resources_and_boto_clients()
+
+    def test_invoke_empty_event_provided(self):
+        command_list = self.get_command_list(stack_name=self.stack_name, resource_id="NodeStreamingFunction")
+
+        expected_streamed_responses = "LambdaFunctionStreamingResponsesTestDone!"
+        remote_invoke_result = run_command(command_list)
+
+        self.assertEqual(0, remote_invoke_result.process.returncode)
+        remote_invoke_result_stdout = remote_invoke_result.stdout.strip().decode()
+        self.assertIn(expected_streamed_responses, remote_invoke_result_stdout)
+
+    def test_invoke_with_only_event_provided(self):
+        command_list = self.get_command_list(
+            stack_name=self.stack_name,
+            resource_id="NodeStreamingFunction",
+            event='{"key1": "Hello", "key2": "serverless", "key3": "world"}',
+        )
+
+        expected_streamed_responses = "LambdaFunctionStreamingResponsesTestDone!"
+        remote_invoke_result = run_command(command_list)
+
+        self.assertEqual(0, remote_invoke_result.process.returncode)
+        remote_invoke_result_stdout = remote_invoke_result.stdout.strip().decode()
+        self.assertIn(expected_streamed_responses, remote_invoke_result_stdout)
+
+    def test_invoke_with_only_event_file_provided(self):
+        event_file_path = str(self.events_folder_path.joinpath("default_event.json"))
+        command_list = self.get_command_list(
+            stack_name=self.stack_name, resource_id="NodeStreamingEventValuesFunction", event_file=event_file_path
+        )
+
+        expected_streamed_responses = "Helloserverlessworld"
+        remote_invoke_result = run_command(command_list)
+
+        self.assertEqual(0, remote_invoke_result.process.returncode)
+        remote_invoke_result_stdout = remote_invoke_result.stdout.strip().decode()
+
+        self.assertEqual(expected_streamed_responses, remote_invoke_result_stdout)
+
+    def test_invoke_json_output_option(self):
+        command_list = self.get_command_list(
+            stack_name=self.stack_name,
+            event='{"key1": "Hello", "key2": "serverless", "key3": "world"}',
+            resource_id="NodeStreamingEventValuesFunction",
+            output="json",
+            parameter_list=[("LogType", "None")],
+        )
+
+        remote_invoke_result = run_command(command_list)
+        expected_output_result = [
+            {"PayloadChunk": {"Payload": "Hello"}},
+            {"PayloadChunk": {"Payload": "serverless"}},
+            {"PayloadChunk": {"Payload": "world"}},
+            {"InvokeComplete": {}},
+        ]
+
+        self.assertEqual(0, remote_invoke_result.process.returncode)
+        remote_invoke_result_stdout = json.loads(remote_invoke_result.stdout.strip().decode())
+
+        response_event_stream = remote_invoke_result_stdout["EventStream"]
+        self.assertEqual(response_event_stream, expected_output_result)
+
+    def test_invoke_different_boto_options(self):
+        command_list = self.get_command_list(
+            stack_name=self.stack_name,
+            event='{"key1": "Hello", "key2": "serverless", "key3": "world"}',
+            resource_id="NodeStreamingEventValuesFunction",
+            output="json",
+            parameter_list=[("LogType", "None"), ("InvocationType", "DryRun"), ("Qualifier", "$LATEST")],
+        )
+
+        remote_invoke_result = run_command(command_list)
+        expected_output_result = [
+            {"PayloadChunk": {"Payload": "Hello"}},
+            {"PayloadChunk": {"Payload": "serverless"}},
+            {"PayloadChunk": {"Payload": "world"}},
+            {"InvokeComplete": {}},
+        ]
+
+        self.assertEqual(0, remote_invoke_result.process.returncode)
+        remote_invoke_result_stdout = json.loads(remote_invoke_result.stdout.strip().decode())
+
+        response_event_stream = remote_invoke_result_stdout["EventStream"]
+        self.assertEqual(response_event_stream, expected_output_result)

tests/integration/testdata/remote_invoke/lambda-fns/src/index.js

@@ -0,0 +1,26 @@
+exports.handler = awslambda.streamifyResponse(
+    async (event, responseStream, context) => {
+        responseStream.write("Lambda");
+        responseStream.write("Function");
+
+        responseStream.write("Streaming");
+        await new Promise(r => setTimeout(r, 1000));
+        responseStream.write("Responses");
+        await new Promise(r => setTimeout(r, 1000));
+        responseStream.write("Test");
+        await new Promise(r => setTimeout(r, 1000));
+
+        responseStream.write("Done!");
+        responseStream.end();
+    }
+);
+
+exports.stream_event_values = awslambda.streamifyResponse(
+    async (event, responseStream, context) => {
+        for (let k in event) {
+            responseStream.write(event[k]);
+            await new Promise(r => setTimeout(r, 1000));
+          }
+        responseStream.end();
+    }
+);

tests/integration/testdata/remote_invoke/template-lambda-response-streaming-fns.yaml

@@ -0,0 +1,32 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform: AWS::Serverless-2016-10-31
+
+Description: >
+  Testing application for lambda functions with response streaming
+
+Resources:
+  NodeStreamingFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ./lambda-fns/src/
+      Handler: index.handler
+      Runtime: nodejs18.x
+      Architectures:
+        - x86_64
+      Timeout: 10
+      FunctionUrlConfig:
+        AuthType: AWS_IAM
+        InvokeMode: RESPONSE_STREAM
+
+  NodeStreamingEventValuesFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: ./lambda-fns/src/
+      Handler: index.stream_event_values
+      Runtime: nodejs18.x
+      Architectures:
+        - x86_64
+      Timeout: 10
+      FunctionUrlConfig:
+        AuthType: AWS_IAM
+        InvokeMode: RESPONSE_STREAM