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

Copied to Clipboard!

How to use _assert_contained method in localstack

Best Python code snippet using localstack_python

test_lambda.py

Source:test_lambda.py

...100            mock_run_lambda.return_value = '~notjsonresponse~'101            response = lambda_api.invoke_function(self.FUNCTION_NAME)102            self.assertEqual('~notjsonresponse~', response[0])103            self.assertEqual(200, response[1])104            self._assert_contained({'Content-Type': 'text/plain'}, response[2])105    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')106    def test_invoke_empty_plain_text_response(self, mock_run_lambda):107        with self.app.test_request_context() as context:108            self._request_response(context)109            mock_run_lambda.return_value = ''110            response = lambda_api.invoke_function(self.FUNCTION_NAME)111            self.assertEqual('', response[0])112            self.assertEqual(200, response[1])113            self._assert_contained({'Content-Type': 'text/plain'}, response[2])114    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')115    def test_invoke_empty_map_json_response(self, mock_run_lambda):116        with self.app.test_request_context() as context:117            self._request_response(context)118            mock_run_lambda.return_value = '{}'119            response = lambda_api.invoke_function(self.FUNCTION_NAME)120            self.assertEqual(b'{}\n', response[0].response[0])121            self.assertEqual(200, response[1])122            self._assert_contained({'Content-Type': 'application/json'}, response[2])123    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')124    def test_invoke_populated_map_json_response(self, mock_run_lambda):125        with self.app.test_request_context() as context:126            self._request_response(context)127            mock_run_lambda.return_value = '{"bool":true,"int":1}'128            response = lambda_api.invoke_function(self.FUNCTION_NAME)129            self.assertEqual(b'{"bool":true,"int":1}\n', response[0].response[0])130            self.assertEqual(200, response[1])131            self._assert_contained({'Content-Type': 'application/json'}, response[2])132    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')133    def test_invoke_empty_list_json_response(self, mock_run_lambda):134        with self.app.test_request_context() as context:135            self._request_response(context)136            mock_run_lambda.return_value = '[]'137            response = lambda_api.invoke_function(self.FUNCTION_NAME)138            self.assertEqual(b'[]\n', response[0].response[0])139            self.assertEqual(200, response[1])140            self._assert_contained({'Content-Type': 'application/json'}, response[2])141    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')142    def test_invoke_populated_list_json_response(self, mock_run_lambda):143        with self.app.test_request_context() as context:144            self._request_response(context)145            mock_run_lambda.return_value = '[true,1,"thing"]'146            response = lambda_api.invoke_function(self.FUNCTION_NAME)147            self.assertEqual(b'[true,1,"thing"]\n', response[0].response[0])148            self.assertEqual(200, response[1])149            self._assert_contained({'Content-Type': 'application/json'}, response[2])150    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')151    def test_invoke_string_json_response(self, mock_run_lambda):152        with self.app.test_request_context() as context:153            self._request_response(context)154            mock_run_lambda.return_value = '"thing"'155            response = lambda_api.invoke_function(self.FUNCTION_NAME)156            self.assertEqual(b'"thing"\n', response[0].response[0])157            self.assertEqual(200, response[1])158            self._assert_contained({'Content-Type': 'application/json'}, response[2])159    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')160    def test_invoke_integer_json_response(self, mock_run_lambda):161        with self.app.test_request_context() as context:162            self._request_response(context)163            mock_run_lambda.return_value = '1234'164            response = lambda_api.invoke_function(self.FUNCTION_NAME)165            self.assertEqual(b'1234\n', response[0].response[0])166            self.assertEqual(200, response[1])167            self._assert_contained({'Content-Type': 'application/json'}, response[2])168    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')169    def test_invoke_float_json_response(self, mock_run_lambda):170        with self.app.test_request_context() as context:171            self._request_response(context)172            mock_run_lambda.return_value = '1.3'173            response = lambda_api.invoke_function(self.FUNCTION_NAME)174            self.assertEqual(b'1.3\n', response[0].response[0])175            self.assertEqual(200, response[1])176            self._assert_contained({'Content-Type': 'application/json'}, response[2])177    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')178    def test_invoke_boolean_json_response(self, mock_run_lambda):179        with self.app.test_request_context() as context:180            self._request_response(context)181            mock_run_lambda.return_value = 'true'182            response = lambda_api.invoke_function(self.FUNCTION_NAME)183            self.assertEqual(b'true\n', response[0].response[0])184            self.assertEqual(200, response[1])185            self._assert_contained({'Content-Type': 'application/json'}, response[2])186    @mock.patch('localstack.services.awslambda.lambda_api.run_lambda')187    def test_invoke_null_json_response(self, mock_run_lambda):188        with self.app.test_request_context() as context:189            self._request_response(context)190            mock_run_lambda.return_value = 'null'191            response = lambda_api.invoke_function(self.FUNCTION_NAME)192            self.assertEqual(b'null\n', response[0].response[0])193            self.assertEqual(200, response[1])194            self._assert_contained({'Content-Type': 'application/json'}, response[2])195    def test_create_event_source_mapping(self):196        self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),197            data=json.dumps({'FunctionName': 'test-lambda-function', 'EventSourceArn': TEST_EVENT_SOURCE_ARN}))198        listResponse = self.client.get('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT))199        listResult = json.loads(listResponse.get_data())200        eventSourceMappings = listResult.get('EventSourceMappings')201        self.assertEqual(1, len(eventSourceMappings))202        self.assertEqual('Enabled', eventSourceMappings[0]['State'])203    def test_create_disabled_event_source_mapping(self):204        createResponse = self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),205                            data=json.dumps({'FunctionName': 'test-lambda-function',206                                             'EventSourceArn': TEST_EVENT_SOURCE_ARN,207                                             'Enabled': 'false'}))208        createResult = json.loads(createResponse.get_data())209        self.assertEqual('Disabled', createResult['State'])210        getResponse = self.client.get('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,211                        createResult.get('UUID')))212        getResult = json.loads(getResponse.get_data())213        self.assertEqual('Disabled', getResult['State'])214    def test_update_event_source_mapping(self):215        createResponse = self.client.post('{0}/event-source-mappings/'.format(lambda_api.PATH_ROOT),216                            data=json.dumps({'FunctionName': 'test-lambda-function',217                                             'EventSourceArn': TEST_EVENT_SOURCE_ARN,218                                             'Enabled': 'true'}))219        createResult = json.loads(createResponse.get_data())220        putResponse = self.client.put('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,221                        createResult.get('UUID')), data=json.dumps({'Enabled': 'false'}))222        putResult = json.loads(putResponse.get_data())223        self.assertEqual('Disabled', putResult['State'])224        getResponse = self.client.get('{0}/event-source-mappings/{1}'.format(lambda_api.PATH_ROOT,225                        createResult.get('UUID')))226        getResult = json.loads(getResponse.get_data())227        self.assertEqual('Disabled', getResult['State'])228    def test_publish_function_version(self):229        with self.app.test_request_context():230            self._create_function(self.FUNCTION_NAME)231            result = json.loads(lambda_api.publish_version(self.FUNCTION_NAME).get_data())232            result.pop('RevisionId', None)  # we need to remove this, since this is random, so we cannot know its value233            expected_result = dict()234            expected_result['CodeSize'] = self.CODE_SIZE235            expected_result['CodeSha256'] = self.CODE_SHA_256236            expected_result['FunctionArn'] = str(lambda_api.func_arn(self.FUNCTION_NAME)) + ':1'237            expected_result['FunctionName'] = str(self.FUNCTION_NAME)238            expected_result['Handler'] = str(self.HANDLER)239            expected_result['Runtime'] = str(self.RUNTIME)240            expected_result['Timeout'] = self.TIMEOUT241            expected_result['Description'] = ''242            expected_result['MemorySize'] = self.MEMORY_SIZE243            expected_result['Role'] = self.ROLE244            expected_result['KMSKeyArn'] = None245            expected_result['VpcConfig'] = None246            expected_result['LastModified'] = self.LAST_MODIFIED247            expected_result['TracingConfig'] = self.TRACING_CONFIG248            expected_result['Version'] = '1'249            expected_result['State'] = 'Active'250            expected_result['LastUpdateStatus'] = 'Successful'251            expected_result['PackageType'] = None252            self.assertDictEqual(expected_result, result)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'640    REVISION_ID = 'e54dbcf8-e3ef-44ab-9af7-8dbef510608a'641    HANDLER = 'index.handler'642    RUNTIME = 'node.js4.3'643    TIMEOUT = 60644    FUNCTION_NAME = 'test1'645    RETRY_ATTEMPTS = 5646    EVENT_AGE = 360...

