Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
