Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.