Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use check_lambda_logs method in localstack

Best Python code snippet using localstack_python
test_lambda.py
Source:test_lambda.py
...40TEST_LAMBDA_JAR_URL = ('{url}/cloud/localstack/{name}/{version}/{name}-{version}-tests.jar').format(41    version=LOCALSTACK_MAVEN_VERSION, url=MAVEN_BASE_URL, name='localstack-utils')42TEST_LAMBDA_LIBS = ['localstack', 'localstack_client', 'requests', 'psutil', 'urllib3', 'chardet', 'certifi', 'idna']43class LambdaTestBase(unittest.TestCase):44    def check_lambda_logs(self, func_name, expected_lines=[]):45        logs_client = aws_stack.connect_to_service('logs')46        log_group_name = '/aws/lambda/%s' % func_name47        streams = logs_client.describe_log_streams(logGroupName=log_group_name)['logStreams']48        streams = sorted(streams, key=lambda x: x['creationTime'], reverse=True)49        log_events = logs_client.get_log_events(50            logGroupName=log_group_name, logStreamName=streams[0]['logStreamName'])['events']51        log_messages = [e['message'] for e in log_events]52        for line in expected_lines:53            if '.*' in line:54                found = [re.match(line, m) for m in log_messages]55                if any(found):56                    continue57            self.assertIn(line, log_messages)58class TestPythonRuntimes(LambdaTestBase):59    @classmethod60    def setUpClass(cls):61        cls.lambda_client = aws_stack.connect_to_service('lambda')62        cls.s3_client = aws_stack.connect_to_service('s3')63        zip_file = testutil.create_lambda_archive(64            load_file(TEST_LAMBDA_PYTHON),65            get_content=True,66            libs=TEST_LAMBDA_LIBS,67            runtime=LAMBDA_RUNTIME_PYTHON2768        )69        testutil.create_lambda_function(70            func_name=TEST_LAMBDA_NAME_PY,71            zip_file=zip_file,72            runtime=LAMBDA_RUNTIME_PYTHON2773        )74    @classmethod75    def tearDownClass(cls):76        cls.lambda_client.delete_function(FunctionName=TEST_LAMBDA_NAME_PY)77    def test_invocation_type_not_set(self):78        result = self.lambda_client.invoke(79            FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}')80        result_data = json.loads(result['Payload'].read())81        self.assertEqual(result['StatusCode'], 200)82        self.assertEqual(result_data['event'], json.loads('{}'))83    def test_invocation_type_request_response(self):84        result = self.lambda_client.invoke(85            FunctionName=TEST_LAMBDA_NAME_PY,86            Payload=b'{}', InvocationType='RequestResponse')87        result_data = result['Payload'].read()88        result_data = json.loads(to_str(result_data))89        self.assertEqual(result['StatusCode'], 200)90        self.assertIsInstance(result_data, dict)91    def test_invocation_type_event(self):92        result = self.lambda_client.invoke(93            FunctionName=TEST_LAMBDA_NAME_PY,94            Payload=b'{}', InvocationType='Event')95        self.assertEqual(result['StatusCode'], 202)96    def test_invocation_type_dry_run(self):97        result = self.lambda_client.invoke(98            FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}',99            InvocationType='DryRun')100        self.assertEqual(result['StatusCode'], 204)101    def test_lambda_environment(self):102        zip_file = testutil.create_lambda_archive(103            load_file(TEST_LAMBDA_ENV),104            get_content=True,105            libs=TEST_LAMBDA_LIBS,106            runtime=LAMBDA_RUNTIME_PYTHON27107        )108        testutil.create_lambda_function(109            func_name=TEST_LAMBDA_NAME_ENV,110            zip_file=zip_file,111            runtime=LAMBDA_RUNTIME_PYTHON27,112            envvars={'Hello': 'World'}113        )114        result = self.lambda_client.invoke(115            FunctionName=TEST_LAMBDA_NAME_ENV, Payload=b'{}')116        result_data = result['Payload']117        self.assertEqual(result['StatusCode'], 200)118        self.assertDictEqual(json.load(result_data), {'Hello': 'World'})119    def test_invocation_with_qualifier(self):120        lambda_name = 'test_lambda_%s' % short_uid()121        bucket_name = 'test_bucket_lambda2'122        bucket_key = 'test_lambda.zip'123        # upload zip file to S3124        zip_file = testutil.create_lambda_archive(125            load_file(TEST_LAMBDA_PYTHON),126            get_content=True,127            libs=TEST_LAMBDA_LIBS,128            runtime=LAMBDA_RUNTIME_PYTHON27129        )130        self.s3_client.create_bucket(Bucket=bucket_name)131        self.s3_client.upload_fileobj(132            BytesIO(zip_file), bucket_name, bucket_key)133        # create lambda function134        response = self.lambda_client.create_function(135            FunctionName=lambda_name, Handler='handler.handler',136            Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',137            Code={138                'S3Bucket': bucket_name,139                'S3Key': bucket_key140            },141            Publish=True142        )143        self.assertIn('Version', response)144        # invoke lambda function145        data_before = b'{"foo": "bar"}'146        result = self.lambda_client.invoke(147            FunctionName=lambda_name,148            Payload=data_before,149            Qualifier=response['Version']150        )151        data_after = json.loads(result['Payload'].read())152        self.assertEqual(json.loads(to_str(data_before)), data_after['event'])153        context = data_after['context']154        self.assertEqual(response['Version'], context['function_version'])155        self.assertEqual(lambda_name, context['function_name'])156        # assert that logs are present157        expected = ['Lambda log message - print function']158        if use_docker():159            # Note that during regular test execution, nosetests captures the output from160            # the logging module - hence we can only expect this when running in Docker161            expected.append('.*Lambda log message - logging module')162        self.check_lambda_logs(lambda_name, expected_lines=expected)163    def test_upload_lambda_from_s3(self):164        lambda_name = 'test_lambda_%s' % short_uid()165        bucket_name = 'test_bucket_lambda'166        bucket_key = 'test_lambda.zip'167        # upload zip file to S3168        zip_file = testutil.create_lambda_archive(169            load_file(TEST_LAMBDA_PYTHON),170            get_content=True,171            libs=TEST_LAMBDA_LIBS,172            runtime=LAMBDA_RUNTIME_PYTHON27173        )174        self.s3_client.create_bucket(Bucket=bucket_name)175        self.s3_client.upload_fileobj(176            BytesIO(zip_file), bucket_name, bucket_key)177        # create lambda function178        self.lambda_client.create_function(179            FunctionName=lambda_name, Handler='handler.handler',180            Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',181            Code={182                'S3Bucket': bucket_name,183                'S3Key': bucket_key184            }185        )186        # invoke lambda function187        data_before = b'{"foo": "bar"}'188        result = self.lambda_client.invoke(189            FunctionName=lambda_name, Payload=data_before)190        data_after = json.loads(result['Payload'].read())191        self.assertEqual(json.loads(to_str(data_before)), data_after['event'])192        context = data_after['context']193        self.assertEqual('$LATEST', context['function_version'])194        self.assertEqual(lambda_name, context['function_name'])195    def test_python_lambda_running_in_docker(self):196        if not use_docker():197            return198        zip_file = testutil.create_lambda_archive(199            load_file(TEST_LAMBDA_PYTHON3),200            get_content=True,201            libs=TEST_LAMBDA_LIBS,202            runtime=LAMBDA_RUNTIME_PYTHON36203        )204        testutil.create_lambda_function(205            func_name=TEST_LAMBDA_NAME_PY3,206            zip_file=zip_file,207            runtime=LAMBDA_RUNTIME_PYTHON36208        )209        result = self.lambda_client.invoke(210            FunctionName=TEST_LAMBDA_NAME_PY3, Payload=b'{}')211        result_data = result['Payload'].read()212        self.assertEqual(result['StatusCode'], 200)213        self.assertEqual(to_str(result_data).strip(), '{}')214class TestNodeJSRuntimes(LambdaTestBase):215    @classmethod216    def setUpClass(cls):217        cls.lambda_client = aws_stack.connect_to_service('lambda')218    def test_nodejs_lambda_running_in_docker(self):219        if not use_docker():220            return221        zip_file = testutil.create_zip_file(222            TEST_LAMBDA_NODEJS, get_content=True)223        testutil.create_lambda_function(224            func_name=TEST_LAMBDA_NAME_JS,225            zip_file=zip_file,226            handler='lambda_integration.handler',227            runtime=LAMBDA_RUNTIME_NODEJS228        )229        result = self.lambda_client.invoke(230            FunctionName=TEST_LAMBDA_NAME_JS, Payload=b'{}')231        result_data = result['Payload'].read()232        self.assertEqual(result['StatusCode'], 200)233        self.assertEqual(to_str(result_data).strip(), '{}')234        # assert that logs are present235        expected = ['.*Node.js Lambda handler executing.']236        self.check_lambda_logs(TEST_LAMBDA_NAME_JS, expected_lines=expected)237class TestCustomRuntimes(LambdaTestBase):238    @classmethod239    def setUpClass(cls):240        cls.lambda_client = aws_stack.connect_to_service('lambda')241    def test_nodejs_lambda_running_in_docker(self):242        if not use_docker():243            return244        zip_file = testutil.create_zip_file(245            TEST_LAMBDA_CUSTOM_RUNTIME, get_content=True)246        testutil.create_lambda_function(247            func_name=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,248            zip_file=zip_file,249            handler='function.handler',250            runtime=LAMBDA_RUNTIME_CUSTOM_RUNTIME251        )252        result = self.lambda_client.invoke(253            FunctionName=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,254            Payload=b'{"text":"Hello"}')255        result_data = result['Payload'].read()256        self.assertEqual(result['StatusCode'], 200)257        self.assertEqual(258            to_str(result_data).strip(),259            """Echoing request: '{"text": "Hello"}'""")260        # assert that logs are present261        expected = ['.*Custom Runtime Lambda handler executing.']262        self.check_lambda_logs(263            TEST_LAMBDA_NAME_CUSTOM_RUNTIME, expected_lines=expected)264class TestDotNetCoreRuntimes(LambdaTestBase):265    @classmethod266    def setUpClass(cls):267        cls.lambda_client = aws_stack.connect_to_service('lambda')268        # lambda .NET Core 2.0 is already a zip269        zip_file = TEST_LAMBDA_DOTNETCORE2270        cls.zip_file_content = None271        with open(zip_file, 'rb') as file_obj:272            cls.zip_file_content = file_obj.read()273    def test_dotnet_lambda_running_in_docker(self):274        if not use_docker():275            return276        testutil.create_lambda_function(277            func_name=TEST_LAMBDA_NAME_DOTNETCORE2,278            zip_file=self.zip_file_content,279            handler='DotNetCore2::DotNetCore2.Lambda.Function::SimpleFunctionHandler',280            runtime=LAMBDA_RUNTIME_DOTNETCORE2281        )282        result = self.lambda_client.invoke(283            FunctionName=TEST_LAMBDA_NAME_DOTNETCORE2, Payload=b'{}')284        result_data = result['Payload'].read()285        self.assertEqual(result['StatusCode'], 200)286        self.assertEqual(to_str(result_data).strip(), '{}')287        # assert that logs are present288        expected = ['Running .NET Core 2.0 Lambda']289        self.check_lambda_logs(TEST_LAMBDA_NAME_DOTNETCORE2, expected_lines=expected)290class TestRubyRuntimes(LambdaTestBase):291    @classmethod292    def setUpClass(cls):293        cls.lambda_client = aws_stack.connect_to_service('lambda')294    def test_ruby_lambda_running_in_docker(self):295        if not use_docker():296            return297        zip_file = testutil.create_zip_file(298            TEST_LAMBDA_RUBY, get_content=True)299        testutil.create_lambda_function(300            func_name=TEST_LAMBDA_NAME_RUBY,301            zip_file=zip_file,302            handler='lambda_integration.handler',303            runtime=LAMBDA_RUNTIME_RUBY25...
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.