test_basic_function.py

import logging
from unittest.case import skipIf

import pytest
from parameterized import parameterized

from integration.config.service_names import (
    ARM,
    CODE_DEPLOY,
    EVENT_INVOKE_CONFIG,
    HTTP_API,
    KMS,
    LAMBDA_URL,
    XRAY,
)
from integration.helpers.base_test import BaseTest
from integration.helpers.resource import current_region_does_not_support

LOG = logging.getLogger(__name__)


class TestBasicFunction(BaseTest):
    """
    Basic AWS::Lambda::Function tests
    """

    @parameterized.expand(
        [
            "single/basic_function",
            "single/basic_function_no_envvar",
            "single/basic_function_openapi",
        ]
    )
    def test_basic_function(self, file_name):
        """
        Creates a basic lambda function
        """
        self.create_and_verify_stack(file_name)

        self.set_template_resource_property("MyLambdaFunction", "Timeout", 10)
        self.update_stack()

        self.assertEqual(self.get_resource_status_by_logical_id("MyLambdaFunction"), "UPDATE_COMPLETE")

    def test_basic_function_with_role_path(self):
        self.create_and_verify_stack("single/function_with_role_path")

        lambda_client = self.client_provider.lambda_client
        function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        role_name = self.get_physical_id_by_type("AWS::IAM::Role")
        response = lambda_client.get_function(FunctionName=function_name)

        role_arn = response.get("Configuration", {}).get("Role")
        self.assertIsNotNone(role_arn)
        self.assertIn("/foo/bar/", role_arn)

        iam_client = self.client_provider.iam_client
        response = iam_client.get_role(RoleName=role_name)

        self.assertEqual(response["Role"]["Path"], "/foo/bar/")

    @parameterized.expand(
        [
            "single/function_with_http_api_events",
            "single/function_alias_with_http_api_events",
        ]
    )
    @pytest.mark.flaky(reruns=5)
    @skipIf(current_region_does_not_support([HTTP_API]), "HTTP API is not supported in this testing region")
    def test_function_with_http_api_events(self, file_name):
        self.create_and_verify_stack(file_name)

        endpoint = self.get_api_v2_endpoint("MyHttpApi")

        self._verify_get_request(endpoint, self.FUNCTION_OUTPUT)

    @parameterized.expand(
        [
            ("single/basic_function", ["x86_64"]),
            ("single/basic_function_no_envvar", ["x86_64"]),
            ("single/basic_function_openapi", ["x86_64"]),
            ("single/basic_function_with_arm_architecture", ["arm64"]),
            ("single/basic_function_with_x86_architecture", ["x86_64"]),
        ]
    )
    @skipIf(current_region_does_not_support([ARM]), "ARM is not supported in this testing region")
    def test_basic_function_with_architecture(self, file_name, architecture):
        """
        Creates a basic lambda function
        """
        self.create_and_verify_stack(file_name)

        lambda_client = self.client_provider.lambda_client
        function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        function_architecture = lambda_client.get_function_configuration(FunctionName=function_name)["Architectures"]

        self.assertEqual(function_architecture, architecture)

    @parameterized.expand(
        [
            ("single/basic_function_with_function_url_config", None),
            ("single/basic_function_with_function_url_with_autopuplishalias", "live"),
        ]
    )
    @skipIf(current_region_does_not_support([LAMBDA_URL]), "Lambda Url is not supported in this testing region")
    def test_basic_function_with_url_config(self, file_name, qualifier):
        """
        Creates a basic lambda function with Function Url enabled
        """
        self.create_and_verify_stack(file_name)

        lambda_client = self.client_provider.lambda_client

        function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        function_url_config = (
            lambda_client.get_function_url_config(FunctionName=function_name, Qualifier=qualifier)
            if qualifier
            else lambda_client.get_function_url_config(FunctionName=function_name)
        )
        cors_config = {
            "AllowOrigins": ["https://foo.com"],
            "AllowMethods": ["POST"],
            "AllowCredentials": True,
            "AllowHeaders": ["x-custom-header"],
            "ExposeHeaders": ["x-amzn-header"],
            "MaxAge": 10,
        }

        self.assertEqual(function_url_config["AuthType"], "AWS_IAM")
        self.assertEqual(function_url_config["Cors"], cors_config)
        self._assert_invoke(lambda_client, function_name, qualifier, 200)

    @parameterized.expand(
        [
            ("single/basic_function_with_function_url_dual_auth", None),
            ("single/basic_function_with_function_url_with_autopuplishalias_dual_auth", "live"),
        ]
    )
    @skipIf(current_region_does_not_support([LAMBDA_URL]), "Lambda Url is not supported in this testing region")
    def test_basic_function_with_url_dual_auth(self, file_name, qualifier):
        """
        Creates a basic lambda function with Function Url with authtype: None
        Verifies that 2 AWS::Lambda::Permission resources are created:
        - lambda:InvokeFunctionUrl
        - lambda:InvokeFunction with InvokedViaFunctionUrl: True
        """
        self.create_and_verify_stack(file_name)

        # Get Lambda permissions
        lambda_permissions = self.get_stack_resources("AWS::Lambda::Permission")

        # Verify we have exactly 2 permissions
        self.assertEqual(len(lambda_permissions), 2, "Expected exactly 2 Lambda permissions")

        # Check for the expected permission logical IDs
        invoke_function_url_permission = None
        invoke_permission = None

        for permission in lambda_permissions:
            logical_id = permission["LogicalResourceId"]
            if "MyLambdaFunctionUrlPublicPermissions" in logical_id:
                invoke_function_url_permission = permission
            elif "MyLambdaFunctionURLInvokeAllowPublicAccess" in logical_id:
                invoke_permission = permission

        # Verify both permissions exist
        self.assertIsNotNone(invoke_function_url_permission, "Expected MyLambdaFunctionUrlPublicPermissions to exist")
        self.assertIsNotNone(invoke_permission, "Expected MyLambdaFunctionURLInvokeAllowPublicAccess to exist")

        # Get the function name and URL
        function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        lambda_client = self.client_provider.lambda_client

        # Get the function URL configuration to verify auth type
        function_url_config = (
            lambda_client.get_function_url_config(FunctionName=function_name, Qualifier=qualifier)
            if qualifier
            else lambda_client.get_function_url_config(FunctionName=function_name)
        )

        # Verify the auth type is NONE
        self.assertEqual(function_url_config["AuthType"], "NONE", "Expected AuthType to be NONE")

        # Get the template to check for InvokedViaFunctionUrl property
        cfn_client = self.client_provider.cfn_client
        template = cfn_client.get_template(StackName=self.stack_name)
        template_body = template["TemplateBody"]

        # Check if the InvokePermission has InvokedViaFunctionUrl: True
        # This is a bit hacky but we don't have direct access to the resource properties
        # We're checking if the string representation of the template contains this property
        template_str = str(template_body)
        self.assertIn("InvokedViaFunctionUrl", template_str, "Expected InvokedViaFunctionUrl property in the template")

        # Get the function URL from stack outputs
        function_url = self.get_stack_output("FunctionUrl")["OutputValue"]
        # Invoke the function URL and verify the response
        self._verify_get_request(function_url, self.FUNCTION_OUTPUT)

    @skipIf(current_region_does_not_support([CODE_DEPLOY]), "CodeDeploy is not supported in this testing region")
    def test_function_with_deployment_preference_alarms_intrinsic_if(self):
        self.create_and_verify_stack("single/function_with_deployment_preference_alarms_intrinsic_if")

    @parameterized.expand(
        [
            ("single/basic_function_with_sns_dlq", "sns:Publish"),
            ("single/basic_function_with_sqs_dlq", "sqs:SendMessage"),
        ]
    )
    def test_basic_function_with_dlq(self, file_name, action):
        """
        Creates a basic lambda function with dead letter queue policy
        """
        dlq_policy_name = "DeadLetterQueuePolicy"
        self.create_and_verify_stack(file_name)

        lambda_function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        function_configuration = self.client_provider.lambda_client.get_function_configuration(
            FunctionName=lambda_function_name
        )
        dlq_arn = function_configuration["DeadLetterConfig"]["TargetArn"]
        self.assertIsNotNone(dlq_arn, "DLQ Arn should be set")

        role_name = self.get_physical_id_by_type("AWS::IAM::Role")
        role_policy_result = self.client_provider.iam_client.get_role_policy(
            RoleName=role_name, PolicyName=dlq_policy_name
        )
        statements = role_policy_result["PolicyDocument"]["Statement"]

        self.assertEqual(len(statements), 1, "Only one statement must be in policy")
        self.assertEqual(statements[0]["Action"], action)
        self.assertEqual(statements[0]["Resource"], dlq_arn)
        self.assertEqual(statements[0]["Effect"], "Allow")

    @skipIf(current_region_does_not_support([KMS]), "KMS is not supported in this testing region")
    def test_basic_function_with_kms_key_arn(self):
        """
        Creates a basic lambda function with KMS key arn
        """
        self.create_and_verify_stack("single/basic_function_with_kmskeyarn")

        lambda_function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        function_configuration = self.client_provider.lambda_client.get_function_configuration(
            FunctionName=lambda_function_name
        )
        kms_key_arn = function_configuration["KMSKeyArn"]

        self.assertIsNotNone(kms_key_arn, "Expecting KmsKeyArn to be set.")

    def test_basic_function_with_tags(self):
        """
        Creates a basic lambda function with tags
        """
        self.create_and_verify_stack("single/basic_function_with_tags")
        lambda_function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
        get_function_result = self.client_provider.lambda_client.get_function(FunctionName=lambda_function_name)
        tags = get_function_result["Tags"]

        self.assertIsNotNone(tags, "Expecting tags on function.")
        self.assertTrue("lambda:createdBy" in tags, "Expected 'lambda:CreatedBy' tag key, but not found.")
        self.assertEqual("SAM", tags["lambda:createdBy"], "Expected 'SAM' tag value, but not found.")
        self.assertTrue("TagKey1" in tags)
        self.assertEqual(tags["TagKey1"], "TagValue1")
        self.assertTrue("TagKey2" in tags)
        self.assertEqual(tags["TagKey2"], "")

    @skipIf(
        current_region_does_not_support([EVENT_INVOKE_CONFIG]),
        "EventInvokeConfig is not supported in this testing region",
    )
    def test_basic_function_event_destinations(self):
        """
        Creates a basic lambda function with event destinations
        """
        self.create_and_verify_stack("single/basic_function_event_destinations")

        test_function_1 = self.get_physical_id_by_logical_id("MyTestFunction")
        test_function_2 = self.get_physical_id_by_logical_id("MyTestFunction2")

        function_invoke_config_result = self.client_provider.lambda_client.get_function_event_invoke_config(
            FunctionName=test_function_1, Qualifier="$LATEST"
        )
        self.assertIsNotNone(
            function_invoke_config_result["DestinationConfig"], "Expecting destination config to be set."
        )
        self.assertEqual(
            int(function_invoke_config_result["MaximumEventAgeInSeconds"]),
            70,
            "MaximumEventAgeInSeconds value is not set or incorrect.",
        )
        self.assertEqual(
            int(function_invoke_config_result["MaximumRetryAttempts"]),
            1,
            "MaximumRetryAttempts value is not set or incorrect.",
        )

        function_invoke_config_result = self.client_provider.lambda_client.get_function_event_invoke_config(
            FunctionName=test_function_2, Qualifier="live"
        )
        self.assertIsNotNone(
            function_invoke_config_result["DestinationConfig"], "Expecting destination config to be set."
        )
        self.assertEqual(
            int(function_invoke_config_result["MaximumEventAgeInSeconds"]),
            80,
            "MaximumEventAgeInSeconds value is not set or incorrect.",
        )
        self.assertEqual(
            int(function_invoke_config_result["MaximumRetryAttempts"]),
            2,
            "MaximumRetryAttempts value is not set or incorrect.",
        )

    @skipIf(current_region_does_not_support([XRAY]), "XRay is not supported in this testing region")
    def test_basic_function_with_tracing(self):
        """
        Creates a basic lambda function with tracing
        """
        self.create_and_verify_stack("single/basic_function_with_tracing", self.get_default_test_template_parameters())

        active_tracing_function_id = self.get_physical_id_by_logical_id("ActiveTracingFunction")
        pass_through_tracing_function_id = self.get_physical_id_by_logical_id("PassThroughTracingFunction")

        function_configuration_result = self.client_provider.lambda_client.get_function_configuration(
            FunctionName=active_tracing_function_id
        )
        self.assertIsNotNone(function_configuration_result["TracingConfig"], "Expecting tracing config to be set.")
        self.assertEqual(
            function_configuration_result["TracingConfig"]["Mode"],
            "Active",
            "Expecting tracing config mode to be set to Active.",
        )

        function_configuration_result = self.client_provider.lambda_client.get_function_configuration(
            FunctionName=pass_through_tracing_function_id
        )
        self.assertIsNotNone(function_configuration_result["TracingConfig"], "Expecting tracing config to be set.")
        self.assertEqual(
            function_configuration_result["TracingConfig"]["Mode"],
            "PassThrough",
            "Expecting tracing config mode to be set to PassThrough.",
        )

    # TODO: add the integration test back after the feature launch on 06/05
    # @skipIf(current_region_does_not_support([KMS]), "KMS is not supported in this testing region")
    # def test_basic_function_with_event_filtering_using_kms(self):
    #     """
    #     Creates a basic lambda function with KMS key arn
    #     """
    #     self.create_and_verify_stack("single/basic_function_with_event_filtering_using_kms")

    #     lambda_function_name = self.get_physical_id_by_type("AWS::Lambda::Function")
    #     event_source_mappings = self.client_provider.lambda_client.list_event_source_mappings(
    #         FunctionName=lambda_function_name
    #     )

    #     event_source_mapping = event_source_mappings["EventSourceMappings"][0]
    #     function_uuid = event_source_mapping["UUID"]

    #     event_source_mapping_config = self.client_provider.lambda_client.get_event_source_mapping(UUID=function_uuid)

    #     kms_key_arn = event_source_mapping_config["KMSKeyArn"]

    #     self.assertIsNotNone(kms_key_arn, "Expecting KmsKeyArn to be set.")

    def _assert_invoke(self, lambda_client, function_name, qualifier=None, expected_status_code=200):
        """
        Assert if a Lambda invocation returns the expected status code

        Parameters
        ----------
        lambda_client : boto3.BaseClient
            boto3 Lambda client
        function_name : string
            Function name
        qualifier : string
            Specify a version or alias to invoke a published version of the function
        expected_status_code : int
            Expected status code from the invocation
        """
        request_params = {
            "FunctionName": function_name,
            "Payload": "{}",
        }
        if qualifier:
            request_params["Qualifier"] = qualifier

        response = lambda_client.invoke(**request_params)
        self.assertEqual(response.get("StatusCode"), expected_status_code)

    def _verify_get_request(self, url, expected_text):
        response = self.verify_get_request_response(url, 200)
        self.assertEqual(response.text, expected_text)