Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _update_function_code method in localstack
Best Python code snippet using localstack_python
test_lambda.py
Source:test_lambda.py 
...253 def test_publish_update_version_increment(self):254 with self.app.test_request_context():255 self._create_function(self.FUNCTION_NAME)256 lambda_api.publish_version(self.FUNCTION_NAME)257 self._update_function_code(self.FUNCTION_NAME)258 result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())259 result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value260 expected_result = dict()261 expected_result['CodeSize'] = self.CODE_SIZE262 expected_result['CodeSha256'] = self.UPDATED_CODE_SHA_256263 expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':2'264 expected_result['FunctionName'] = str(self.FUNCTION_NAME)265 expected_result['Handler'] = str(self.HANDLER)266 expected_result['Runtime'] = str(self.RUNTIME)267 expected_result['Timeout'] = self.TIMEOUT268 expected_result['Description'] = ''269 expected_result['MemorySize'] = self.MEMORY_SIZE270 expected_result['Role'] = self.ROLE271 expected_result['KMSKeyArn'] = None272 expected_result['VpcConfig'] = None273 expected_result['LastModified'] = self.LAST_MODIFIED274 expected_result['TracingConfig'] = self.TRACING_CONFIG275 expected_result['Version'] = '2'276 expected_result['State'] = 'Active'277 expected_result['LastUpdateStatus'] = 'Successful'278 expected_result['PackageType'] = None279 self.assertDictEqual(expected_result, result)280 def test_publish_non_existant_function_version_returns_error(self):281 with self.app.test_request_context():282 result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())283 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])284 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),285 result['message'])286 def test_list_function_versions(self):287 with self.app.test_request_context():288 self._create_function(self.FUNCTION_NAME)289 lambda_api.publish_version(self.FUNCTION_NAME)290 lambda_api.publish_version(self.FUNCTION_NAME)291 result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())292 for version in result['Versions']:293 # we need to remove this, since this is random, so we cannot know its value294 version.pop('RevisionId', None)295 latest_version = dict()296 latest_version['CodeSize'] = self.CODE_SIZE297 latest_version['CodeSha256'] = self.CODE_SHA_256298 latest_version['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':$LATEST'299 latest_version['FunctionName'] = str(self.FUNCTION_NAME)300 latest_version['Handler'] = str(self.HANDLER)301 latest_version['Runtime'] = str(self.RUNTIME)302 latest_version['Timeout'] = self.TIMEOUT303 latest_version['Description'] = ''304 latest_version['MemorySize'] = self.MEMORY_SIZE305 latest_version['Role'] = self.ROLE306 latest_version['KMSKeyArn'] = None307 latest_version['VpcConfig'] = None308 latest_version['LastModified'] = self.LAST_MODIFIED309 latest_version['TracingConfig'] = self.TRACING_CONFIG310 latest_version['Version'] = '$LATEST'311 latest_version['State'] = 'Active'312 latest_version['LastUpdateStatus'] = 'Successful'313 latest_version['PackageType'] = None314 version1 = dict(latest_version)315 version1['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'316 version1['Version'] = '1'317 expected_result = {'Versions': sorted([latest_version, version],318 key=lambda k: str(k.get('Version')))}319 self.assertDictEqual(expected_result, result)320 def test_list_non_existant_function_versions_returns_error(self):321 with self.app.test_request_context():322 result = json.loads(lambda_api.list_versions(self.FUNCTION_NAME).get_data())323 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])324 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),325 result['message'])326 def test_create_alias(self):327 self._create_function(self.FUNCTION_NAME)328 self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))329 response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),330 data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',331 'Description': ''}))332 result = json.loads(response.get_data())333 result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value334 expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,335 'FunctionVersion': '1', 'Description': '', 'Name': self.ALIAS_NAME}336 self.assertDictEqual(expected_result, result)337 def test_create_alias_on_non_existant_function_returns_error(self):338 with self.app.test_request_context():339 result = json.loads(lambda_api.create_alias(self.FUNCTION_NAME).get_data())340 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])341 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),342 result['message'])343 def test_create_alias_returns_error_if_already_exists(self):344 self._create_function(self.FUNCTION_NAME)345 self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))346 data = json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''})347 self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME), data=data)348 response = self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),349 data=data)350 result = json.loads(response.get_data())351 alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME352 self.assertEqual(self.ALIASEXISTS_EXCEPTION, result['__type'])353 self.assertEqual(self.ALIASEXISTS_MESSAGE % alias_arn,354 result['message'])355 def test_update_alias(self):356 self._create_function(self.FUNCTION_NAME)357 self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))358 self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),359 data=json.dumps({360 'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))361 response = self.client.put('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,362 self.ALIAS_NAME),363 data=json.dumps({'FunctionVersion': '$LATEST', 'Description': 'Test-Description'}))364 result = json.loads(response.get_data())365 result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value366 expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,367 'FunctionVersion': '$LATEST', 'Description': 'Test-Description',368 'Name': self.ALIAS_NAME}369 self.assertDictEqual(expected_result, result)370 def test_update_alias_on_non_existant_function_returns_error(self):371 with self.app.test_request_context():372 result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())373 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])374 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),375 result['message'])376 def test_update_alias_on_non_existant_alias_returns_error(self):377 with self.app.test_request_context():378 self._create_function(self.FUNCTION_NAME)379 result = json.loads(lambda_api.update_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())380 alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME381 self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])382 self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])383 def test_get_alias(self):384 self._create_function(self.FUNCTION_NAME)385 self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))386 self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),387 data=json.dumps({388 'Name': self.ALIAS_NAME, 'FunctionVersion': '1', 'Description': ''}))389 response = self.client.get('{0}/functions/{1}/aliases/{2}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME,390 self.ALIAS_NAME))391 result = json.loads(response.get_data())392 result.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value393 expected_result = {'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,394 'FunctionVersion': '1', 'Description': '',395 'Name': self.ALIAS_NAME}396 self.assertDictEqual(expected_result, result)397 def test_get_alias_on_non_existant_function_returns_error(self):398 with self.app.test_request_context():399 result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())400 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])401 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),402 result['message'])403 def test_get_alias_on_non_existant_alias_returns_error(self):404 with self.app.test_request_context():405 self._create_function(self.FUNCTION_NAME)406 result = json.loads(lambda_api.get_alias(self.FUNCTION_NAME, self.ALIAS_NAME).get_data())407 alias_arn = lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME408 self.assertEqual(self.ALIASNOTFOUND_EXCEPTION, result['__type'])409 self.assertEqual(self.ALIASNOTFOUND_MESSAGE % alias_arn, result['message'])410 def test_list_aliases(self):411 self._create_function(self.FUNCTION_NAME)412 self.client.post('{0}/functions/{1}/versions'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))413 self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),414 data=json.dumps({'Name': self.ALIAS2_NAME, 'FunctionVersion': '$LATEST'}))415 self.client.post('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME),416 data=json.dumps({'Name': self.ALIAS_NAME, 'FunctionVersion': '1',417 'Description': self.ALIAS_NAME}))418 response = self.client.get('{0}/functions/{1}/aliases'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))419 result = json.loads(response.get_data())420 for alias in result['Aliases']:421 alias.pop('RevisionId', None) # we need to remove this, since this is random, so we cannot know its value422 expected_result = {'Aliases': [423 {424 'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS_NAME,425 'FunctionVersion': '1',426 'Name': self.ALIAS_NAME,427 'Description': self.ALIAS_NAME428 },429 {430 'AliasArn': lambda_api.func_arn(self.FUNCTION_NAME) + ':' + self.ALIAS2_NAME,431 'FunctionVersion': '$LATEST',432 'Name': self.ALIAS2_NAME,433 'Description': ''434 }435 ]}436 self.assertDictEqual(expected_result, result)437 def test_list_non_existant_function_aliases_returns_error(self):438 with self.app.test_request_context():439 result = json.loads(lambda_api.list_aliases(self.FUNCTION_NAME).get_data())440 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])441 self.assertEqual(self.RESOURCENOTFOUND_MESSAGE % lambda_api.func_arn(self.FUNCTION_NAME),442 result['message'])443 def test_get_container_name(self):444 executor = lambda_executors.EXECUTOR_CONTAINERS_REUSE445 name = executor.get_container_name('arn:aws:lambda:us-east-1:00000000:function:my_function_name')446 self.assertEqual(name, 'localstack_lambda_arn_aws_lambda_us-east-1_00000000_function_my_function_name')447 def test_concurrency(self):448 with self.app.test_request_context():449 self._create_function(self.FUNCTION_NAME)450 # note: PutFunctionConcurrency is mounted at: /2017-10-31451 # NOT lambda_api.PATH_ROOT452 # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html453 concurrency_data = {'ReservedConcurrentExecutions': 10}454 response = self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),455 data=json.dumps(concurrency_data))456 result = json.loads(response.get_data())457 self.assertDictEqual(concurrency_data, result)458 response = self.client.get('/2019-09-30/functions/{0}/concurrency'.format(self.FUNCTION_NAME))459 self.assertDictEqual(concurrency_data, result)460 response = self.client.delete('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME))461 self.assertIsNotNone('ReservedConcurrentExecutions', result)462 def test_concurrency_get_function(self):463 with self.app.test_request_context():464 self._create_function(self.FUNCTION_NAME)465 # note: PutFunctionConcurrency is mounted at: /2017-10-31466 # NOT lambda_api.PATH_ROOT467 # https://docs.aws.amazon.com/lambda/latest/dg/API_PutFunctionConcurrency.html468 concurrency_data = {'ReservedConcurrentExecutions': 10}469 self.client.put('/2017-10-31/functions/{0}/concurrency'.format(self.FUNCTION_NAME),470 data=json.dumps(concurrency_data))471 response = self.client.get('{0}/functions/{1}'.format(lambda_api.PATH_ROOT, self.FUNCTION_NAME))472 result = json.loads(response.get_data())473 self.assertTrue('Concurrency' in result)474 self.assertDictEqual(concurrency_data, result['Concurrency'])475 def test_list_tags(self):476 with self.app.test_request_context():477 self._create_function(self.FUNCTION_NAME, self.TAGS)478 arn = lambda_api.func_arn(self.FUNCTION_NAME)479 response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))480 result = json.loads(response.get_data())481 self.assertTrue('Tags' in result)482 self.assertDictEqual(self.TAGS, result['Tags'])483 def test_tag_resource(self):484 with self.app.test_request_context():485 self._create_function(self.FUNCTION_NAME)486 arn = lambda_api.func_arn(self.FUNCTION_NAME)487 response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))488 result = json.loads(response.get_data())489 self.assertTrue('Tags' in result)490 self.assertDictEqual({}, result['Tags'])491 self.client.post('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn), data=json.dumps({'Tags': self.TAGS}))492 response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))493 result = json.loads(response.get_data())494 self.assertTrue('Tags' in result)495 self.assertDictEqual(self.TAGS, result['Tags'])496 def test_tag_non_existent_function_returns_error(self):497 with self.app.test_request_context():498 arn = lambda_api.func_arn('non-existent-function')499 response = self.client.post(500 '{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn),501 data=json.dumps({'Tags': self.TAGS}))502 result = json.loads(response.get_data())503 self.assertEqual(self.RESOURCENOTFOUND_EXCEPTION, result['__type'])504 self.assertEqual(505 self.RESOURCENOTFOUND_MESSAGE % arn,506 result['message'])507 def test_untag_resource(self):508 with self.app.test_request_context():509 self._create_function(self.FUNCTION_NAME, tags=self.TAGS)510 arn = lambda_api.func_arn(self.FUNCTION_NAME)511 response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))512 result = json.loads(response.get_data())513 self.assertTrue('Tags' in result)514 self.assertDictEqual(self.TAGS, result['Tags'])515 self.client.delete('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn), query_string={'tagKeys': 'env'})516 response = self.client.get('{0}/tags/{1}'.format(lambda_api.PATH_ROOT, arn))517 result = json.loads(response.get_data())518 self.assertTrue('Tags' in result)519 self.assertDictEqual({'hello': 'world'}, result['Tags'])520 def test_java_options_empty_return_empty_value(self):521 lambda_executors.config.LAMBDA_JAVA_OPTS = ''522 result = lambda_executors.Util.get_java_opts()523 self.assertFalse(result)524 def test_java_options_with_only_memory_options(self):525 expected = '-Xmx512M'526 result = self.prepare_java_opts(expected)527 self.assertEqual(expected, result)528 def test_java_options_with_memory_options_and_agentlib_option(self):529 expected = '.*transport=dt_socket,server=y,suspend=y,address=[0-9]+'530 result = self.prepare_java_opts('-Xmx512M -agentlib:jdwp=transport=dt_socket,server=y'531 ',suspend=y,address=_debug_port_')532 self.assertTrue(re.match(expected, result))533 self.assertTrue(lambda_executors.Util.debug_java_port is not False)534 def test_java_options_with_unset_debug_port(self):535 options = [536 '-agentlib:jdwp=transport=dt_socket,server=y,address=_debug_port_,suspend=y',537 '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:_debug_port_,suspend=y',538 '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:_debug_port_,suspend=y',539 '-agentlib:jdwp=transport=dt_socket,server=y,address=*:_debug_port_,suspend=y',540 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=_debug_port_',541 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:_debug_port_',542 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:_debug_port_',543 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:_debug_port_'544 ]545 expected_results = [546 '-agentlib:jdwp=transport=dt_socket,server=y,address=([0-9]+),suspend=y',547 '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:([0-9]+),suspend=y',548 '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:([0-9]+),suspend=y',549 '-agentlib:jdwp=transport=dt_socket,server=y,address=\\*:([0-9]+),suspend=y',550 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=([0-9]+)',551 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:([0-9]+)',552 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:([0-9]+)',553 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=\\*:([0-9]+)'554 ]555 for i in range(len(options)):556 result = self.prepare_java_opts(options[i])557 m = re.match(expected_results[i], result)558 self.assertTrue(m)559 self.assertEqual(m.groups()[0], lambda_executors.Util.debug_java_port)560 def test_java_options_with_configured_debug_port(self):561 options = [562 '-agentlib:jdwp=transport=dt_socket,server=y,address=1234,suspend=y',563 '-agentlib:jdwp=transport=dt_socket,server=y,address=localhost:1234,suspend=y',564 '-agentlib:jdwp=transport=dt_socket,server=y,address=127.0.0.1:1234,suspend=y',565 '-agentlib:jdwp=transport=dt_socket,server=y,address=*:1234,suspend=y',566 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=1234',567 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=localhost:1234',568 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=127.0.0.1:1234',569 '-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=*:1234'570 ]571 for item in options:572 result = self.prepare_java_opts(item)573 self.assertEqual('1234', lambda_executors.Util.debug_java_port)574 self.assertEqual(item, result)575 def prepare_java_opts(self, java_opts):576 lambda_executors.config.LAMBDA_JAVA_OPTS = java_opts577 result = lambda_executors.Util.get_java_opts()578 return result579 def test_get_java_lib_folder_classpath(self):580 jar_file = os.path.join(new_tmp_dir(), 'foo.jar')581 save_file(jar_file, '')582 classpath = lambda_executors.Util.get_java_classpath(jar_file)583 self.assertIn('.:foo.jar', classpath)584 self.assertIn('*.jar', classpath)585 def test_get_java_lib_folder_classpath_no_directories(self):586 base_dir = new_tmp_dir()587 jar_file = os.path.join(base_dir, 'foo.jar')588 save_file(jar_file, '')589 lib_file = os.path.join(base_dir, 'lib', 'lib.jar')590 mkdir(os.path.dirname(lib_file))591 save_file(lib_file, '')592 classpath = lambda_executors.Util.get_java_classpath(jar_file)593 self.assertIn(':foo.jar', classpath)594 self.assertIn('lib/lib.jar:', classpath)595 self.assertIn(':*.jar', classpath)596 def test_get_java_lib_folder_classpath_archive_is_None(self):597 self.assertRaises(TypeError, lambda_executors.Util.get_java_classpath, None)598 @mock.patch('localstack.services.awslambda.lambda_executors.store_cloudwatch_logs')599 def test_executor_store_logs_can_handle_milliseconds(self, mock_store_cloudwatch_logs):600 mock_details = mock.Mock()601 t_sec = time.time() # plain old epoch secs602 t_ms = time.time() * 1000 # epoch ms as a long-int like AWS603 # pass t_ms millisecs to _store_logs604 lambda_executors._store_logs(mock_details, 'mock log output', t_ms)605 # expect the computed log-stream-name to having a prefix matching the date derived from t_sec606 today = datetime.datetime.utcfromtimestamp(t_sec).strftime('%Y/%m/%d')607 log_stream_name = mock_store_cloudwatch_logs.call_args_list[0].args[1]608 parts = log_stream_name.split('/')609 date_part = '/'.join(parts[:3])610 self.assertEqual(date_part, today)611 def _create_function(self, function_name, tags={}):612 arn = lambda_api.func_arn(function_name)613 lambda_api.ARN_TO_LAMBDA[arn] = LambdaFunction(arn)614 lambda_api.ARN_TO_LAMBDA[arn].versions = {615 '$LATEST': {'CodeSize': self.CODE_SIZE, 'CodeSha256': self.CODE_SHA_256, 'RevisionId': self.REVISION_ID}616 }617 lambda_api.ARN_TO_LAMBDA[arn].handler = self.HANDLER618 lambda_api.ARN_TO_LAMBDA[arn].runtime = self.RUNTIME619 lambda_api.ARN_TO_LAMBDA[arn].timeout = self.TIMEOUT620 lambda_api.ARN_TO_LAMBDA[arn].tags = tags621 lambda_api.ARN_TO_LAMBDA[arn].envvars = {}622 lambda_api.ARN_TO_LAMBDA[arn].last_modified = self.LAST_MODIFIED623 lambda_api.ARN_TO_LAMBDA[arn].role = self.ROLE624 lambda_api.ARN_TO_LAMBDA[arn].memory_size = self.MEMORY_SIZE625 def _update_function_code(self, function_name, tags={}):626 arn = lambda_api.func_arn(function_name)627 lambda_api.ARN_TO_LAMBDA[arn].versions.update({628 '$LATEST': {'CodeSize': self.CODE_SIZE,629 'CodeSha256': self.UPDATED_CODE_SHA_256,630 'RevisionId': self.REVISION_ID}631 })632 def _assert_contained(self, child, parent):633 self.assertTrue(set(child.items()).issubset(set(parent.items())))634class TestLambdaEventInvokeConfig(unittest.TestCase):635 CODE_SIZE = 50636 CODE_SHA_256 = '/u60ZpAA9bzZPVwb8d4390i5oqP1YAObUwV03CZvsWA='637 MEMORY_SIZE = 128638 ROLE = LAMBDA_TEST_ROLE639 LAST_MODIFIED = '2019-05-25T17:00:48.260+0000'...
awsclient.py
Source:awsclient.py
...220 This method only updates the values provided to it. If a parameter221 is not provided, no changes will be made for that that parameter on222 the targeted lambda function.223 """224 return_value = self._update_function_code(function_name=function_name,225 zip_contents=zip_contents)226 self._update_function_config(227 environment_variables=environment_variables,228 runtime=runtime,229 timeout=timeout,230 memory_size=memory_size,231 role_arn=role_arn,232 subnet_ids=subnet_ids,233 security_group_ids=security_group_ids,234 function_name=function_name235 )236 if tags is not None:237 self._update_function_tags(return_value['FunctionArn'], tags)238 return return_value239 def _update_function_code(self, function_name, zip_contents):240 # type: (str, str) -> Dict[str, Any]241 lambda_client = self._client('lambda')242 try:243 return lambda_client.update_function_code(244 FunctionName=function_name, ZipFile=zip_contents)245 except _REMOTE_CALL_ERRORS as e:246 context = LambdaErrorContext(247 function_name,248 'update_function_code',249 len(zip_contents)250 )251 raise self._get_lambda_code_deployment_error(e, context)252 def put_function_concurrency(self, function_name,253 reserved_concurrent_executions):...
awslambda.py
Source:awslambda.py
...151 LOG.debug('Update alias error: %s', error)152 LOG.info("Alias update failed. Retrying...")153 raise154 @retries(max_attempts=5, wait=lambda n: exponential_backoff(n, 4), exceptions=ClientError)155 def _update_function_code(self):156 code_args = self._get_default_lambda_code()157 self.lambda_client.update_function_code(158 FunctionName=self.app_name,159 **code_args160 )161 LOG.info("Successfully updated Lambda code.")162 @retries(max_attempts=5, wait=lambda n: exponential_backoff(n, 4), exceptions=ClientError)163 def _update_function_configuration(self, vpc_config):164 """Update existing Lambda function configuration.165 Args:166 vpc_config (dict): Dictionary of SubnetIds and SecurityGroupsIds for using167 a VPC in lambda168 """169 LOG.info('Updating configuration for lambda function: %s', self.app_name)170 default_tags = {'app_group': self.group, 'app_name': self.app_name}171 lambda_tags = {**default_tags, **self.custom_tags}172 try:173 lambda_args = self._get_lambda_args("update", vpc_config, lambda_tags)174 self.lambda_client.update_function_configuration(**lambda_args)175 if self._check_lambda_alias():176 self._update_alias()177 else:178 self._create_alias()179 self._put_concurrent_limits(delete_old_config=True)180 self._put_destinations()181 self._put_provisioned_throughput(delete_old_config=True)182 except boto3.exceptions.botocore.exceptions.ClientError as error:183 if 'CreateNetworkInterface' in error.response['Error']['Message']:184 message = '{0} is missing "ec2:CreateNetworkInterface"'.format(self.role_arn)185 LOG.debug(message)186 raise SystemExit(message)187 LOG.warning("Client error {} during {}. Retrying with backoff."188 .format(error.response['Error']['Code'], error.operation_name))189 raise190 LOG.info('Updating Lambda function tags')191 lambda_arn = get_lambda_arn(self.app_name, self.env, self.region)192 self.lambda_client.tag_resource(Resource=lambda_arn, Tags=lambda_tags)193 LOG.info("Successfully updated Lambda configuration.")194 @retries(max_attempts=3, wait=exponential_backoff, exceptions=ClientError)195 def _put_provisioned_throughput(self, delete_old_config):196 """Update existing Lambda function provisioned throughput config197 Args:198 delete_old_config (bool): delete existing config if no provisioned199 throughput set in pipeline files200 """201 try:202 if self.lambda_provisioned_throughput:203 self.lambda_client.put_provisioned_concurrency_config(204 FuctionName=self.app_name,205 Qualifier=self.env,206 ProvisionedConcurrentExecutions=self.lambda_provisioned_throughput)207 elif delete_old_config:208 self.lambda_client.delete_provisioned_concurrency_config(209 FunctionName=self.app_name,210 Qualifier=self.env)211 except ClientError as error:212 LOG.warning("Client error {} during {}. Retrying with backoff."213 .format(error.response['Error']['Code'], error.operation_name))214 raise215 @retries(max_attempts=3, wait=exponential_backoff, exceptions=ClientError)216 def _put_destinations(self):217 try:218 if self.lambda_destinations:219 self.lambda_client.put_function_event_invoke_config(220 FunctionName=self.app_name,221 DestinationConfig=self.lambda_destinations222 )223 except ClientError as error:224 LOG.warning("Client error {} during {}. Retrying with backoff."225 .format(error.response['Error']['Code'], error.operation_name))226 raise227 @retries(max_attempts=3, wait=exponential_backoff, exceptions=ClientError)228 def _put_concurrent_limits(self, delete_old_config):229 """Update existing Lambda function concurrency230 Args:231 delete_old_config (bool): delete existing config if no concurrency232 limit set in pipeline files233 """234 try:235 if self.concurrency_limit:236 self.lambda_client.put_function_concurrency(237 FunctionName=self.app_name,238 ReservedConcurrentExecutions=self.concurrency_limit239 )240 elif delete_old_config:241 self.lambda_client.delete_function_concurrency(FunctionName=self.app_name)242 except ClientError as error:243 LOG.warning("Client error {} during {}. Retrying with backoff."244 .format(error.response['Error']['Code'], error.operation_name))245 raise246 @retries(max_attempts=3, wait=exponential_backoff, exceptions=SystemExit)247 def _create_function(self, vpc_config):248 """Create lambda function, configures lambda parameters.249 We need to upload non-zero zip when creating function. Uploading250 hello_world python lambda function since AWS doesn't care which251 executable is in ZIP.252 Args:253 vpc_config (dict): Dictionary of SubnetIds and SecurityGroupsIds for using254 a VPC in lambda255 """256 LOG.info('Creating lambda function: %s', self.app_name)257 default_tags = {'app_group': self.group, 'app_name': self.app_name}258 lambda_tags = {**default_tags, **self.custom_tags}259 try:260 lambda_args = self._get_lambda_args("create", vpc_config, lambda_tags)261 self.lambda_client.create_function(**lambda_args)262 if self._check_lambda_alias():263 self._update_alias()264 else:265 self._create_alias()266 267 self._put_concurrent_limits(delete_old_config=False)268 self._put_destinations()269 self._put_provisioned_throughput(delete_old_config=False)270 except ClientError as error:271 if 'CreateNetworkInterface' in error.response['Error']['Message']:272 message = '{0} is missing "ec2:CreateNetworkInterface"'.format(self.role_arn)273 LOG.critical(message)274 raise SystemExit(message)275 raise276 LOG.info("Successfully created Lambda function and alias")277 def deploy_lambda_function(self):278 """Create or update Lambda function."""279 vpc_config = self._vpc_config()280 if self._check_lambda():281 self._update_function_configuration(vpc_config)282 if LAMBDA_STANDALONE_MODE:283 self._update_function_code()284 else:285 self._create_function(vpc_config)286 def _get_lambda_args(self, action, vpc_config, lambda_tags):287 """Gets lambda args as a dictionary, depending on properties such as package_type.288 Args:289 action (str): Action taken place, either create or update. This adjust the290 outputted dictionary and it's keys291 """292 # Default args that apply to all package types293 lambda_args = {294 "Environment": self.lambda_environment,295 "FunctionName": self.app_name,296 "Role": self.role_arn,297 "Description": self.description,...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