test_modify.py

Source:test_modify.py

...83        for instance in db_nodes:84            self.assertEqual('db', instance['name'])85            self.assertIn('db_', instance['id'])86        self._assert_each_node_valid_hosted(db_nodes, host_nodes)87        self._assert_contained(self._nodes_relationships(db_nodes),88                               self._node_ids(host_nodes), 'host')89    def test_modified_single_node_removed_with_child_contained_in_1(self):90        yaml = self.BASE_BLUEPRINT + """91    host:92        type: cloudify.nodes.Compute93        capabilities:94            scalable:95                properties:96                    default_instances: 297    db:98        type: db99        relationships:100            -   type: cloudify.relationships.contained_in101                target: host102"""103        plan = self.parse_multi(yaml)104        modification = self.modify_multi(plan, {105            'host': {'instances': 1}106        })107        self._assert_modification(modification, 0, 2, 0, 2)108        self._assert_removed_in_previous(plan, modification)109        removed_and_related = modification['removed_and_related']110        host_nodes = self._nodes_by_name(removed_and_related, 'host')111        self.assertEqual(1, len(host_nodes))112        db_nodes = self._nodes_by_name(removed_and_related, 'db')113        self.assertEqual(1, len(db_nodes))114        self._assert_contained(self._nodes_relationships(db_nodes),115                               self._node_ids(host_nodes), 'host')116        self._assert_each_node_valid_hosted(db_nodes, host_nodes)117    def test_modified_single_node_added_with_child_contained_in_2(self):118        yaml = self.BASE_BLUEPRINT + """119    host:120        type: cloudify.nodes.Compute121    db:122        type: db123        relationships:124            -   type: cloudify.relationships.contained_in125                target: host126"""127        plan = self.parse_multi(yaml)128        modification = self.modify_multi(plan, {129            'db': {'instances': 2}130        })131        self._assert_modification(modification, 2, 0, 1, 0)132        self._assert_added_not_in_previous(plan, modification)133        added_and_related = modification['added_and_related']134        host_nodes = self._nodes_by_name(added_and_related, 'host')135        self.assertEqual(1, len(host_nodes))136        for instance in host_nodes:137            self.assertEqual('host', instance['name'])138            self.assertIn('host_', instance['id'])139            self.assertEqual(instance['host_id'], instance['id'])140        db_nodes = self._nodes_by_name(added_and_related, 'db')141        self.assertEqual(1, len(db_nodes))142        for instance in db_nodes:143            self.assertEqual('db', instance['name'])144            self.assertIn('db_', instance['id'])145        self._assert_each_node_valid_hosted(db_nodes, host_nodes)146        self._assert_contained(self._nodes_relationships(db_nodes),147                               self._node_ids(host_nodes), 'host')148    def test_modified_single_node_removed_with_child_contained_in_2(self):149        yaml = self.BASE_BLUEPRINT + """150    host:151        type: cloudify.nodes.Compute152    db:153        type: db154        capabilities:155            scalable:156                properties:157                    default_instances: 2158        relationships:159            -   type: cloudify.relationships.contained_in160                target: host161"""162        plan = self.parse_multi(yaml)163        modification = self.modify_multi(plan, {164            'db': {'instances': 1}165        })166        self._assert_modification(modification, 0, 2, 0, 1)167        self._assert_removed_in_previous(plan, modification)168        removed_and_related = modification['removed_and_related']169        host_nodes = self._nodes_by_name(removed_and_related, 'host')170        self.assertEqual(1, len(host_nodes))171        db_nodes = self._nodes_by_name(removed_and_related, 'db')172        self.assertEqual(1, len(db_nodes))173        self._assert_contained(self._nodes_relationships(db_nodes),174                               self._node_ids(host_nodes), 'host')175        self._assert_each_node_valid_hosted(db_nodes, host_nodes)176    def test_modified_two_nodes_added_with_child_contained_in(self):177        yaml = self.BASE_BLUEPRINT + """178    host:179        type: cloudify.nodes.Compute180    db:181        type: db182        relationships:183            -   type: cloudify.relationships.contained_in184                target: host185"""186        plan = self.parse_multi(yaml)187        modification = self.modify_multi(plan, {188            'host': {'instances': 2},189            'db': {'instances': 2}190        })191        self._assert_modification(modification, 5, 0, 4, 0)192        self._assert_added_not_in_previous(plan, modification)193        added_and_related = modification['added_and_related']194        host_nodes = self._nodes_by_name(added_and_related, 'host')195        self.assertEqual(2, len(host_nodes))196        for instance in host_nodes:197            self.assertEqual('host', instance['name'])198            self.assertIn('host_', instance['id'])199            self.assertEqual(instance['host_id'], instance['id'])200        db_nodes = self._nodes_by_name(added_and_related, 'db')201        self.assertEqual(3, len(db_nodes))202        for instance in db_nodes:203            self.assertEqual('db', instance['name'])204            self.assertIn('db_', instance['id'])205        self._assert_contained(self._nodes_relationships(db_nodes),206                               self._node_ids(host_nodes), 'host')207    def test_modified_two_nodes_removed_with_child_contained_in(self):208        yaml = self.BASE_BLUEPRINT + """209    host:210        type: cloudify.nodes.Compute211        capabilities:212            scalable:213                properties:214                    default_instances: 2215    db:216        type: db217        capabilities:218            scalable:219                properties:220                    default_instances: 2221        relationships:222            -   type: cloudify.relationships.contained_in223                target: host224"""225        plan = self.parse_multi(yaml)226        modification = self.modify_multi(plan, {227            'host': {'instances': 1},228            'db': {'instances': 1}229        })230        self._assert_modification(modification, 0, 5, 0, 4)231        self._assert_removed_in_previous(plan, modification)232        removed_and_related = modification['removed_and_related']233        host_nodes = self._nodes_by_name(removed_and_related, 'host')234        self.assertEqual(2, len(host_nodes))235        db_nodes = self._nodes_by_name(removed_and_related, 'db')236        self.assertEqual(3, len(db_nodes))237        self._assert_contained(self._nodes_relationships(db_nodes),238                               self._node_ids(host_nodes), 'host')239    def test_modified_single_node_added_with_connected_1(self):240        yaml = self.BASE_BLUEPRINT + """241    host1:242        type: cloudify.nodes.Compute243    host2:244        type: cloudify.nodes.Compute245    db:246        type: db247        relationships:248            -   type: cloudify.relationships.contained_in249                target: host1250            -   type: cloudify.relationships.connected_to251                target: host2252"""253        plan = self.parse_multi(yaml)254        modification = self.modify_multi(plan, {255            'host1': {'instances': 2}256        })257        self._assert_modification(modification, 3, 0, 2, 0)258        self._assert_added_not_in_previous(plan, modification)259        added_and_related = modification['added_and_related']260        host1_nodes = self._nodes_by_name(added_and_related, 'host1')261        host2_nodes = self._nodes_by_name(added_and_related, 'host2')262        self.assertEqual(1, len(host1_nodes))263        self.assertEqual(1, len(host2_nodes))264        for i, host_nodes in enumerate([host1_nodes, host2_nodes], start=1):265            for instance in host_nodes:266                self.assertEqual('host{0}'.format(i), instance['name'])267                self.assertIn('host{0}_'.format(i), instance['id'])268                self.assertEqual(instance['host_id'], instance['id'])269        db_nodes = self._nodes_by_name(added_and_related, 'db')270        self.assertEqual(1, len(db_nodes))271        for instance in db_nodes:272            self.assertEqual('db', instance['name'])273            self.assertIn('db_', instance['id'])274        self._assert_each_node_valid_hosted(db_nodes, host1_nodes)275        self._assert_contained(self._nodes_relationships(db_nodes),276                               self._node_ids(host1_nodes), 'host1')277        self._assert_all_to_all([node['relationships'] for node in db_nodes],278                                self._node_ids(host2_nodes), 'host2')279    def test_modified_single_node_removed_with_connected_1(self):280        yaml = self.BASE_BLUEPRINT + """281    host1:282        type: cloudify.nodes.Compute283        capabilities:284            scalable:285                properties:286                    default_instances: 2287    host2:288        type: cloudify.nodes.Compute289        capabilities:290            scalable:291                properties:292                    default_instances: 2293    db:294        type: db295        capabilities:296            scalable:297                properties:298                    default_instances: 2299        relationships:300            -   type: cloudify.relationships.contained_in301                target: host1302            -   type: cloudify.relationships.connected_to303                target: host2304"""305        plan = self.parse_multi(yaml)306        modification = self.modify_multi(plan, {307            'host1': {'instances': 1}308        })309        self._assert_modification(modification, 0, 5, 0, 3)310        self._assert_removed_in_previous(plan, modification)311        removed_and_related = modification['removed_and_related']312        host1_nodes = self._nodes_by_name(removed_and_related, 'host1')313        host2_nodes = self._nodes_by_name(removed_and_related, 'host2')314        self.assertEqual(1, len(host1_nodes))315        self.assertEqual(2, len(host2_nodes))316        for i, host_nodes in enumerate([host1_nodes, host2_nodes], start=1):317            for instance in host_nodes:318                self.assertEqual('host{0}'.format(i), instance['name'])319                self.assertIn('host{0}_'.format(i), instance['id'])320                self.assertEqual(instance['host_id'], instance['id'])321        db_nodes = self._nodes_by_name(removed_and_related, 'db')322        self.assertEqual(2, len(db_nodes))323        for instance in db_nodes:324            self.assertEqual('db', instance['name'])325            self.assertIn('db_', instance['id'])326        self._assert_each_node_valid_hosted(db_nodes, host1_nodes)327        self._assert_contained(self._nodes_relationships(db_nodes),328                               self._node_ids(host1_nodes), 'host1')329        self._assert_all_to_all([node['relationships'] for node in db_nodes],330                                self._node_ids(host2_nodes), 'host2')331    def test_modified_single_node_added_with_connected_2(self):332        yaml = self.BASE_BLUEPRINT + """333    host1:334        type: cloudify.nodes.Compute335    host2:336        type: cloudify.nodes.Compute337    db:338        type: db339        relationships:340            -   type: cloudify.relationships.contained_in341                target: host1342            -   type: cloudify.relationships.connected_to343                target: host2344"""345        plan = self.parse_multi(yaml)346        modification = self.modify_multi(plan, {347            'host2': {'instances': 2}348        })349        self._assert_modification(modification, 2, 0, 1, 0)350        self._assert_added_not_in_previous(plan, modification)351        added_and_related = modification['added_and_related']352        host2_nodes = self._nodes_by_name(added_and_related, 'host2')353        self.assertEqual(1, len(host2_nodes))354        for instance in host2_nodes:355            self.assertEqual('host2', instance['name'])356            self.assertIn('host2_', instance['id'])357            self.assertEqual(instance['host_id'], instance['id'])358        db_nodes = self._nodes_by_name(added_and_related, 'db')359        self.assertEqual(1, len(db_nodes))360        for instance in db_nodes:361            self.assertEqual('db', instance['name'])362            self.assertIn('db_', instance['id'])363        self._assert_all_to_all([node['relationships'] for node in db_nodes],364                                self._node_ids(host2_nodes), 'host2')365    def test_modified_single_node_removed_with_connected_2(self):366        yaml = self.BASE_BLUEPRINT + """367    host1:368        type: cloudify.nodes.Compute369        capabilities:370            scalable:371                properties:372                    default_instances: 2373    host2:374        type: cloudify.nodes.Compute375        capabilities:376            scalable:377                properties:378                    default_instances: 2379    db:380        type: db381        capabilities:382            scalable:383                properties:384                    default_instances: 2385        relationships:386            -   type: cloudify.relationships.contained_in387                target: host1388            -   type: cloudify.relationships.connected_to389                target: host2390"""391        plan = self.parse_multi(yaml)392        modification = self.modify_multi(plan, {393            'host2': {'instances': 1}394        })395        self._assert_modification(modification, 0, 5, 0, 1)396        self._assert_removed_in_previous(plan, modification)397        removed_and_related = modification['removed_and_related']398        host2_nodes = self._nodes_by_name(removed_and_related, 'host2')399        self.assertEqual(1, len(host2_nodes))400        for instance in host2_nodes:401            self.assertEqual('host2', instance['name'])402            self.assertIn('host2_', instance['id'])403            self.assertEqual(instance['host_id'], instance['id'])404        db_nodes = self._nodes_by_name(removed_and_related, 'db')405        self.assertEqual(4, len(db_nodes))406        for instance in db_nodes:407            self.assertEqual('db', instance['name'])408            self.assertIn('db_', instance['id'])409        self._assert_all_to_all([node['relationships'] for node in db_nodes],410                                self._node_ids(host2_nodes), 'host2')411    def test_modified_single_node_added_with_connected_3(self):412        yaml = self.BASE_BLUEPRINT + """413    host1:414        type: cloudify.nodes.Compute415    host2:416        type: cloudify.nodes.Compute417    db:418        type: db419        relationships:420            -   type: cloudify.relationships.contained_in421                target: host1422            -   type: cloudify.relationships.connected_to423                target: host2424"""425        plan = self.parse_multi(yaml)426        modification = self.modify_multi(plan, {427            'db': {'instances': 2}428        })429        self._assert_modification(modification, 3, 0, 1, 0)430        self._assert_added_not_in_previous(plan, modification)431        added_and_related = modification['added_and_related']432        host1_nodes = self._nodes_by_name(added_and_related, 'host1')433        host2_nodes = self._nodes_by_name(added_and_related, 'host2')434        self.assertEqual(1, len(host1_nodes))435        self.assertEqual(1, len(host2_nodes))436        for i, host_nodes in enumerate([host1_nodes, host2_nodes], start=1):437            for instance in host_nodes:438                self.assertEqual('host{0}'.format(i), instance['name'])439                self.assertIn('host{0}_'.format(i), instance['id'])440                self.assertEqual(instance['host_id'], instance['id'])441        db_nodes = self._nodes_by_name(added_and_related, 'db')442        self.assertEqual(1, len(db_nodes))443        for instance in db_nodes:444            self.assertEqual('db', instance['name'])445            self.assertIn('db_', instance['id'])446        self._assert_each_node_valid_hosted(db_nodes, host1_nodes)447        self._assert_contained(self._nodes_relationships(db_nodes),448                               self._node_ids(host1_nodes), 'host1')449        self._assert_all_to_all([node['relationships'] for node in db_nodes],450                                self._node_ids(host2_nodes), 'host2')451    def test_modified_single_node_removed_with_connected_3(self):452        yaml = self.BASE_BLUEPRINT + """453    host1:454        type: cloudify.nodes.Compute455        capabilities:456            scalable:457                properties:458                    default_instances: 2459    host2:460        type: cloudify.nodes.Compute461        capabilities:462            scalable:463                properties:464                    default_instances: 2465    db:466        type: db467        capabilities:468            scalable:469                properties:470                    default_instances: 2471        relationships:472            -   type: cloudify.relationships.contained_in473                target: host1474            -   type: cloudify.relationships.connected_to475                target: host2476"""477        plan = self.parse_multi(yaml)478        modification = self.modify_multi(plan, {479            'db': {'instances': 1}480        })481        self._assert_modification(modification, 0, 6, 0, 2)482        self._assert_removed_in_previous(plan, modification)483        removed_and_related = modification['removed_and_related']484        host1_nodes = self._nodes_by_name(removed_and_related, 'host1')485        host2_nodes = self._nodes_by_name(removed_and_related, 'host2')486        self.assertEqual(2, len(host1_nodes))487        self.assertEqual(2, len(host2_nodes))488        for i, host_nodes in enumerate([host1_nodes, host2_nodes], start=1):489            for instance in host_nodes:490                self.assertEqual('host{0}'.format(i), instance['name'])491                self.assertIn('host{0}_'.format(i), instance['id'])492                self.assertEqual(instance['host_id'], instance['id'])493        db_nodes = self._nodes_by_name(removed_and_related, 'db')494        self.assertEqual(2, len(db_nodes))495        for instance in db_nodes:496            self.assertEqual('db', instance['name'])497            self.assertIn('db_', instance['id'])498        self._assert_each_node_valid_hosted(db_nodes, host1_nodes)499        self._assert_contained(self._nodes_relationships(db_nodes),500                               self._node_ids(host1_nodes), 'host1')501        self._assert_all_to_all([node['relationships'] for node in db_nodes],502                                self._node_ids(host2_nodes), 'host2')503    def test_modified_added_with_connected_all_to_one(self):504        yaml = self.BASE_BLUEPRINT + """505    host1:506        type: cloudify.nodes.Compute507    host2:508        type: cloudify.nodes.Compute509        capabilities:510            scalable:511                properties:512                    default_instances: 5513    db:...

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.