Unlock 30% off on Manual Testing Annual Plans this Holiday Season.
Copied to Clipboard!
How to use code_signing_arn method in localstack

Best Python code snippet using localstack_python
test_lambda.py
Source:test_lambda.py
1import base642import json3import logging4import os5import re6import shutil7import time8from datetime import datetime9from io import BytesIO10import pytest11from localstack import config12from localstack.constants import LAMBDA_TEST_ROLE, TEST_AWS_ACCOUNT_ID13from localstack.services.awslambda import lambda_api14from localstack.services.awslambda.lambda_api import (15    LAMBDA_DEFAULT_HANDLER,16    get_lambda_policy_name,17    use_docker,18)19from localstack.services.awslambda.lambda_utils import (20    LAMBDA_RUNTIME_DOTNETCORE2,21    LAMBDA_RUNTIME_DOTNETCORE31,22    LAMBDA_RUNTIME_GOLANG,23    LAMBDA_RUNTIME_JAVA8,24    LAMBDA_RUNTIME_JAVA8_AL2,25    LAMBDA_RUNTIME_JAVA11,26    LAMBDA_RUNTIME_NODEJS10X,27    LAMBDA_RUNTIME_NODEJS12X,28    LAMBDA_RUNTIME_NODEJS14X,29    LAMBDA_RUNTIME_PROVIDED,30    LAMBDA_RUNTIME_PROVIDED_AL2,31    LAMBDA_RUNTIME_PYTHON36,32    LAMBDA_RUNTIME_PYTHON37,33    LAMBDA_RUNTIME_PYTHON38,34    LAMBDA_RUNTIME_PYTHON39,35    LAMBDA_RUNTIME_RUBY27,36)37from localstack.services.install import (38    GO_RUNTIME_VERSION,39    INSTALL_PATH_LOCALSTACK_FAT_JAR,40    TEST_LAMBDA_JAVA,41    download_and_extract,42)43from localstack.utils import testutil44from localstack.utils.aws import aws_stack45from localstack.utils.common import (46    cp_r,47    get_arch,48    get_os,49    load_file,50    mkdir,51    new_tmp_dir,52    retry,53    run_safe,54    save_file,55    short_uid,56    to_bytes,57    to_str,58    unzip,59)60from localstack.utils.generic.wait_utils import wait_until61from localstack.utils.testutil import (62    check_expected_lambda_log_events_length,63    create_lambda_archive,64    get_lambda_log_events,65)66from .functions import lambda_integration67THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))68TEST_LAMBDA_PYTHON = os.path.join(THIS_FOLDER, "functions", "lambda_integration.py")69TEST_LAMBDA_PYTHON_ECHO = os.path.join(THIS_FOLDER, "functions", "lambda_echo.py")70TEST_LAMBDA_PYTHON_VERSION = os.path.join(THIS_FOLDER, "functions", "lambda_python_version.py")71TEST_LAMBDA_PYTHON3 = os.path.join(THIS_FOLDER, "functions", "lambda_python3.py")72TEST_LAMBDA_INTEGRATION_NODEJS = os.path.join(THIS_FOLDER, "functions", "lambda_integration.js")73TEST_LAMBDA_NODEJS = os.path.join(THIS_FOLDER, "functions", "lambda_handler.js")74TEST_LAMBDA_GOLANG_ZIP = os.path.join(THIS_FOLDER, "functions", "golang", "handler.zip")75TEST_LAMBDA_RUBY = os.path.join(THIS_FOLDER, "functions", "lambda_integration.rb")76TEST_LAMBDA_DOTNETCORE2 = os.path.join(THIS_FOLDER, "functions", "dotnetcore2", "dotnetcore2.zip")77TEST_LAMBDA_DOTNETCORE31 = os.path.join(78    THIS_FOLDER, "functions", "dotnetcore31", "dotnetcore31.zip"79)80TEST_LAMBDA_CUSTOM_RUNTIME = os.path.join(THIS_FOLDER, "functions", "custom-runtime")81TEST_LAMBDA_JAVA_WITH_LIB = os.path.join(82    THIS_FOLDER, "functions", "java", "lambda_echo", "lambda-function-with-lib-0.0.1.jar"83)84TEST_LAMBDA_JAVA_MULTIPLE_HANDLERS = os.path.join(85    THIS_FOLDER,86    "functions",87    "java",88    "lambda_multiple_handlers",89    "build",90    "distributions",91    "lambda-function-with-multiple-handlers.zip",92)93TEST_LAMBDA_ENV = os.path.join(THIS_FOLDER, "functions", "lambda_environment.py")94TEST_LAMBDA_SEND_MESSAGE_FILE = os.path.join(THIS_FOLDER, "functions", "lambda_send_message.py")95TEST_LAMBDA_PUT_ITEM_FILE = os.path.join(THIS_FOLDER, "functions", "lambda_put_item.py")96TEST_LAMBDA_START_EXECUTION_FILE = os.path.join(97    THIS_FOLDER, "functions", "lambda_start_execution.py"98)99TEST_LAMBDA_FUNCTION_PREFIX = "lambda-function"100TEST_GOLANG_LAMBDA_URL_TEMPLATE = "https://github.com/localstack/awslamba-go-runtime/releases/download/v{version}/example-handler-{os}-{arch}.tar.gz"101TEST_LAMBDA_LIBS = [102    "localstack_client",103    "requests",104    "psutil",105    "urllib3",106    "chardet",107    "certifi",108    "idna",109    "pip",110    "dns",111]112PYTHON_TEST_RUNTIMES = (113    [114        LAMBDA_RUNTIME_PYTHON36,115        LAMBDA_RUNTIME_PYTHON37,116        LAMBDA_RUNTIME_PYTHON38,117        LAMBDA_RUNTIME_PYTHON39,118    ]119    if use_docker()120    else [LAMBDA_RUNTIME_PYTHON38]121)122NODE_TEST_RUNTIMES = (123    [124        LAMBDA_RUNTIME_NODEJS10X,125        LAMBDA_RUNTIME_NODEJS12X,126        LAMBDA_RUNTIME_NODEJS14X,127    ]128    if use_docker()129    else [LAMBDA_RUNTIME_NODEJS14X]130)131JAVA_TEST_RUNTIMES = (132    [133        LAMBDA_RUNTIME_JAVA8,134        LAMBDA_RUNTIME_JAVA8_AL2,135        LAMBDA_RUNTIME_JAVA11,136    ]137    if use_docker()138    else [LAMBDA_RUNTIME_JAVA11]139)140PROVIDED_TEST_RUNTIMES = [141    LAMBDA_RUNTIME_PROVIDED,142    # TODO remove skip once we use correct images143    pytest.param(144        LAMBDA_RUNTIME_PROVIDED_AL2, marks=pytest.mark.skip("curl missing in provided.al2 image")145    ),146]147@pytest.fixture148def check_lambda_logs(logs_client):149    def _check_logs(func_name, expected_lines=None):150        if not expected_lines:151            expected_lines = []152        log_events = get_lambda_logs(func_name, logs_client=logs_client)153        log_messages = [e["message"] for e in log_events]154        for line in expected_lines:155            if ".*" in line:156                found = [re.match(line, m) for m in log_messages]157                if any(found):158                    continue159            assert line in log_messages160    return _check_logs161def get_lambda_logs(func_name, logs_client=None):162    logs_client = logs_client or aws_stack.create_external_boto_client("logs")163    log_group_name = "/aws/lambda/%s" % func_name164    streams = logs_client.describe_log_streams(logGroupName=log_group_name)["logStreams"]165    streams = sorted(streams, key=lambda x: x["creationTime"], reverse=True)166    log_events = logs_client.get_log_events(167        logGroupName=log_group_name, logStreamName=streams[0]["logStreamName"]168    )["events"]169    return log_events170# API only functions (no lambda execution itself)171class TestLambdaAPI:172    def test_create_lambda_function(self, lambda_client):173        """Basic test that creates and deletes a Lambda function"""174        func_name = f"lambda_func-{short_uid()}"175        kms_key_arn = f"arn:{aws_stack.get_partition()}:kms:{aws_stack.get_region()}:{TEST_AWS_ACCOUNT_ID}:key11"176        vpc_config = {177            "SubnetIds": ["subnet-123456789"],178            "SecurityGroupIds": ["sg-123456789"],179        }180        tags = {"env": "testing"}181        kwargs = {182            "FunctionName": func_name,183            "Runtime": LAMBDA_RUNTIME_PYTHON37,184            "Handler": LAMBDA_DEFAULT_HANDLER,185            "Role": LAMBDA_TEST_ROLE,186            "KMSKeyArn": kms_key_arn,187            "Code": {188                "ZipFile": create_lambda_archive(189                    load_file(TEST_LAMBDA_PYTHON_ECHO), get_content=True190                )191            },192            "Timeout": 3,193            "VpcConfig": vpc_config,194            "Tags": tags,195            "Environment": {"Variables": {"foo": "bar"}},196        }197        result = lambda_client.create_function(**kwargs)198        function_arn = result["FunctionArn"]199        assert testutil.response_arn_matches_partition(lambda_client, function_arn)200        partial_function_arn = ":".join(function_arn.split(":")[3:])201        # Get function by Name, ARN and partial ARN202        for func_ref in [func_name, function_arn, partial_function_arn]:203            rs = lambda_client.get_function(FunctionName=func_ref)204            assert rs["Configuration"].get("KMSKeyArn", "") == kms_key_arn205            assert rs["Configuration"].get("VpcConfig", {}) == vpc_config206            assert rs["Tags"] == tags207        # clean up208        lambda_client.delete_function(FunctionName=func_name)209        with pytest.raises(Exception) as exc:210            lambda_client.delete_function(FunctionName=func_name)211        assert "ResourceNotFoundException" in str(exc)212    def test_add_lambda_permission(self, lambda_client, iam_client, create_lambda_function):213        function_name = f"lambda_func-{short_uid()}"214        create_lambda_function(215            handler_file=TEST_LAMBDA_PYTHON_ECHO,216            func_name=function_name,217            runtime=LAMBDA_RUNTIME_PYTHON36,218        )219        # create lambda permission220        action = "lambda:InvokeFunction"221        sid = "s3"222        principal = "s3.amazonaws.com"223        resp = lambda_client.add_permission(224            FunctionName=function_name,225            Action=action,226            StatementId=sid,227            Principal=principal,228            SourceArn=aws_stack.s3_bucket_arn("test-bucket"),229        )230        assert "Statement" in resp231        # fetch lambda policy232        policy = lambda_client.get_policy(FunctionName=function_name)["Policy"]233        assert isinstance(policy, str)234        policy = json.loads(to_str(policy))235        assert action == policy["Statement"][0]["Action"]236        assert sid == policy["Statement"][0]["Sid"]237        assert lambda_api.func_arn(function_name) == policy["Statement"][0]["Resource"]238        assert principal == policy["Statement"][0]["Principal"]["Service"]239        assert (240            aws_stack.s3_bucket_arn("test-bucket")241            == policy["Statement"][0]["Condition"]["ArnLike"]["AWS:SourceArn"]242        )243        # fetch IAM policy244        policies = iam_client.list_policies(Scope="Local", MaxItems=500)["Policies"]245        policy_name = get_lambda_policy_name(function_name)246        matching = [p for p in policies if p["PolicyName"] == policy_name]247        assert len(matching) == 1248        assert ":policy/" in matching[0]["Arn"]249        # remove permission that we just added250        resp = lambda_client.remove_permission(251            FunctionName=function_name,252            StatementId=sid,253            Qualifier="qual1",254            RevisionId="r1",255        )256        assert 200 == resp["ResponseMetadata"]["HTTPStatusCode"]257    def test_add_lambda_multiple_permission(258        self, iam_client, lambda_client, create_lambda_function259    ):260        function_name = f"lambda_func-{short_uid()}"261        create_lambda_function(262            handler_file=TEST_LAMBDA_PYTHON_ECHO,263            func_name=function_name,264            runtime=LAMBDA_RUNTIME_PYTHON36,265        )266        # create lambda permissions267        action = "lambda:InvokeFunction"268        principal = "s3.amazonaws.com"269        statement_ids = ["s4", "s5"]270        for sid in statement_ids:271            resp = lambda_client.add_permission(272                FunctionName=function_name,273                Action=action,274                StatementId=sid,275                Principal=principal,276                SourceArn=aws_stack.s3_bucket_arn("test-bucket"),277            )278            assert "Statement" in resp279        # fetch IAM policy280        policies = iam_client.list_policies(Scope="Local", MaxItems=500)["Policies"]281        policy_name = get_lambda_policy_name(function_name)282        matching = [p for p in policies if p["PolicyName"] == policy_name]283        assert 1 == len(matching)284        assert ":policy/" in matching[0]["Arn"]285        # validate both statements286        policy = matching[0]287        versions = iam_client.list_policy_versions(PolicyArn=policy["Arn"])["Versions"]288        assert 1 == len(versions)289        statements = versions[0]["Document"]["Statement"]290        for i in range(len(statement_ids)):291            assert action == statements[i]["Action"]292            assert lambda_api.func_arn(function_name) == statements[i]["Resource"]293            assert principal == statements[i]["Principal"]["Service"]294            assert (295                aws_stack.s3_bucket_arn("test-bucket")296                == statements[i]["Condition"]["ArnLike"]["AWS:SourceArn"]297            )298            # check statement_ids in reverse order299            assert statement_ids[abs(i - 1)] == statements[i]["Sid"]300        # remove permission that we just added301        resp = lambda_client.remove_permission(302            FunctionName=function_name,303            StatementId=sid,304            Qualifier="qual1",305            RevisionId="r1",306        )307        assert 200 == resp["ResponseMetadata"]["HTTPStatusCode"]308    def test_lambda_asynchronous_invocations(self, lambda_client, create_lambda_function):309        function_name = f"lambda_func-{short_uid()}"310        create_lambda_function(311            handler_file=TEST_LAMBDA_PYTHON_ECHO,312            func_name=function_name,313            runtime=LAMBDA_RUNTIME_PYTHON36,314        )315        # adding event invoke config316        response = lambda_client.put_function_event_invoke_config(317            FunctionName=function_name,318            MaximumRetryAttempts=123,319            MaximumEventAgeInSeconds=123,320            DestinationConfig={321                "OnSuccess": {"Destination": function_name},322                "OnFailure": {"Destination": function_name},323            },324        )325        destination_config = {326            "OnSuccess": {"Destination": function_name},327            "OnFailure": {"Destination": function_name},328        }329        # checking for parameter configuration330        assert 123 == response["MaximumRetryAttempts"]331        assert 123 == response["MaximumEventAgeInSeconds"]332        assert destination_config == response["DestinationConfig"]333        # over writing event invoke config334        response = lambda_client.put_function_event_invoke_config(335            FunctionName=function_name,336            MaximumRetryAttempts=123,337            DestinationConfig={338                "OnSuccess": {"Destination": function_name},339                "OnFailure": {"Destination": function_name},340            },341        )342        # checking if 'MaximumEventAgeInSeconds' is removed343        assert "MaximumEventAgeInSeconds" not in response344        assert isinstance(response["LastModified"], datetime)345        # updating event invoke config346        response = lambda_client.update_function_event_invoke_config(347            FunctionName=function_name,348            MaximumRetryAttempts=111,349        )350        # checking for updated and existing configuration351        assert 111 == response["MaximumRetryAttempts"]352        assert destination_config == response["DestinationConfig"]353        # clean up354        lambda_client.delete_function_event_invoke_config(FunctionName=function_name)355    def test_function_concurrency(self, lambda_client, create_lambda_function):356        function_name = f"lambda_func-{short_uid()}"357        create_lambda_function(358            handler_file=TEST_LAMBDA_PYTHON_ECHO,359            func_name=function_name,360            runtime=LAMBDA_RUNTIME_PYTHON36,361        )362        response = lambda_client.put_function_concurrency(363            FunctionName=function_name, ReservedConcurrentExecutions=123364        )365        assert "ReservedConcurrentExecutions" in response366        response = lambda_client.get_function_concurrency(FunctionName=function_name)367        assert "ReservedConcurrentExecutions" in response368        response = lambda_client.delete_function_concurrency(FunctionName=function_name)369        assert "ReservedConcurrentExecutions" not in response370    def test_function_code_signing_config(self, lambda_client, create_lambda_function):371        function_name = f"lambda_func-{short_uid()}"372        create_lambda_function(373            handler_file=TEST_LAMBDA_PYTHON_ECHO,374            func_name=function_name,375            runtime=LAMBDA_RUNTIME_PYTHON36,376        )377        response = lambda_client.create_code_signing_config(378            Description="Testing CodeSigning Config",379            AllowedPublishers={380                "SigningProfileVersionArns": [381                    "arn:aws:signer:%s:000000000000:/signing-profiles/test"382                    % aws_stack.get_region(),383                ]384            },385            CodeSigningPolicies={"UntrustedArtifactOnDeployment": "Enforce"},386        )387        assert "Description" in response["CodeSigningConfig"]388        assert "SigningProfileVersionArns" in response["CodeSigningConfig"]["AllowedPublishers"]389        assert (390            "UntrustedArtifactOnDeployment" in response["CodeSigningConfig"]["CodeSigningPolicies"]391        )392        code_signing_arn = response["CodeSigningConfig"]["CodeSigningConfigArn"]393        response = lambda_client.update_code_signing_config(394            CodeSigningConfigArn=code_signing_arn,395            CodeSigningPolicies={"UntrustedArtifactOnDeployment": "Warn"},396        )397        assert (398            "Warn"399            == response["CodeSigningConfig"]["CodeSigningPolicies"]["UntrustedArtifactOnDeployment"]400        )401        response = lambda_client.get_code_signing_config(CodeSigningConfigArn=code_signing_arn)402        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]403        response = lambda_client.put_function_code_signing_config(404            CodeSigningConfigArn=code_signing_arn, FunctionName=function_name405        )406        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]407        response = lambda_client.get_function_code_signing_config(FunctionName=function_name)408        assert 200 == response["ResponseMetadata"]["HTTPStatusCode"]409        assert code_signing_arn == response["CodeSigningConfigArn"]410        assert function_name == response["FunctionName"]411        response = lambda_client.delete_function_code_signing_config(FunctionName=function_name)412        assert 204 == response["ResponseMetadata"]["HTTPStatusCode"]413        response = lambda_client.delete_code_signing_config(CodeSigningConfigArn=code_signing_arn)414        assert 204 == response["ResponseMetadata"]["HTTPStatusCode"]415    def create_multiple_lambda_permissions(self, lambda_client, iam_client, create_lambda_function):416        role_name = f"role-{short_uid()}"417        function_name = f"test-function-{short_uid()}"418        assume_policy_document = {419            "Version": "2012-10-17",420            "Statement": [421                {422                    "Action": "sts:AssumeRole",423                    "Principal": {"Service": "lambda.amazonaws.com"},424                }425            ],426        }427        iam_client.create_role(428            RoleName=role_name,429            AssumeRolePolicyDocument=json.dumps(assume_policy_document),430        )431        create_lambda_function(432            funct_name=function_name,433            runtime=LAMBDA_RUNTIME_PYTHON37,434            libs=TEST_LAMBDA_LIBS,435        )436        action = "lambda:InvokeFunction"437        sid = "logs"438        resp = lambda_client.add_permission(439            FunctionName=function_name,440            Action=action,441            StatementId=sid,442            Principal="logs.amazonaws.com",443        )444        assert "Statement" in resp445        sid = "kinesis"446        resp = lambda_client.add_permission(447            FunctionName=function_name,448            Action=action,449            StatementId=sid,450            Principal="kinesis.amazonaws.com",451        )452        assert "Statement" in resp453class TestLambdaBaseFeatures:454    def test_dead_letter_queue(455        self, lambda_client, create_lambda_function, sqs_client, sqs_create_queue, sqs_queue_arn456    ):457        # create DLQ and Lambda function458        queue_name = f"test-{short_uid()}"459        lambda_name = f"test-{short_uid()}"460        queue_url = sqs_create_queue(QueueName=queue_name)461        queue_arn = sqs_queue_arn(queue_url)462        create_lambda_function(463            handler_file=TEST_LAMBDA_PYTHON,464            func_name=lambda_name,465            libs=TEST_LAMBDA_LIBS,466            runtime=LAMBDA_RUNTIME_PYTHON36,467            DeadLetterConfig={"TargetArn": queue_arn},468        )469        # invoke Lambda, triggering an error470        payload = {lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1}471        lambda_client.invoke(472            FunctionName=lambda_name,473            Payload=json.dumps(payload),474            InvocationType="Event",475        )476        # assert that message has been received on the DLQ477        def receive_dlq():478            result = sqs_client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])479            assert len(result["Messages"]) > 0480            msg_attrs = result["Messages"][0]["MessageAttributes"]481            assert "RequestID" in msg_attrs482            assert "ErrorCode" in msg_attrs483            assert "ErrorMessage" in msg_attrs484        retry(receive_dlq, retries=10, sleep=2)485        # update DLQ config486        lambda_client.update_function_configuration(FunctionName=lambda_name, DeadLetterConfig={})487        # invoke Lambda again, assert that status code is 200 and error details contained in the payload488        result = lambda_client.invoke(489            FunctionName=lambda_name, Payload=json.dumps(payload), LogType="Tail"490        )491        payload = json.loads(to_str(result["Payload"].read()))492        assert 200 == result["StatusCode"]493        assert "Unhandled" == result["FunctionError"]494        assert "$LATEST" == result["ExecutedVersion"]495        assert "Test exception" in payload["errorMessage"]496        assert "Exception" in payload["errorType"]497        assert isinstance(payload["stackTrace"], list)498        log_result = result.get("LogResult")499        assert log_result500        logs = to_str(base64.b64decode(to_str(log_result)))501        assert "START" in logs502        assert "Test exception" in logs503        assert "END" in logs504        assert "REPORT" in logs505    @pytest.mark.parametrize(506        "condition,payload",507        [508            ("Success", {}),509            ("RetriesExhausted", {lambda_integration.MSG_BODY_RAISE_ERROR_FLAG: 1}),510        ],511    )512    def test_assess_lambda_destination_invocation(513        self,514        condition,515        payload,516        lambda_client,517        sqs_client,518        create_lambda_function,519        sqs_create_queue,520        sqs_queue_arn,521    ):522        # create DLQ and Lambda function523        queue_name = f"test-{short_uid()}"524        lambda_name = f"test-{short_uid()}"525        queue_url = sqs_create_queue(QueueName=queue_name)526        queue_arn = sqs_queue_arn(queue_url)527        create_lambda_function(528            handler_file=TEST_LAMBDA_PYTHON,529            func_name=lambda_name,530            libs=TEST_LAMBDA_LIBS,531        )532        lambda_client.put_function_event_invoke_config(533            FunctionName=lambda_name,534            DestinationConfig={535                "OnSuccess": {"Destination": queue_arn},536                "OnFailure": {"Destination": queue_arn},537            },538        )539        lambda_client.invoke(540            FunctionName=lambda_name,541            Payload=json.dumps(payload),542            InvocationType="Event",543        )544        def receive_message():545            rs = sqs_client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])546            assert len(rs["Messages"]) > 0547            msg = rs["Messages"][0]["Body"]548            msg = json.loads(msg)549            assert condition == msg["requestContext"]["condition"]550        retry(receive_message, retries=10, sleep=3)551    def test_large_payloads(self, caplog, lambda_client, create_lambda_function):552        # Set the loglevel to INFO for this test to avoid breaking a CI environment (due to excessive log outputs)553        caplog.set_level(logging.INFO)554        function_name = f"large_payload-{short_uid()}"555        create_lambda_function(556            handler_file=TEST_LAMBDA_PYTHON_ECHO,557            func_name=function_name,558            runtime=LAMBDA_RUNTIME_PYTHON36,559        )560        payload = {"test": "test123456" * 100 * 1000 * 5}  # 5MB payload561        payload_bytes = to_bytes(json.dumps(payload))562        result = lambda_client.invoke(FunctionName=function_name, Payload=payload_bytes)563        assert 200 == result["ResponseMetadata"]["HTTPStatusCode"]564        result_data = result["Payload"].read()565        result_data = json.loads(to_str(result_data))566        assert payload == result_data567parametrize_python_runtimes = pytest.mark.parametrize(568    "runtime",569    PYTHON_TEST_RUNTIMES,570)571class TestPythonRuntimes:572    @pytest.fixture(573        params=PYTHON_TEST_RUNTIMES,574    )575    def python_function_name(self, request, lambda_client, create_lambda_function):576        function_name = f"python-test-function-{short_uid()}"577        create_lambda_function(578            func_name=function_name,579            handler_file=TEST_LAMBDA_PYTHON,580            libs=TEST_LAMBDA_LIBS,581            runtime=request.param,582        )583        return function_name584    def test_invocation_type_not_set(self, lambda_client, python_function_name):585        result = lambda_client.invoke(586            FunctionName=python_function_name, Payload=b"{}", LogType="Tail"587        )588        result_data = json.loads(result["Payload"].read())589        # assert response details590        assert 200 == result["StatusCode"]591        assert {} == result_data["event"]592        # assert that logs are contained in response593        logs = result.get("LogResult", "")594        logs = to_str(base64.b64decode(to_str(logs)))595        assert "START" in logs596        assert "Lambda log message" in logs597        assert "END" in logs598        assert "REPORT" in logs599    def test_invocation_type_request_response(self, lambda_client, python_function_name):600        result = lambda_client.invoke(601            FunctionName=python_function_name,602            Payload=b"{}",603            InvocationType="RequestResponse",604        )605        result_data = result["Payload"].read()606        result_data = json.loads(to_str(result_data))607        assert "application/json" == result["ResponseMetadata"]["HTTPHeaders"]["content-type"]608        assert 200 == result["StatusCode"]609        assert isinstance(result_data, dict)610    def test_invocation_type_event(self, lambda_client, python_function_name):611        result = lambda_client.invoke(612            FunctionName=python_function_name, Payload=b"{}", InvocationType="Event"613        )614        assert 202 == result["StatusCode"]615    def test_invocation_type_dry_run(self, lambda_client, python_function_name):616        result = lambda_client.invoke(617            FunctionName=python_function_name, Payload=b"{}", InvocationType="DryRun"618        )619        assert 204 == result["StatusCode"]620    @parametrize_python_runtimes621    def test_lambda_environment(self, lambda_client, create_lambda_function, runtime):622        function_name = f"env-test-function-{short_uid()}"623        env_vars = {"Hello": "World"}624        create_lambda_function(625            handler_file=TEST_LAMBDA_ENV,626            libs=TEST_LAMBDA_LIBS,627            func_name=function_name,628            envvars=env_vars,629            runtime=runtime,630        )631        # invoke function and assert result contains env vars632        result = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")633        result_data = result["Payload"]634        assert 200 == result["StatusCode"]635        assert json.load(result_data) == env_vars636        # get function config and assert result contains env vars637        result = lambda_client.get_function_configuration(FunctionName=function_name)638        assert result["Environment"] == {"Variables": env_vars}639    @parametrize_python_runtimes640    def test_invocation_with_qualifier(641        self, lambda_client, s3_client, s3_bucket, runtime, check_lambda_logs642    ):643        function_name = f"test_lambda_{short_uid()}"644        bucket_key = "test_lambda.zip"645        # upload zip file to S3646        zip_file = create_lambda_archive(647            load_file(TEST_LAMBDA_PYTHON), get_content=True, libs=TEST_LAMBDA_LIBS, runtime=runtime648        )649        s3_client.upload_fileobj(BytesIO(zip_file), s3_bucket, bucket_key)650        # create lambda function651        response = lambda_client.create_function(652            FunctionName=function_name,653            Runtime=runtime,654            Role="r1",  # TODO655            Publish=True,656            Handler="handler.handler",657            Code={"S3Bucket": s3_bucket, "S3Key": bucket_key},658            Timeout=10,659        )660        assert "Version" in response661        # invoke lambda function662        data_before = b'{"foo": "bar with \'quotes\\""}'663        result = lambda_client.invoke(664            FunctionName=function_name, Payload=data_before, Qualifier=response["Version"]665        )666        data_after = json.loads(result["Payload"].read())667        assert json.loads(to_str(data_before)) == data_after["event"]668        context = data_after["context"]669        assert response["Version"] == context["function_version"]670        assert context.get("aws_request_id")671        assert function_name == context["function_name"]672        assert "/aws/lambda/%s" % function_name == context["log_group_name"]673        assert context.get("log_stream_name")674        assert context.get("memory_limit_in_mb")675        # assert that logs are present676        expected = ["Lambda log message - print function"]677        if use_docker():678            # Note that during regular test execution, nosetests captures the output from679            # the logging module - hence we can only expect this when running in Docker680            expected.append(".*Lambda log message - logging module")681        check_lambda_logs(function_name, expected_lines=expected)682        lambda_client.delete_function(FunctionName=function_name)683    @parametrize_python_runtimes684    def test_upload_lambda_from_s3(self, lambda_client, s3_client, s3_bucket, runtime):685        function_name = f"test_lambda_{short_uid()}"686        bucket_key = "test_lambda.zip"687        # upload zip file to S3688        zip_file = testutil.create_lambda_archive(689            load_file(TEST_LAMBDA_PYTHON), get_content=True, libs=TEST_LAMBDA_LIBS, runtime=runtime690        )691        s3_client.upload_fileobj(BytesIO(zip_file), s3_bucket, bucket_key)692        # create lambda function693        lambda_client.create_function(694            FunctionName=function_name,695            Runtime=runtime,696            Handler="handler.handler",697            Role="r1",698            Code={"S3Bucket": s3_bucket, "S3Key": bucket_key},699            Timeout=10,700        )701        # invoke lambda function702        data_before = b'{"foo": "bar with \'quotes\\""}'703        result = lambda_client.invoke(FunctionName=function_name, Payload=data_before)704        data_after = json.loads(result["Payload"].read())705        assert json.loads(to_str(data_before)) == data_after["event"]706        context = data_after["context"]707        assert "$LATEST" == context["function_version"]708        assert function_name == context["function_name"]709        # clean up710        lambda_client.delete_function(FunctionName=function_name)711    @parametrize_python_runtimes712    def test_handler_in_submodule(self, lambda_client, create_lambda_function, runtime):713        function_name = f"test-function-{short_uid()}"714        zip_file = testutil.create_lambda_archive(715            load_file(TEST_LAMBDA_PYTHON),716            get_content=True,717            libs=TEST_LAMBDA_LIBS,718            runtime=runtime,719            file_name="localstack_package/def/main.py",720        )721        create_lambda_function(722            func_name=function_name,723            zip_file=zip_file,724            handler="localstack_package.def.main.handler",725            runtime=runtime,726        )727        # invoke function and assert result728        result = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")729        result_data = json.loads(result["Payload"].read())730        assert 200 == result["StatusCode"]731        assert json.loads("{}") == result_data["event"]732    @parametrize_python_runtimes733    def test_lambda_send_message_to_sqs(734        self, lambda_client, create_lambda_function, sqs_client, sqs_create_queue, runtime735    ):736        function_name = f"test-function-{short_uid()}"737        queue_name = f"lambda-queue-{short_uid()}"738        queue_url = sqs_create_queue(QueueName=queue_name)739        create_lambda_function(740            handler_file=TEST_LAMBDA_SEND_MESSAGE_FILE,741            func_name=function_name,742            runtime=runtime,743        )744        event = {745            "message": f"message-from-test-lambda-{short_uid()}",746            "queue_name": queue_name,747            "region_name": config.DEFAULT_REGION,748        }749        lambda_client.invoke(FunctionName=function_name, Payload=json.dumps(event))750        # assert that message has been received on the Queue751        def receive_message():752            rs = sqs_client.receive_message(QueueUrl=queue_url, MessageAttributeNames=["All"])753            assert len(rs["Messages"]) > 0754            return rs["Messages"][0]755        message = retry(receive_message, retries=3, sleep=2)756        assert event["message"] == message["Body"]757    @parametrize_python_runtimes758    def test_lambda_put_item_to_dynamodb(759        self, lambda_client, create_lambda_function, dynamodb_create_table, runtime760    ):761        table_name = f"ddb-table-{short_uid()}"762        function_name = f"test-function-{short_uid()}"763        dynamodb_create_table(table_name=table_name, partition_key="id")764        create_lambda_function(765            handler_file=TEST_LAMBDA_PUT_ITEM_FILE,766            func_name=function_name,767            runtime=runtime,768        )769        data = {short_uid(): f"data-{i}" for i in range(3)}770        event = {771            "table_name": table_name,772            "region_name": config.DEFAULT_REGION,773            "items": [{"id": k, "data": v} for k, v in data.items()],774        }775        lambda_client.invoke(FunctionName=function_name, Payload=json.dumps(event))776        dynamodb = aws_stack.connect_to_resource("dynamodb")  # TODO convert to fixture777        rs = dynamodb.Table(table_name).scan()778        items = rs["Items"]779        assert len(items) == len(data.keys())780        for item in items:781            assert data[item["id"]] == item["data"]782    @parametrize_python_runtimes783    def test_lambda_start_stepfunctions_execution(784        self, lambda_client, stepfunctions_client, create_lambda_function, runtime785    ):786        function_name = f"test-function-{short_uid()}"787        resource_lambda_name = f"test-resource-{short_uid()}"788        state_machine_name = f"state-machine-{short_uid()}"789        create_lambda_function(790            handler_file=TEST_LAMBDA_START_EXECUTION_FILE,791            func_name=function_name,792            runtime=runtime,793        )794        create_lambda_function(795            handler_file=TEST_LAMBDA_PYTHON_ECHO,796            func_name=resource_lambda_name,797            runtime=runtime,798        )799        state_machine_def = {800            "StartAt": "step1",801            "States": {802                "step1": {803                    "Type": "Task",804                    "Resource": aws_stack.lambda_function_arn(resource_lambda_name),805                    "ResultPath": "$.result_value",806                    "End": True,807                }808            },809        }810        rs = stepfunctions_client.create_state_machine(811            name=state_machine_name,812            definition=json.dumps(state_machine_def),813            roleArn=aws_stack.role_arn("sfn_role"),814        )815        sm_arn = rs["stateMachineArn"]816        try:817            lambda_client.invoke(818                FunctionName=function_name,819                Payload=json.dumps(820                    {821                        "state_machine_arn": sm_arn,822                        "region_name": config.DEFAULT_REGION,823                        "input": {},824                    }825                ),826            )827            time.sleep(1)828            rs = stepfunctions_client.list_executions(stateMachineArn=sm_arn)829            # assert that state machine get executed 1 time830            assert 1 == len([ex for ex in rs["executions"] if ex["stateMachineArn"] == sm_arn])831        finally:832            # clean up833            stepfunctions_client.delete_state_machine(stateMachineArn=sm_arn)834    @pytest.mark.skipif(835        not use_docker(), reason="Test for docker python runtimes not applicable if run locally"836    )837    @parametrize_python_runtimes838    def test_python_runtime_correct_versions(self, lambda_client, create_lambda_function, runtime):839        function_name = f"test_python_executor_{short_uid()}"840        create_lambda_function(841            func_name=function_name,842            handler_file=TEST_LAMBDA_PYTHON_VERSION,843            runtime=runtime,844        )845        result = lambda_client.invoke(846            FunctionName=function_name,847            Payload=b"{}",848        )849        result = json.loads(to_str(result["Payload"].read()))850        assert result["version"] == runtime851parametrize_node_runtimes = pytest.mark.parametrize(852    "runtime",853    NODE_TEST_RUNTIMES,854)855class TestNodeJSRuntimes:856    @pytest.mark.skipif(857        not use_docker(), reason="Test for docker nodejs runtimes not applicable if run locally"858    )859    @parametrize_node_runtimes860    def test_nodejs_lambda_with_context(861        self, lambda_client, create_lambda_function, runtime, check_lambda_logs862    ):863        function_name = f"test-function-{short_uid()}"864        create_lambda_function(865            func_name=function_name,866            handler_file=TEST_LAMBDA_INTEGRATION_NODEJS,867            handler="lambda_integration.handler",868            runtime=runtime,869        )870        ctx = {871            "custom": {"foo": "bar"},872            "client": {"snap": ["crackle", "pop"]},873            "env": {"fizz": "buzz"},874        }875        result = lambda_client.invoke(876            FunctionName=function_name,877            Payload=b"{}",878            ClientContext=to_str(base64.b64encode(to_bytes(json.dumps(ctx)))),879        )880        result_data = result["Payload"].read()881        assert 200 == result["StatusCode"]882        assert "bar" == json.loads(json.loads(result_data)["context"]["clientContext"]).get(883            "custom"884        ).get("foo")885        # assert that logs are present886        expected = [".*Node.js Lambda handler executing."]887        check_lambda_logs(function_name, expected_lines=expected)888    @parametrize_node_runtimes889    def test_invoke_nodejs_lambda(self, lambda_client, create_lambda_function, runtime):890        function_name = f"test-function-{short_uid()}"891        create_lambda_function(892            func_name=function_name,893            zip_file=testutil.create_zip_file(TEST_LAMBDA_NODEJS, get_content=True),894            runtime=runtime,895            handler="lambda_handler.handler",896        )897        rs = lambda_client.invoke(898            FunctionName=function_name,899            Payload=json.dumps({"event_type": "test_lambda"}),900        )901        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]902        payload = rs["Payload"].read()903        response = json.loads(to_str(payload))904        assert "response from localstack lambda" in response["body"]905        events = get_lambda_log_events(function_name)906        assert len(events) > 0907    @parametrize_node_runtimes908    def test_invoke_nodejs_lambda_with_payload_containing_quotes(909        self, lambda_client, create_lambda_function, runtime910    ):911        function_name = "test_lambda_%s" % short_uid()912        create_lambda_function(913            func_name=function_name,914            zip_file=testutil.create_zip_file(TEST_LAMBDA_NODEJS, get_content=True),915            runtime=runtime,916            handler="lambda_handler.handler",917        )918        test_string = "test_string' with some quotes"919        body = '{"test_var": "%s"}' % test_string920        rs = lambda_client.invoke(921            FunctionName=function_name,922            Payload=body,923        )924        assert 200 == rs["ResponseMetadata"]["HTTPStatusCode"]925        payload = rs["Payload"].read()926        response = json.loads(to_str(payload))927        assert "response from localstack lambda" in response["body"]928        events = get_lambda_log_events(function_name)929        assert len(events) > 0930        assert test_string in str(events[0])931class TestCustomRuntimes:932    @pytest.mark.skipif(933        not use_docker(), reason="Test for docker provided runtimes not applicable if run locally"934    )935    @pytest.mark.parametrize(936        "runtime",937        PROVIDED_TEST_RUNTIMES,938    )939    def test_provided_runtimes(940        self, lambda_client, create_lambda_function, runtime, check_lambda_logs941    ):942        function_name = f"test-function-{short_uid()}"943        create_lambda_function(944            func_name=function_name,945            handler_file=TEST_LAMBDA_CUSTOM_RUNTIME,946            handler="function.handler",947            runtime=runtime,948        )949        result = lambda_client.invoke(950            FunctionName=function_name,951            Payload=b'{"text": "bar with \'quotes\\""}',952        )953        result_data = result["Payload"].read()954        assert 200 == result["StatusCode"]955        result_data = to_str(result_data).strip()956        # jsonify in pro (re-)formats the event json so we allow both versions here957        assert result_data in (958            """Echoing request: '{"text": "bar with \'quotes\\""}'""",959            """Echoing request: '{"text":"bar with \'quotes\\""}'""",960        )961        # assert that logs are present962        expected = [".*Custom Runtime Lambda handler executing."]963        check_lambda_logs(function_name, expected_lines=expected)964class TestDotNetCoreRuntimes:965    @pytest.mark.skipif(966        not use_docker(), reason="Dotnet functions only supported with docker executor"967    )968    @pytest.mark.parametrize(969        "zip_file,handler,runtime,expected_lines",970        [971            (972                TEST_LAMBDA_DOTNETCORE2,973                "DotNetCore2::DotNetCore2.Lambda.Function::SimpleFunctionHandler",974                LAMBDA_RUNTIME_DOTNETCORE2,975                ["Running .NET Core 2.0 Lambda"],976            ),977            (978                TEST_LAMBDA_DOTNETCORE31,979                "dotnetcore31::dotnetcore31.Function::FunctionHandler",980                LAMBDA_RUNTIME_DOTNETCORE31,981                ["Running .NET Core 3.1 Lambda"],982            ),983        ],984    )985    def test_dotnet_lambda(986        self, zip_file, handler, runtime, expected_lines, lambda_client, create_lambda_function987    ):988        function_name = f"test-function-{short_uid()}"989        create_lambda_function(990            func_name=function_name,991            zip_file=load_file(zip_file, mode="rb"),992            handler=handler,993            runtime=runtime,994        )995        result = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")996        result_data = result["Payload"].read()997        assert 200 == result["StatusCode"]998        assert "{}" == to_str(result_data).strip()999        # TODO make lambda log checks more resilient to various formats1000        # self.check_lambda_logs(func_name, expected_lines=expected_lines)1001class TestRubyRuntimes:1002    @pytest.mark.skipif(not use_docker(), reason="ruby runtimes not supported in local invocation")1003    def test_ruby_lambda_running_in_docker(self, lambda_client, create_lambda_function):1004        function_name = f"test-function-{short_uid()}"1005        create_lambda_function(1006            func_name=function_name,1007            handler_file=TEST_LAMBDA_RUBY,1008            handler="lambda_integration.handler",1009            runtime=LAMBDA_RUNTIME_RUBY27,1010        )1011        result = lambda_client.invoke(FunctionName=function_name, Payload=b"{}")1012        result_data = result["Payload"].read()1013        assert 200 == result["StatusCode"]1014        assert "{}" == to_str(result_data).strip()1015class TestGolangRuntimes:1016    def test_golang_lambda(self, lambda_client, tmp_path, create_lambda_function):1017        # fetch platform-specific example handler1018        url = TEST_GOLANG_LAMBDA_URL_TEMPLATE.format(1019            version=GO_RUNTIME_VERSION,1020            os=get_os(),1021            arch=get_arch(),1022        )1023        handler = tmp_path / "go-handler"1024        download_and_extract(url, handler)1025        # create function1026        func_name = f"test_lambda_{short_uid()}"1027        create_lambda_function(1028            func_name=func_name,1029            handler_file=handler,1030            handler="handler",1031            runtime=LAMBDA_RUNTIME_GOLANG,1032        )1033        # invoke1034        result = lambda_client.invoke(1035            FunctionName=func_name, Payload=json.dumps({"name": "pytest"})1036        )1037        result_data = result["Payload"].read()1038        assert result["StatusCode"] == 2001039        assert to_str(result_data).strip() == '"Hello pytest!"'1040parametrize_java_runtimes = pytest.mark.parametrize(1041    "runtime",1042    JAVA_TEST_RUNTIMES,1043)1044class TestJavaRuntimes:1045    @pytest.fixture(scope="class")1046    def test_java_jar(self) -> bytes:1047        # The TEST_LAMBDA_JAVA jar file is downloaded with `make init-testlibs`.1048        java_file = load_file(TEST_LAMBDA_JAVA, mode="rb")1049        if not java_file:1050            raise Exception(1051                f"Test dependency {TEST_LAMBDA_JAVA} not found."1052                "Please make sure to run 'make init-testlibs' to ensure the file is available."1053            )1054        return java_file1055    @pytest.fixture(scope="class")1056    def test_java_zip(self, tmpdir_factory, test_java_jar) -> bytes:1057        tmpdir = tmpdir_factory.mktemp("tmp-java-zip")1058        zip_lib_dir = os.path.join(tmpdir, "lib")1059        zip_jar_path = os.path.join(zip_lib_dir, "test.lambda.jar")1060        mkdir(zip_lib_dir)1061        cp_r(1062            INSTALL_PATH_LOCALSTACK_FAT_JAR,1063            os.path.join(zip_lib_dir, "executor.lambda.jar"),1064        )1065        save_file(zip_jar_path, test_java_jar)1066        return testutil.create_zip_file(tmpdir, get_content=True)1067    @pytest.fixture(1068        params=JAVA_TEST_RUNTIMES,1069    )1070    def simple_java_lambda(self, create_lambda_function, test_java_zip, request):1071        function_name = f"java-test-function-{short_uid()}"1072        create_lambda_function(1073            func_name=function_name,1074            zip_file=test_java_zip,1075            runtime=request.param,1076            handler="cloud.localstack.sample.LambdaHandler",1077        )1078        return function_name1079    def test_java_runtime(self, lambda_client, simple_java_lambda):1080        result = lambda_client.invoke(1081            FunctionName=simple_java_lambda,1082            Payload=b'{"echo":"echo"}',1083        )1084        result_data = result["Payload"].read()1085        assert 200 == result["StatusCode"]1086        # TODO: find out why the assertion below does not work in Travis-CI! (seems to work locally)1087        assert "LinkedHashMap" in to_str(result_data)1088        assert result_data is not None1089    def test_java_runtime_with_large_payload(self, lambda_client, simple_java_lambda, caplog):1090        # Set the loglevel to INFO for this test to avoid breaking a CI environment (due to excessive log outputs)1091        caplog.set_level(logging.INFO)1092        payload = {"test": "test123456" * 100 * 1000 * 5}  # 5MB payload1093        payload = to_bytes(json.dumps(payload))1094        result = lambda_client.invoke(FunctionName=simple_java_lambda, Payload=payload)1095        result_data = result["Payload"].read()1096        assert 200 == result["StatusCode"]1097        assert "LinkedHashMap" in to_str(result_data)1098        assert result_data is not None1099    def test_java_runtime_with_lib(self, lambda_client, create_lambda_function):1100        java_jar_with_lib = load_file(TEST_LAMBDA_JAVA_WITH_LIB, mode="rb")1101        # create ZIP file from JAR file1102        jar_dir = new_tmp_dir()1103        zip_dir = new_tmp_dir()1104        unzip(TEST_LAMBDA_JAVA_WITH_LIB, jar_dir)1105        zip_lib_dir = os.path.join(zip_dir, "lib")1106        shutil.move(os.path.join(jar_dir, "lib"), zip_lib_dir)1107        jar_without_libs_file = testutil.create_zip_file(jar_dir)1108        shutil.copy(jar_without_libs_file, os.path.join(zip_lib_dir, "lambda.jar"))1109        java_zip_with_lib = testutil.create_zip_file(zip_dir, get_content=True)1110        java_zip_with_lib_gradle = load_file(1111            os.path.join(1112                THIS_FOLDER,1113                "functions",1114                "java",1115                "lambda_echo",1116                "build",1117                "distributions",1118                "lambda-function-built-by-gradle.zip",1119            ),1120            mode="rb",1121        )1122        for archive in [java_jar_with_lib, java_zip_with_lib, java_zip_with_lib_gradle]:1123            lambda_name = "test-function-%s" % short_uid()1124            create_lambda_function(1125                func_name=lambda_name,1126                zip_file=archive,1127                runtime=LAMBDA_RUNTIME_JAVA11,1128                handler="cloud.localstack.sample.LambdaHandlerWithLib",1129            )1130            result = lambda_client.invoke(FunctionName=lambda_name, Payload=b'{"echo":"echo"}')1131            result_data = result["Payload"].read()1132            assert 200 == result["StatusCode"]1133            assert "echo" in to_str(result_data)1134    def test_sns_event(self, lambda_client, simple_java_lambda):1135        result = lambda_client.invoke(1136            FunctionName=simple_java_lambda,1137            InvocationType="Event",1138            Payload=b'{"Records": [{"Sns": {"Message": "{}"}}]}',1139        )1140        assert 202 == result["StatusCode"]1141    def test_ddb_event(self, lambda_client, simple_java_lambda):1142        result = lambda_client.invoke(1143            FunctionName=simple_java_lambda,1144            InvocationType="Event",1145            Payload=b'{"Records": [{"dynamodb": {"Message": "{}"}}]}',1146        )1147        assert 202 == result["StatusCode"]1148    @parametrize_java_runtimes1149    def test_kinesis_invocation(1150        self, lambda_client, create_lambda_function, test_java_zip, runtime1151    ):1152        payload = (1153            b'{"Records": [{'1154            b'"kinesis": {"data": "dGVzdA==", "partitionKey": "partition"},'1155            b'"eventID": "shardId-000000000001:12345678901234567890123456789012345678901234567890",'1156            b'"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/test"}]}'1157        )1158        # deploy lambda - Java with Kinesis input object1159        function_name = f"test-lambda-{short_uid()}"1160        create_lambda_function(1161            func_name=function_name,1162            zip_file=test_java_zip,1163            runtime=runtime,1164            handler="cloud.localstack.awssdkv1.sample.KinesisLambdaHandler",1165        )1166        result = lambda_client.invoke(FunctionName=function_name, Payload=payload)1167        result_data = result["Payload"].read()1168        assert 200 == result["StatusCode"]1169        assert '"test "' == to_str(result_data).strip()1170    def test_kinesis_event(self, lambda_client, simple_java_lambda):1171        payload = (1172            b'{"Records": [{'1173            b'"kinesis": {"data": "dGVzdA==", "partitionKey": "partition"},'1174            b'"eventID": "shardId-000000000001:12345678901234567890123456789012345678901234567890",'1175            b'"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/test"}]}'1176        )1177        result = lambda_client.invoke(1178            FunctionName=simple_java_lambda,1179            InvocationType="Event",1180            Payload=payload,1181        )1182        result_data = result["Payload"].read()1183        assert 202 == result["StatusCode"]1184        assert "" == to_str(result_data).strip()1185    @parametrize_java_runtimes1186    def test_stream_handler(self, lambda_client, create_lambda_function, test_java_jar, runtime):1187        function_name = f"test-lambda-{short_uid()}"1188        create_lambda_function(1189            func_name=function_name,1190            zip_file=test_java_jar,1191            runtime=runtime,1192            handler="cloud.localstack.awssdkv1.sample.LambdaStreamHandler",1193        )1194        result = lambda_client.invoke(1195            FunctionName=function_name,1196            Payload=b'{"echo":"echo"}',1197        )1198        result_data = result["Payload"].read()1199        assert 200 == result["StatusCode"]1200        assert "{}" == to_str(result_data).strip()1201    @parametrize_java_runtimes1202    def test_serializable_input_object(1203        self, lambda_client, create_lambda_function, test_java_zip, runtime1204    ):1205        # deploy lambda - Java with serializable input object1206        function_name = f"test-lambda-{short_uid()}"1207        create_lambda_function(1208            func_name=function_name,1209            zip_file=test_java_zip,1210            runtime=runtime,1211            handler="cloud.localstack.awssdkv1.sample.SerializedInputLambdaHandler",1212        )1213        result = lambda_client.invoke(1214            FunctionName=function_name,1215            Payload=b'{"bucket": "test_bucket", "key": "test_key"}',1216        )1217        result_data = result["Payload"].read()1218        assert 200 == result["StatusCode"]1219        assert json.loads(to_str(result_data)) == {1220            "validated": True,1221            "bucket": "test_bucket",1222            "key": "test_key",1223        }1224    def test_trigger_java_lambda_through_sns(1225        self, lambda_client, s3_client, sns_client, simple_java_lambda, s3_bucket, sns_create_topic1226    ):1227        topic_name = "topic-%s" % short_uid()1228        key = "key-%s" % short_uid()1229        function_name = simple_java_lambda1230        topic_arn = sns_create_topic(Name=topic_name)["TopicArn"]1231        s3_client.put_bucket_notification_configuration(1232            Bucket=s3_bucket,1233            NotificationConfiguration={1234                "TopicConfigurations": [{"TopicArn": topic_arn, "Events": ["s3:ObjectCreated:*"]}]1235            },1236        )1237        sns_client.subscribe(1238            TopicArn=topic_arn,1239            Protocol="lambda",1240            Endpoint=aws_stack.lambda_function_arn(function_name),1241        )1242        events_before = run_safe(get_lambda_log_events, function_name, regex_filter="Records") or []1243        s3_client.put_object(Bucket=s3_bucket, Key=key, Body="something")1244        time.sleep(2)1245        # We got an event that confirm lambda invoked1246        retry(1247            function=check_expected_lambda_log_events_length,1248            retries=3,1249            sleep=1,1250            expected_length=len(events_before) + 1,1251            function_name=function_name,1252            regex_filter="Records",1253        )1254        # clean up1255        s3_client.delete_objects(Bucket=s3_bucket, Delete={"Objects": [{"Key": key}]})1256    @pytest.mark.parametrize(1257        "handler,expected_result",1258        [1259            (1260                "cloud.localstack.sample.LambdaHandlerWithInterfaceAndCustom::handleRequestCustom",1261                "CUSTOM",1262            ),1263            ("cloud.localstack.sample.LambdaHandlerWithInterfaceAndCustom", "INTERFACE"),1264            (1265                "cloud.localstack.sample.LambdaHandlerWithInterfaceAndCustom::handleRequest",1266                "INTERFACE",1267            ),1268        ],1269    )1270    # this test is only compiled against java 111271    def test_java_custom_handler_method_specification(1272        self, lambda_client, create_lambda_function, handler, expected_result, check_lambda_logs1273    ):1274        java_handler_multiple_handlers = load_file(TEST_LAMBDA_JAVA_MULTIPLE_HANDLERS, mode="rb")1275        expected = ['.*"echo": "echo".*']1276        function_name = "lambda_handler_test_%s" % short_uid()1277        create_lambda_function(1278            func_name=function_name,1279            zip_file=java_handler_multiple_handlers,1280            runtime=LAMBDA_RUNTIME_JAVA11,1281            handler=handler,1282        )1283        result = lambda_client.invoke(FunctionName=function_name, Payload=b'{"echo":"echo"}')1284        result_data = result["Payload"].read()1285        assert 200 == result["StatusCode"]1286        assert expected_result == to_str(result_data).strip('"\n ')1287        check_lambda_logs(function_name, expected_lines=expected)1288TEST_LAMBDA_CACHE_NODEJS = os.path.join(THIS_FOLDER, "functions", "lambda_cache.js")1289TEST_LAMBDA_CACHE_PYTHON = os.path.join(THIS_FOLDER, "functions", "lambda_cache.py")1290TEST_LAMBDA_TIMEOUT_PYTHON = os.path.join(THIS_FOLDER, "functions", "lambda_timeout.py")1291class TestLambdaBehavior:1292    @pytest.mark.parametrize(1293        ["lambda_fn", "lambda_runtime"],1294        [1295            (1296                TEST_LAMBDA_CACHE_NODEJS,1297                LAMBDA_RUNTIME_NODEJS12X,1298            ),  # TODO: can we do some kind of nested parametrize here?1299            (TEST_LAMBDA_CACHE_PYTHON, LAMBDA_RUNTIME_PYTHON38),1300        ],1301        ids=["nodejs", "python"],1302    )1303    @pytest.mark.xfail(1304        os.environ.get("TEST_TARGET") != "AWS_CLOUD",1305        reason="lambda caching not supported currently",1306    )  # TODO: should be removed after the lambda rework1307    def test_lambda_cache_local(1308        self, lambda_client, create_lambda_function, lambda_fn, lambda_runtime1309    ):1310        """tests the local context reuse of packages in AWS lambda"""1311        func_name = f"test_lambda_{short_uid()}"1312        create_lambda_function(1313            func_name=func_name,1314            handler_file=lambda_fn,1315            runtime=lambda_runtime,1316            client=lambda_client,1317        )1318        result = lambda_client.invoke(FunctionName=func_name)1319        result_data = result["Payload"].read()1320        assert result["StatusCode"] == 2001321        assert json.loads(result_data)["counter"] == 01322        result = lambda_client.invoke(FunctionName=func_name)1323        result_data = result["Payload"].read()1324        assert result["StatusCode"] == 2001325        assert json.loads(result_data)["counter"] == 11326    @pytest.mark.parametrize(1327        ["lambda_fn", "lambda_runtime"],1328        [1329            (TEST_LAMBDA_TIMEOUT_PYTHON, LAMBDA_RUNTIME_PYTHON38),1330        ],1331        ids=["python"],1332    )1333    @pytest.mark.xfail(1334        os.environ.get("TEST_TARGET") != "AWS_CLOUD",1335        reason="lambda timeouts not supported currently",1336    )  # TODO: should be removed after the lambda rework1337    def test_lambda_timeout_logs(1338        self, lambda_client, create_lambda_function, lambda_fn, lambda_runtime, logs_client1339    ):1340        """tests the local context reuse of packages in AWS lambda"""1341        func_name = f"test_lambda_{short_uid()}"1342        create_lambda_function(1343            func_name=func_name,1344            handler_file=lambda_fn,1345            runtime=lambda_runtime,1346            client=lambda_client,1347            timeout=1,1348        )1349        result = lambda_client.invoke(FunctionName=func_name, Payload=json.dumps({"wait": 2}))1350        assert result["StatusCode"] == 2001351        log_group_name = f"/aws/lambda/{func_name}"1352        ls_result = logs_client.describe_log_streams(logGroupName=log_group_name)1353        log_stream_name = ls_result["logStreams"][0]["logStreamName"]1354        log_events = logs_client.get_log_events(1355            logGroupName=log_group_name, logStreamName=log_stream_name1356        )["events"]1357        assert any(["starting wait" in e["message"] for e in log_events])1358        assert not any(["done waiting" in e["message"] for e in log_events])1359    @pytest.mark.parametrize(1360        ["lambda_fn", "lambda_runtime"],1361        [1362            (TEST_LAMBDA_TIMEOUT_PYTHON, LAMBDA_RUNTIME_PYTHON38),1363        ],1364        ids=["python"],1365    )1366    def test_lambda_no_timeout_logs(1367        self, lambda_client, create_lambda_function, lambda_fn, lambda_runtime, logs_client1368    ):1369        """tests the local context reuse of packages in AWS lambda"""1370        func_name = f"test_lambda_{short_uid()}"1371        create_lambda_function(1372            func_name=func_name,1373            handler_file=lambda_fn,1374            runtime=lambda_runtime,1375            client=lambda_client,1376            timeout=2,1377        )1378        result = lambda_client.invoke(FunctionName=func_name, Payload=json.dumps({"wait": 1}))1379        assert result["StatusCode"] == 2001380        log_group_name = f"/aws/lambda/{func_name}"1381        def _log_stream_available():1382            result = logs_client.describe_log_streams(logGroupName=log_group_name)["logStreams"]1383            return len(result) > 01384        wait_until(_log_stream_available, strategy="linear")1385        ls_result = logs_client.describe_log_streams(logGroupName=log_group_name)1386        log_stream_name = ls_result["logStreams"][0]["logStreamName"]1387        def _assert_log_output():1388            log_events = logs_client.get_log_events(1389                logGroupName=log_group_name, logStreamName=log_stream_name1390            )["events"]1391            return any(["starting wait" in e["message"] for e in log_events]) and any(1392                ["done waiting" in e["message"] for e in log_events]1393            )...
Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.