Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

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.