KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use cloudformation method in localstack
Best Python code snippet using localstack_python
cloudformation_starter.py
Source:cloudformation_starter.py 
...21from localstack.services.awslambda.lambda_api import BUCKET_MARKER_LOCAL22LOG = logging.getLogger(__name__)23# Maps (stack_name,resource_logical_id) -> Bool to indicate which resources are currently being updated24CURRENTLY_UPDATING_RESOURCES = {}25def start_cloudformation(port=None, asynchronous=False, update_listener=None):26 port = port or config.PORT_CLOUDFORMATION27 backend_port = DEFAULT_PORT_CLOUDFORMATION_BACKEND28 cmd = 'python "%s" cloudformation -p %s -H 0.0.0.0' % (__file__, backend_port)29 print('Starting mock CloudFormation (%s port %s)...' % (get_service_protocol(), port))30 start_proxy_for_service('dynamodb', port, backend_port, update_listener)31 env_vars = {'PYTHONPATH': ':'.join(sys.path)}32 return do_run(cmd, asynchronous, env_vars=env_vars)33def apply_patches():34 """ Apply patches to make LocalStack seamlessly interact with the moto backend.35 TODO: Eventually, these patches should be contributed to the upstream repo! """36 # Patch S3Backend.get_key method in moto to use S3 API from LocalStack37 def get_key(self, bucket_name, key_name, version_id=None):38 s3_client = aws_stack.connect_to_service('s3')39 value = b''...
build-cloudformation-template.py
Source:build-cloudformation-template.py
1#!/usr/bin/env python2from __future__ import print_function3import os, sys, json4try:5 import ruamel.yaml6except Exception as e:7 raise Exception("Unable to import ruamel.yaml.\nTry \"pip install ruamel.yaml\".")8 sys.exit(1)9# These limits are set by CloudFormation.10function_max_characters = 409611cloudformation_template_max_size = 460800 # Max for template loaded from S312input_word_list_separator = "\n"13output_word_list_separator = "\n"14function_name_source_map = {15 "ConfigurationSetupFunction": "initial_configuration.py",16 "QueueRefreshFunction": "refresh_queue_messages.py",17 "WordGeneratorFunction": "generate_word.py",18 "WordListTableTriggerFunction": "word_list_table_trigger.py"19}20web_resource_name_source_map = {21 "WordGeneratorApiRootGet": "index.html"22}23word_lists_directory_name = "word-lists"24function_source_directory_name = "functions"25web_source_directory_name = "web"26cloudformation_source_template_file_name = "unique-word-api-source.yaml"27repository_directory_path = os.path.dirname(os.path.realpath(__file__))28'''29 LOAD CLOUDFORMATION SOURCE TEMPLATE30'''31cloudformation_source_template_file_path = os.path.join(32 repository_directory_path,33 cloudformation_source_template_file_name34)35cloudformation_source_template_string = None36try:37 cloudformation_source_template_string = open(cloudformation_source_template_file_path).read()38except:39 raise Exception("Unable to read CloudFormation source template at {}.".format(cloudformation_source_template_file_path))40cloudformation_template_object = None41try:42 cloudformation_template_object = ruamel.yaml.round_trip_load(cloudformation_source_template_string)43except:44 raise Exception("Error decoding JSON from CloudFormation source template at {}.".format(cloudformation_source_template_file_path))45'''46 LOAD WEB SOURCE CODE47'''48web_resource_names = web_resource_name_source_map.keys()49for each_web_resource_name in web_resource_names:50 each_web_resource_file_name = web_resource_name_source_map[each_web_resource_name]51 52 web_resource_file_path = os.path.join(53 repository_directory_path,54 web_source_directory_name,55 each_web_resource_file_name56 )57 58 web_resource_source_string = None59 try:60 web_resource_source_string = open(web_resource_file_path).read()61 except:62 raise Exception("Unable to read web content source for {} at {}.".format(each_web_resource_name, web_resource_file_path))63 64 try:65 cloudformation_template_object["Resources"][each_web_resource_name]["Properties"]["Integration"]["IntegrationResponses"][0]["ResponseTemplates"]["text/html"] = web_resource_source_string66 except:67 raise Exception("Unable to add web content for {} to CloudFormation template.".format(each_web_resource_name))68 69 print("Web content for {} loaded into CloudFormation template.".format(each_web_resource_name), file=sys.stderr)70'''71 LOAD LAMBDA FUNCTION SOURCE CODE72'''73function_names = function_name_source_map.keys()74for each_function_name in function_names:75 each_function_source_file_name = function_name_source_map[each_function_name]76 77 function_source_file_path = os.path.join(78 repository_directory_path,79 function_source_directory_name,80 each_function_source_file_name81 )82 83 function_source_string = None84 try:85 function_source_string = open(function_source_file_path).read()86 except:87 raise Exception("Unable to read source code for {} function at {}.".format(each_function_name, function_source_file_path))88 89 # Reduce character count by converting four space indents to tabs.90 function_source_string = function_source_string.replace(" ", "\t")91 92 if len(function_source_string) > function_max_characters:93 raise Exception("Source code for function {} is too long. It is {} characters long ({} max).".format(94 each_function_name,95 len(function_source_string),96 function_max_characters97 ))98 99 try:100 cloudformation_template_object["Resources"][each_function_name]["Properties"]["Code"] = {101 "ZipFile": {102 "Fn::Sub": function_source_string103 }104 }105 except:106 raise Exception("Unable to add source code for {} to CloudFormation template.".format(each_function_name))107 108 print("Code for {} loaded into CloudFormation template.".format(each_function_name), file=sys.stderr)109'''110 LOAD WORD LIST111'''112word_lists_directory_path = os.path.join(113 repository_directory_path,114 word_lists_directory_name115)116word_list_file_names = None117for (dir_path, dir_names, file_names) in os.walk(word_lists_directory_path):118 if dir_path != word_lists_directory_path:119 continue120 121 word_list_file_names = file_names122if not word_list_file_names:123 raise Exception("Unable to find word lists in \"{}\" directory.".format(word_lists_directory_name))124user_selected_word_list_file_name = None125if len(word_list_file_names) == 1:126 print("Using word list: {}.".format(word_list_file_names[0]), file=sys.stderr)127 user_selected_word_list_file_name[word_list_file_names[0]]128if user_selected_word_list_file_name is None:129 print("Select a word list:", file=sys.stderr)130 for i, each_file_name in enumerate(word_list_file_names):131 print(" [{}] {}".format(i+1, each_file_name), file=sys.stderr)132 while user_selected_word_list_file_name is None:133 print("> ", end="", file=sys.stderr)134 user_input_text = raw_input("")135 136 try:137 user_selected_word_list_file_name = word_list_file_names[int(user_input_text)-1]138 except:139 print("Invalid selection.", file=sys.stderr)140 print("You selected {}.".format(user_selected_word_list_file_name), file=sys.stderr)141# Read in input word list.142input_word_list = open(os.path.join(word_lists_directory_path, user_selected_word_list_file_name)).read().strip().split(input_word_list_separator)143# Verify no words in the list are duplicates.144words_found_map = {}145for each_word in input_word_list:146 if each_word is None or each_word == "":147 continue148 elif each_word in words_found_map:149 raise Exception("Word \"{}\" appears more than once in input word list.".format(each_word))150 words_found_map[each_word] = True151clean_input_word_list = words_found_map.keys()152clean_input_word_list.sort()153# We don't need this map any more.154del words_found_map155print("Word list contains {} words.".format(len(clean_input_word_list)), file=sys.stderr)156output_word_list_string = output_word_list_separator.join(clean_input_word_list)157try:158 cloudformation_template_object["Mappings"]["StaticVariables"]["Main"]["WordList"] = output_word_list_string159except Exception as e:160 raise Exception("Unable to set new word list in CloudFormation template.")161del input_word_list162del clean_input_word_list163del output_word_list_string164print("Word list added to CloudFormation template successfully.", file=sys.stderr)165'''166 OUTPUT TEMPLATE167'''168# Try exporting in a friendly, readable format.169output_template_string = ruamel.yaml.round_trip_dump(cloudformation_template_object)170print(output_template_string)171if sys.stdout.isatty():172 print("", file=sys.stderr)173 print("To save the template, try piping the output of this script to a file.", file=sys.stderr)174 175 python_name = "python"176 if sys.version_info[0] > 2:177 python_name += sys.version_info[0]178 179 print("For example, \n {} {} > new-cloudformation-template.json".format(180 python_name,181 " ".join(sys.argv)...
test_cloudformation.py
Source:test_cloudformation.py
1import os2import unittest3from localstack.utils.aws import aws_stack4from localstack.utils.common import load_file, retry5from localstack.utils.cloudformation import template_deployer6from botocore.exceptions import ClientError7from botocore.parsers import ResponseParserError8THIS_FOLDER = os.path.dirname(os.path.realpath(__file__))9TEST_TEMPLATE_1 = os.path.join(THIS_FOLDER, 'templates', 'template1.yaml')10TEST_TEMPLATE_2 = os.path.join(THIS_FOLDER, 'templates', 'template2.yaml')11TEST_STACK_NAME = 'test-cf-stack-1'12TEST_STACK_NAME_2 = 'test-cf-stack-2'13def bucket_exists(name):14 s3_client = aws_stack.connect_to_service('s3')15 buckets = s3_client.list_buckets()16 for bucket in buckets['Buckets']:17 if bucket['Name'] == name:18 return True19def queue_exists(name):20 sqs_client = aws_stack.connect_to_service('sqs')21 queues = sqs_client.list_queues()22 for queue_url in queues['QueueUrls']:23 if queue_url.endswith('/%s' % name):24 return True25def queue_url_exists(queue_url):26 sqs_client = aws_stack.connect_to_service('sqs')27 queues = sqs_client.list_queues()28 return queue_url in queues['QueueUrls']29def stream_exists(name):30 kinesis_client = aws_stack.connect_to_service('kinesis')31 streams = kinesis_client.list_streams()32 return name in streams['StreamNames']33def get_stack_details(stack_name):34 cloudformation = aws_stack.connect_to_service('cloudformation')35 stacks = cloudformation.describe_stacks(StackName=stack_name)36 for stack in stacks['Stacks']:37 if stack['StackName'] == stack_name:38 return stack39def describe_stack_resource(stack_name, resource_logical_id):40 cloudformation = aws_stack.connect_to_service('cloudformation')41 response = cloudformation.describe_stack_resources(StackName=stack_name)42 for resource in response['StackResources']:43 if resource['LogicalResourceId'] == resource_logical_id:44 return resource45def list_stack_resources(stack_name):46 cloudformation = aws_stack.connect_to_service('cloudformation')47 response = cloudformation.list_stack_resources(StackName=stack_name)48 return response['StackResourceSummaries']49def get_queue_urls():50 sqs = aws_stack.connect_to_service('sqs')51 response = sqs.list_queues()52 return response['QueueUrls']53def get_topic_arns():54 sqs = aws_stack.connect_to_service('sns')55 response = sqs.list_topics()56 return [t['TopicArn'] for t in response['Topics']]57class CloudFormationTest(unittest.TestCase):58 def test_apply_template(self):59 cloudformation = aws_stack.connect_to_resource('cloudformation')60 template = template_deployer.template_to_json(load_file(TEST_TEMPLATE_1))61 # deploy template62 cloudformation.create_stack(StackName=TEST_STACK_NAME, TemplateBody=template)63 # wait for deployment to finish64 def check_stack():65 stack = get_stack_details(TEST_STACK_NAME)66 assert stack['StackStatus'] == 'CREATE_COMPLETE'67 retry(check_stack, retries=3, sleep=2)68 # assert that bucket has been created69 assert bucket_exists('cf-test-bucket-1')70 # assert that queue has been created71 assert queue_exists('cf-test-queue-1')72 # assert that stream has been created73 assert stream_exists('cf-test-stream-1')74 # assert that queue has been created75 resource = describe_stack_resource(TEST_STACK_NAME, 'SQSQueueNoNameProperty')76 assert queue_exists(resource['PhysicalResourceId'])77 def test_validate_template(self):78 cloudformation = aws_stack.connect_to_service('cloudformation')79 template = template_deployer.template_to_json(load_file(TEST_TEMPLATE_1))80 response = cloudformation.validate_template(TemplateBody=template)81 assert response['ResponseMetadata']['HTTPStatusCode'] == 20082 def test_validate_invalid_json_template_should_fail(self):83 cloudformation = aws_stack.connect_to_service('cloudformation')84 invalid_json = '{"this is invalid JSON"="bobbins"}'85 try:86 cloudformation.validate_template(TemplateBody=invalid_json)87 self.fail('Should raise ValidationError')88 except (ClientError, ResponseParserError) as err:89 if isinstance(err, ClientError):90 self.assertEqual(err.response['ResponseMetadata']['HTTPStatusCode'], 400)91 self.assertEqual(err.response['Error']['Message'], 'Template Validation Error')92 def test_list_stack_resources_returns_queue_urls(self):93 cloudformation = aws_stack.connect_to_resource('cloudformation')94 template = template_deployer.template_to_json(load_file(TEST_TEMPLATE_2))95 cloudformation.create_stack(StackName=TEST_STACK_NAME_2, TemplateBody=template)96 def check_stack():97 stack = get_stack_details(TEST_STACK_NAME_2)98 assert stack['StackStatus'] == 'CREATE_COMPLETE'99 retry(check_stack, retries=3, sleep=2)100 list_stack_summaries = list_stack_resources(TEST_STACK_NAME_2)101 queue_urls = get_queue_urls()102 topic_arns = get_topic_arns()103 stack_queues = [r for r in list_stack_summaries if r['ResourceType'] == 'AWS::SQS::Queue']104 for resource in stack_queues:105 url = aws_stack.get_sqs_queue_url(resource['PhysicalResourceId'])106 self.assertIn(url, queue_urls)107 stack_topics = [r for r in list_stack_summaries if r['ResourceType'] == 'AWS::SNS::Topic']108 for resource in stack_topics:...
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?
