Best Python code snippet using localstack_python
test_lambda.py
Source:test_lambda.py
...109 runtime=LAMBDA_RUNTIME_PYTHON27110 )111 @classmethod112 def tearDownClass(cls):113 testutil.delete_lambda_function(TEST_LAMBDA_NAME_PY)114 def test_invocation_type_not_set(self):115 result = self.lambda_client.invoke(116 FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}')117 result_data = json.loads(result['Payload'].read())118 self.assertEqual(result['StatusCode'], 200)119 self.assertEqual(result_data['event'], json.loads('{}'))120 def test_invocation_type_request_response(self):121 result = self.lambda_client.invoke(122 FunctionName=TEST_LAMBDA_NAME_PY,123 Payload=b'{}', InvocationType='RequestResponse')124 result_data = result['Payload'].read()125 result_data = json.loads(to_str(result_data))126 self.assertEqual(result['StatusCode'], 200)127 self.assertIsInstance(result_data, dict)128 def test_invocation_type_event(self):129 result = self.lambda_client.invoke(130 FunctionName=TEST_LAMBDA_NAME_PY,131 Payload=b'{}', InvocationType='Event')132 self.assertEqual(result['StatusCode'], 202)133 def test_invocation_type_dry_run(self):134 result = self.lambda_client.invoke(135 FunctionName=TEST_LAMBDA_NAME_PY, Payload=b'{}',136 InvocationType='DryRun')137 self.assertEqual(result['StatusCode'], 204)138 def test_lambda_environment(self):139 zip_file = testutil.create_lambda_archive(140 load_file(TEST_LAMBDA_ENV),141 get_content=True,142 libs=TEST_LAMBDA_LIBS,143 runtime=LAMBDA_RUNTIME_PYTHON27144 )145 testutil.create_lambda_function(146 func_name=TEST_LAMBDA_NAME_ENV,147 zip_file=zip_file,148 runtime=LAMBDA_RUNTIME_PYTHON27,149 envvars={'Hello': 'World'}150 )151 result = self.lambda_client.invoke(152 FunctionName=TEST_LAMBDA_NAME_ENV, Payload=b'{}')153 result_data = result['Payload']154 self.assertEqual(result['StatusCode'], 200)155 self.assertDictEqual(json.load(result_data), {'Hello': 'World'})156 # clean up157 testutil.delete_lambda_function(TEST_LAMBDA_NAME_ENV)158 def test_invocation_with_qualifier(self):159 lambda_name = 'test_lambda_%s' % short_uid()160 bucket_name = 'test_bucket_lambda2'161 bucket_key = 'test_lambda.zip'162 # upload zip file to S3163 zip_file = testutil.create_lambda_archive(164 load_file(TEST_LAMBDA_PYTHON),165 get_content=True,166 libs=TEST_LAMBDA_LIBS,167 runtime=LAMBDA_RUNTIME_PYTHON27168 )169 self.s3_client.create_bucket(Bucket=bucket_name)170 self.s3_client.upload_fileobj(171 BytesIO(zip_file), bucket_name, bucket_key)172 # create lambda function173 response = self.lambda_client.create_function(174 FunctionName=lambda_name, Handler='handler.handler',175 Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',176 Code={177 'S3Bucket': bucket_name,178 'S3Key': bucket_key179 },180 Publish=True181 )182 self.assertIn('Version', response)183 # invoke lambda function184 data_before = b'{"foo": "bar with \'quotes\\""}'185 result = self.lambda_client.invoke(186 FunctionName=lambda_name,187 Payload=data_before,188 Qualifier=response['Version']189 )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(response['Version'], context['function_version'])194 self.assertEqual(lambda_name, context['function_name'])195 # assert that logs are present196 expected = ['Lambda log message - print function']197 if use_docker():198 # Note that during regular test execution, nosetests captures the output from199 # the logging module - hence we can only expect this when running in Docker200 expected.append('.*Lambda log message - logging module')201 self.check_lambda_logs(lambda_name, expected_lines=expected)202 # clean up203 testutil.delete_lambda_function(lambda_name)204 def test_upload_lambda_from_s3(self):205 lambda_name = 'test_lambda_%s' % short_uid()206 bucket_name = 'test_bucket_lambda'207 bucket_key = 'test_lambda.zip'208 # upload zip file to S3209 zip_file = testutil.create_lambda_archive(210 load_file(TEST_LAMBDA_PYTHON),211 get_content=True,212 libs=TEST_LAMBDA_LIBS,213 runtime=LAMBDA_RUNTIME_PYTHON27214 )215 self.s3_client.create_bucket(Bucket=bucket_name)216 self.s3_client.upload_fileobj(217 BytesIO(zip_file), bucket_name, bucket_key)218 # create lambda function219 self.lambda_client.create_function(220 FunctionName=lambda_name, Handler='handler.handler',221 Runtime=lambda_api.LAMBDA_RUNTIME_PYTHON27, Role='r1',222 Code={223 'S3Bucket': bucket_name,224 'S3Key': bucket_key225 }226 )227 # invoke lambda function228 data_before = b'{"foo": "bar with \'quotes\\""}'229 result = self.lambda_client.invoke(230 FunctionName=lambda_name, Payload=data_before)231 data_after = json.loads(result['Payload'].read())232 self.assertEqual(json.loads(to_str(data_before)), data_after['event'])233 context = data_after['context']234 self.assertEqual('$LATEST', context['function_version'])235 self.assertEqual(lambda_name, context['function_name'])236 # clean up237 testutil.delete_lambda_function(lambda_name)238 def test_python_lambda_running_in_docker(self):239 if not use_docker():240 return241 zip_file = testutil.create_lambda_archive(242 load_file(TEST_LAMBDA_PYTHON3),243 get_content=True,244 libs=TEST_LAMBDA_LIBS,245 runtime=LAMBDA_RUNTIME_PYTHON36246 )247 testutil.create_lambda_function(248 func_name=TEST_LAMBDA_NAME_PY3,249 zip_file=zip_file,250 runtime=LAMBDA_RUNTIME_PYTHON36251 )252 result = self.lambda_client.invoke(253 FunctionName=TEST_LAMBDA_NAME_PY3, Payload=b'{}')254 result_data = result['Payload'].read()255 self.assertEqual(result['StatusCode'], 200)256 self.assertEqual(to_str(result_data).strip(), '{}')257 # clean up258 testutil.delete_lambda_function(TEST_LAMBDA_NAME_PY3)259class TestNodeJSRuntimes(LambdaTestBase):260 @classmethod261 def setUpClass(cls):262 cls.lambda_client = aws_stack.connect_to_service('lambda')263 def test_nodejs_lambda_running_in_docker(self):264 if not use_docker():265 return266 zip_file = testutil.create_zip_file(267 TEST_LAMBDA_NODEJS, get_content=True)268 testutil.create_lambda_function(269 func_name=TEST_LAMBDA_NAME_JS,270 zip_file=zip_file,271 handler='lambda_integration.handler',272 runtime=LAMBDA_RUNTIME_NODEJS810273 )274 result = self.lambda_client.invoke(275 FunctionName=TEST_LAMBDA_NAME_JS, Payload=b'{}')276 result_data = result['Payload'].read()277 self.assertEqual(result['StatusCode'], 200)278 self.assertEqual(to_str(result_data).strip(), '{}')279 # assert that logs are present280 expected = ['.*Node.js Lambda handler executing.']281 self.check_lambda_logs(TEST_LAMBDA_NAME_JS, expected_lines=expected)282 # clean up283 testutil.delete_lambda_function(TEST_LAMBDA_NAME_JS)284class TestCustomRuntimes(LambdaTestBase):285 @classmethod286 def setUpClass(cls):287 cls.lambda_client = aws_stack.connect_to_service('lambda')288 def test_nodejs_lambda_running_in_docker(self):289 if not use_docker():290 return291 zip_file = testutil.create_zip_file(292 TEST_LAMBDA_CUSTOM_RUNTIME, get_content=True)293 testutil.create_lambda_function(294 func_name=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,295 zip_file=zip_file,296 handler='function.handler',297 runtime=LAMBDA_RUNTIME_CUSTOM_RUNTIME298 )299 result = self.lambda_client.invoke(300 FunctionName=TEST_LAMBDA_NAME_CUSTOM_RUNTIME,301 Payload=b'{"text":"bar with \'quotes\\""}')302 result_data = result['Payload'].read()303 self.assertEqual(result['StatusCode'], 200)304 self.assertEqual(305 to_str(result_data).strip(),306 """Echoing request: '{"text": "bar with \'quotes\\""}'""")307 # assert that logs are present308 expected = ['.*Custom Runtime Lambda handler executing.']309 self.check_lambda_logs(310 TEST_LAMBDA_NAME_CUSTOM_RUNTIME, expected_lines=expected)311 # clean up312 testutil.delete_lambda_function(TEST_LAMBDA_NAME_CUSTOM_RUNTIME)313class TestDotNetCoreRuntimes(LambdaTestBase):314 @classmethod315 def setUpClass(cls):316 cls.lambda_client = aws_stack.connect_to_service('lambda')317 # lambda .NET Core 2.0 is already a zip318 zip_file = TEST_LAMBDA_DOTNETCORE2319 cls.zip_file_content = None320 with open(zip_file, 'rb') as file_obj:321 cls.zip_file_content = file_obj.read()322 def test_dotnet_lambda_running_in_docker(self):323 if not use_docker():324 return325 testutil.create_lambda_function(326 func_name=TEST_LAMBDA_NAME_DOTNETCORE2,327 zip_file=self.zip_file_content,328 handler='DotNetCore2::DotNetCore2.Lambda.Function::SimpleFunctionHandler',329 runtime=LAMBDA_RUNTIME_DOTNETCORE2330 )331 result = self.lambda_client.invoke(332 FunctionName=TEST_LAMBDA_NAME_DOTNETCORE2, Payload=b'{}')333 result_data = result['Payload'].read()334 self.assertEqual(result['StatusCode'], 200)335 self.assertEqual(to_str(result_data).strip(), '{}')336 # assert that logs are present337 expected = ['Running .NET Core 2.0 Lambda']338 self.check_lambda_logs(TEST_LAMBDA_NAME_DOTNETCORE2, expected_lines=expected)339 # clean up340 testutil.delete_lambda_function(TEST_LAMBDA_NAME_DOTNETCORE2)341class TestRubyRuntimes(LambdaTestBase):342 @classmethod343 def setUpClass(cls):344 cls.lambda_client = aws_stack.connect_to_service('lambda')345 def test_ruby_lambda_running_in_docker(self):346 if not use_docker():347 return348 zip_file = testutil.create_zip_file(349 TEST_LAMBDA_RUBY, get_content=True)350 testutil.create_lambda_function(351 func_name=TEST_LAMBDA_NAME_RUBY,352 zip_file=zip_file,353 handler='lambda_integration.handler',354 runtime=LAMBDA_RUNTIME_RUBY25355 )356 result = self.lambda_client.invoke(357 FunctionName=TEST_LAMBDA_NAME_RUBY, Payload=b'{}')358 result_data = result['Payload'].read()359 self.assertEqual(result['StatusCode'], 200)360 self.assertEqual(to_str(result_data).strip(), '{}')361 # clean up362 testutil.delete_lambda_function(TEST_LAMBDA_NAME_RUBY)363class TestJavaRuntimes(LambdaTestBase):364 @classmethod365 def setUpClass(cls):366 cls.lambda_client = aws_stack.connect_to_service('lambda')367 # deploy lambda - Java368 if not os.path.exists(TEST_LAMBDA_JAVA):369 mkdir(os.path.dirname(TEST_LAMBDA_JAVA))370 download(TEST_LAMBDA_JAR_URL, TEST_LAMBDA_JAVA)371 # Lambda supports single JAR deployments without the zip,372 # so we upload the JAR directly.373 cls.test_java_jar = load_file(TEST_LAMBDA_JAVA, mode='rb')374 testutil.create_lambda_function(375 func_name=TEST_LAMBDA_NAME_JAVA,376 zip_file=cls.test_java_jar,377 runtime=LAMBDA_RUNTIME_JAVA8,378 handler='cloud.localstack.sample.LambdaHandler'379 )380 # deploy lambda - Java with stream handler381 testutil.create_lambda_function(382 func_name=TEST_LAMBDA_NAME_JAVA_STREAM,383 zip_file=cls.test_java_jar,384 runtime=LAMBDA_RUNTIME_JAVA8,385 handler='cloud.localstack.sample.LambdaStreamHandler'386 )387 # deploy lambda - Java with serializable input object388 testutil.create_lambda_function(389 func_name=TEST_LAMBDA_NAME_JAVA_SERIALIZABLE,390 zip_file=cls.test_java_jar,391 runtime=LAMBDA_RUNTIME_JAVA8,392 handler='cloud.localstack.sample.SerializedInputLambdaHandler'393 )394 @classmethod395 def tearDownClass(cls):396 # clean up397 testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA)398 testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA_STREAM)399 testutil.delete_lambda_function(TEST_LAMBDA_NAME_JAVA_SERIALIZABLE)400 def test_java_runtime(self):401 self.assertIsNotNone(self.test_java_jar)402 result = self.lambda_client.invoke(403 FunctionName=TEST_LAMBDA_NAME_JAVA, Payload=b'{}')404 result_data = result['Payload'].read()405 self.assertEqual(result['StatusCode'], 200)406 self.assertIn('LinkedHashMap', to_str(result_data))407 def test_sns_event(self):408 result = self.lambda_client.invoke(409 FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',410 Payload=b'{"Records": [{"Sns": {"Message": "{}"}}]}')411 self.assertEqual(result['StatusCode'], 202)412 def test_ddb_event(self):413 result = self.lambda_client.invoke(414 FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',415 Payload=b'{"Records": [{"dynamodb": {"Message": "{}"}}]}')416 self.assertEqual(result['StatusCode'], 202)417 def test_kinesis_invocation(self):418 result = self.lambda_client.invoke(419 FunctionName=TEST_LAMBDA_NAME_JAVA,420 Payload=b'{"Records": [{"Kinesis": {"Data": "data", "PartitionKey": "partition"}}]}')421 result_data = result['Payload'].read()422 self.assertEqual(result['StatusCode'], 200)423 self.assertIn('KinesisEvent', to_str(result_data))424 def test_kinesis_event(self):425 result = self.lambda_client.invoke(426 FunctionName=TEST_LAMBDA_NAME_JAVA, InvocationType='Event',427 Payload=b'{"Records": [{"Kinesis": {"Data": "data", "PartitionKey": "partition"}}]}')428 result_data = result['Payload'].read()429 self.assertEqual(result['StatusCode'], 202)430 self.assertEqual(to_str(result_data).strip(), '')431 def test_stream_handler(self):432 result = self.lambda_client.invoke(433 FunctionName=TEST_LAMBDA_NAME_JAVA_STREAM, Payload=b'{}')434 result_data = result['Payload'].read()435 self.assertEqual(result['StatusCode'], 200)436 self.assertEqual(to_str(result_data).strip(), '{}')437 def test_serializable_input_object(self):438 result = self.lambda_client.invoke(439 FunctionName=TEST_LAMBDA_NAME_JAVA_SERIALIZABLE,440 Payload=b'{"bucket": "test_bucket", "key": "test_key"}')441 result_data = result['Payload'].read()442 self.assertEqual(result['StatusCode'], 200)443 self.assertDictEqual(444 json.loads(to_str(result_data)),445 {'validated': True, 'bucket': 'test_bucket', 'key': 'test_key'}446 )447class TestDockerBehaviour(LambdaTestBase):448 @classmethod449 def setUpClass(cls):450 cls.lambda_client = aws_stack.connect_to_service('lambda')451 def test_prime_and_destroy_containers(self):452 # run these tests only for the "reuse containers" Lambda executor453 if not isinstance(lambda_api.LAMBDA_EXECUTOR,454 lambda_executors.LambdaExecutorReuseContainers):455 return456 executor = lambda_api.LAMBDA_EXECUTOR457 func_name = 'test_prime_and_destroy_containers'458 func_arn = lambda_api.func_arn(func_name)459 # make sure existing containers are gone460 executor.cleanup()461 self.assertEqual(len(executor.get_all_container_names()), 0)462 # deploy and invoke lambda without Docker463 zip_file = testutil.create_lambda_archive(464 load_file(TEST_LAMBDA_ENV),465 get_content=True,466 libs=TEST_LAMBDA_LIBS,467 runtime=LAMBDA_RUNTIME_PYTHON27468 )469 testutil.create_lambda_function(470 func_name=func_name,471 zip_file=zip_file,472 runtime=LAMBDA_RUNTIME_PYTHON27,473 envvars={'Hello': 'World'}474 )475 self.assertEqual(len(executor.get_all_container_names()), 0)476 self.assertDictEqual(executor.function_invoke_times, {})477 # invoke a few times.478 durations = []479 num_iterations = 3480 for i in range(0, num_iterations + 1):481 prev_invoke_time = None482 if i > 0:483 prev_invoke_time = executor.function_invoke_times[func_arn]484 start_time = time.time()485 self.lambda_client.invoke(FunctionName=func_name, Payload=b'{}')486 duration = time.time() - start_time487 self.assertEqual(len(executor.get_all_container_names()), 1)488 # ensure the last invoke time is being updated properly.489 if i > 0:490 self.assertGreater(executor.function_invoke_times[func_arn], prev_invoke_time)491 else:492 self.assertGreater(executor.function_invoke_times[func_arn], 0)493 durations.append(duration)494 # the first call would have created the container. subsequent calls would reuse and be faster.495 for i in range(1, num_iterations + 1):496 self.assertLess(durations[i], durations[0])497 status = executor.get_docker_container_status(func_arn)498 self.assertEqual(status, 1)499 container_network = executor.get_docker_container_network(func_arn)500 self.assertEqual(container_network, 'default')501 executor.cleanup()502 status = executor.get_docker_container_status(func_arn)503 self.assertEqual(status, 0)504 self.assertEqual(len(executor.get_all_container_names()), 0)505 # clean up506 testutil.delete_lambda_function(func_name)507 def test_docker_command_for_separate_container_lambda_executor(self):508 # run these tests only for the "separate containers" Lambda executor509 if not isinstance(lambda_api.LAMBDA_EXECUTOR,510 lambda_executors.LambdaExecutorSeparateContainers):511 return512 executor = lambda_api.LAMBDA_EXECUTOR513 func_name = 'test_docker_command_for_separate_container_lambda_executor'514 func_arn = lambda_api.func_arn(func_name)515 handler = 'handler'516 lambda_cwd = '/app/lambda'517 network = 'compose_network'518 config.LAMBDA_DOCKER_NETWORK = network519 cmd = executor.prepare_execution(func_arn, {}, LAMBDA_RUNTIME_NODEJS810, '', handler, lambda_cwd)520 expected = 'docker run -v "%s":/var/task --network="%s" --rm "lambci/lambda:%s" "%s"' % (521 lambda_cwd, network, LAMBDA_RUNTIME_NODEJS810, handler)522 self.assertIn(('--network="%s"' % network), cmd, 'cmd=%s expected=%s' % (cmd, expected))523 config.LAMBDA_DOCKER_NETWORK = ''524 def test_destroy_idle_containers(self):525 # run these tests only for the "reuse containers" Lambda executor526 if not isinstance(lambda_api.LAMBDA_EXECUTOR,527 lambda_executors.LambdaExecutorReuseContainers):528 return529 executor = lambda_api.LAMBDA_EXECUTOR530 func_name = 'test_destroy_idle_containers'531 func_arn = lambda_api.func_arn(func_name)532 # make sure existing containers are gone533 executor.destroy_existing_docker_containers()534 self.assertEqual(len(executor.get_all_container_names()), 0)535 # deploy and invoke lambda without Docker536 zip_file = testutil.create_lambda_archive(537 load_file(TEST_LAMBDA_ENV),538 get_content=True,539 libs=TEST_LAMBDA_LIBS,540 runtime=LAMBDA_RUNTIME_PYTHON27541 )542 testutil.create_lambda_function(543 func_name=func_name,544 zip_file=zip_file,545 runtime=LAMBDA_RUNTIME_PYTHON27,546 envvars={'Hello': 'World'}547 )548 self.assertEqual(len(executor.get_all_container_names()), 0)549 self.lambda_client.invoke(FunctionName=func_name, Payload=b'{}')550 self.assertEqual(len(executor.get_all_container_names()), 1)551 # try to destroy idle containers.552 executor.idle_container_destroyer()553 self.assertEqual(len(executor.get_all_container_names()), 1)554 # simulate an idle container555 executor.function_invoke_times[func_arn] = time.time() - lambda_executors.MAX_CONTAINER_IDLE_TIME_MS556 executor.idle_container_destroyer()557 self.assertEqual(len(executor.get_all_container_names()), 0)558 # clean up...
aws_lambda_handler.py
Source:aws_lambda_handler.py
...6 def create_aws_lambda(self):7 _lambda_client = Utils.get_client(self, "lambda", self._lambda_endpoint_url, self._aws_region)8 _exist_result = Utils.exist_lambda_function(self, _lambda_client)9 if _exist_result:10 LambdaHandler.delete_lambda_function(self, _lambda_client)11 LambdaHandler.create_lambda_function(self, _lambda_client)12 # register stream consumer13 _rtn_consumer_map = LambdaHandler.connect_datastream_to_lambda(self, _lambda_client)14 print("[Log] consumer status {} Ok: ".format(_rtn_consumer_map.get("ResponseMetadata").get("HTTPStatusCode")))15 def create_lambda_function(self, _lambda_client):16 LambdaHandler.create_lambda_zip(self)17 with open(self._lambda_zip_path, 'rb') as f:18 zipped_code = f.read()19 _lambda_client.create_function(20 FunctionName=self._lambda_func_name,21 Runtime='python3.6',22 Role='arn:aws:iam::000000000000:role/adminRole',23 Handler=self._lambda_func_name + '.handler',24 Code=dict(ZipFile=zipped_code)25 )26 def create_lambda_zip(self):27 with ZipFile(self._lambda_zip_path, 'w') as z:28 z.write(self._copy_destination_path)29 z.close()30 def invoke_function_and_get_message(self, _lambda_client, invo_type):31 _response = _lambda_client.invoke(32 FunctionName=self._lambda_func_name,33 InvocationType=invo_type34 )35 if invo_type == 'RequestResponse':36 _rtn_data=json.loads(_response['Payload'].read().decode('utf-8'))37 else:38 _rtn_data=_response['StatusCode']39 return _rtn_data40 def delete_lambda_function(self, _lambda_client):41 _response = _lambda_client.delete_function(42 FunctionName=self._lambda_func_name43 )44 return _response45 def connect_datastream_to_lambda(self, _lambda_client):46 _res_evt_map = LambdaHandler.create_event_mapping(self, _lambda_client)47 return _res_evt_map48 def create_event_mapping(self, _lambda_client):49 _res_evt_map = _lambda_client.create_event_source_mapping(50 EventSourceArn='arn:aws:kinesis:us-east-1:000000000000:stream/weblog-kinesis-datastream',51 FunctionName=self._lambda_func_name,52 BatchSize=500,53 StartingPosition='AT_TIMESTAMP',54 StartingPositionTimestamp=1541139109,...
app.py
Source:app.py
...56 lambda_client, lambda_function_name, lambda_parms)57 print(f"The {lambda_parms['action']} of {lambda_parms['number']} resulted in "58 f"{json.load(response['Payload'])}")59 ### TO USE DELETE LAMBDA FUNCTION ###60 delete_lambda_function(lambda_client, lambda_function_name)61 print(f"Deleted function {lambda_function_name}.")62if __name__ == '__main__':...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!