KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use cloudformation_type method in localstack

Best Python code snippet using localstack_python

base.py

Source:base.py

1import json2from classes import fields3from classes.functions import Ref4from classes.resource import Resource, ConfigurationError5CIDR_BLOCK_REGEX = r"(?:\d{1,3}\.){3}\d{1,3}(?:/\d\d?)?"6class Vpc(Resource):7    cloudformation_type = "AWS::EC2::VPC"8    # Write out the fields that we could need9    cidr_block = fields.StringField(default="10.0.0.0/16",regex=CIDR_BLOCK_REGEX, required=True)10    enable_dns_hostnames = fields.BooleanField(default=False, required=False)11    enable_dns_support = fields.BooleanField(default=False, required=False)12    instance_tenancy = fields.StringField(default="default", required=False, choices=["dedicated", "default", "host"])13    # GetAtt attributes14    GetAtt_attributes = ["CidrBlock", "CidrBlockAssociations", "DefaultNetworkAcl", "DefaultSecurityGroup", "Ipv6CidrBlocks"]15    def __init__(self, name:str, cidr_block:str=None, enable_dns_hostnames:bool=None, enable_dns_support:bool=None, instance_tenancy:str=None, **kwargs):16        super().__init__(name, **kwargs)17        # instantiate the fields18        self.cidr_block = cidr_block19        self.enable_dns_hostnames = enable_dns_hostnames20        self.enable_dns_support = enable_dns_support21        self.instance_tenancy = instance_tenancy22class Subnet(Resource):23    cloudformation_type = "AWS::EC2::Subnet"24    exclude_null=True25    # Fields26    assign_ipv6_address_on_creation = fields.BooleanField(default=False, required=False)27    availability_zone = fields.StringField(required=True)    28    cidr_block = fields.StringField(regex=CIDR_BLOCK_REGEX, required=True)29    ipv6_cidr_block = fields.StringField(default=None, required=False)30    map_public_ip_on_launch = fields.BooleanField(default=False, required=False)31    outpost_arn = fields.StringField(default=None, required=False)32    vpc_id = fields.RefField(Vpc, required=True)33    GetAtt_attributes = []34    def __init__(35        self,36        name:str,37        assign_ipv6_address_on_creation:bool=assign_ipv6_address_on_creation.default,38        availability_zone:str=None,39        cidr_block:str=None,40        ipv6_cidr_block:str=ipv6_cidr_block.default,41        map_public_ip_on_launch:bool=map_public_ip_on_launch.default,42        outpost_arn:str=None,43        vpc_id:Ref=None,44        **tags45    ):46        super().__init__(name, **tags)47        # assign the fields48        self.assign_ipv6_address_on_creation = assign_ipv6_address_on_creation49        self.availability_zone = availability_zone50        self.cidr_block = cidr_block51        self.ipv6_cidr_block = ipv6_cidr_block52        self.map_public_ip_on_launch = map_public_ip_on_launch53        self.outpost_arn = outpost_arn54        self.vpc_id = vpc_id55class RouteTable(Resource):56    pass57class RouteTableAssociation(Resource):58    pass59class InternetGateway(Resource):60    cloudformation_type = "AWS::EC2::InternetGateway"61    def __init__(self, name:str, **tags):62        super().__init__(name, **tags)63class VpnGateway(Resource):64    pass65class VpcGatewayAttachment(Resource):66    cloudformation_type = "AWS::EC2::VPCGatewayAttachment"67    # Fields68    internet_gateway_id = fields.RefField(InternetGateway, required=False)69    vpc_id = fields.RefField(Vpc, required=True)70    vpn_gateway_id = fields.RefField(VpnGateway, required=False)71    def __init__(72        self,73        name:str,74        internet_gateway_id:Ref=None,75        vpc_id:Ref=None,76        vpn_gateway_id:Ref=None77    ):78        super().__init__(name, include_tags=False)79        # Cannot have both IG and Vpn80        if not (bool(internet_gateway_id) != bool(vpn_gateway_id)):81            raise ConfigurationError("Can only assign InternetGateway -or- VpnGateway")82        self.internet_gateway_id = internet_gateway_id83        self.vpc_id = vpc_id84        self.vpn_gateway_id = vpn_gateway_id85class NatGateway(Resource):...

resource.py

Source:resource.py

1import re, json, inspect2from .fields import Field, TagField3from .functions import AwsNoValue, Ref, GetAtt4from .utils import snake_to_camel5class Resource:6    GetAtt_attributes = []7    cloudformation_type = ""8    exclude_null=False9    def __init__(self, name:str, include_tags:bool=True, depends_on=[], **tags):10        # ensure that the name matches the name regex11        PATTERN_STRING = "^[a-zA-Z]{1}[a-zA-Z0-9]+$"12        NAME_PATTERN = re.compile(PATTERN_STRING)13        if not NAME_PATTERN.match(name):14            raise ValueError(f"The resource's name must match the expression `{PATTERN_STRING}`")15        self.name = name16        # set the tags17        if include_tags:18            self.tags = TagField(**tags)19    20    def __call__(self, attr:str=None):21        """22        Return a 'Ref' or 'GetAtt' object23        """24        if attr is None:25            # Looking for a Ref object26            return Ref(self)27        28        else:29            if attr not in self.GetAtt_attributes:30                raise ValueError(f"Parameter `attr` must be in {self.__class__.__name__}.GetAtt_attributes")31            return GetAtt(self, attr)32    def __setattr__(self, name, value):33        current_value = getattr(self, name, None)34        # If the current value of the changing variable is a Field object,35        # change the field's value instead of the attribute36        if isinstance(current_value, Field):37            try:38                self.__dict__[name].value = value39            except KeyError:40                self.__dict__[name] = self.__class__.__dict__[name]41                self.__dict__[name].value = value42        else:43            super().__setattr__(name, value)44    @property45    def fields(self):46        # Get all variables in the instance that are of type 'fields.Field'47        _func_name = inspect.stack()[0][3]48        fields = {snake_to_camel(attr): getattr(self, attr).value for attr in dir(self) if attr != _func_name and isinstance(getattr(self, attr), Field)}49        new_fields = {}50        # Convert the 'None' field values to {"Ref": "AWS::NoValue"}51        for key, value in fields.items():52            if value is None:53                fields[key] = AwsNoValue.value54            55            if not self.exclude_null:56                new_fields[key] = value57        58        return new_fields59    60    def to_cf_json(self):61        return {62            self.name: {63                "Type": self.cloudformation_type,64                "Properties": self.fields65            }66        }67class CompoundResource(Resource):68    nested_resources = []69    def __call__(self):70        return NotImplementedError()71    def to_cf_json(self):72        result = {}73        for res in self.nested_resources:74            result.update(res.to_cf_json())75        return result76class ConfigurationError(Exception):...

fail_all_resources.py

Source:fail_all_resources.py

1import boto32import json3import os4config = boto3.client('config')5CLOUDFORMATION_TYPE = 'AWS::CloudFormation::Stack'6def handler(event, context):7    invoking_event = json.loads(event['invokingEvent'])8    configItem = invoking_event.get('configurationItem') or invoking_event['configurationItemSummary']9    resource_id = configItem['resourceId']10    resource_type = configItem['resourceType']11    print("FunctionName: " + context.function_name)12    print("ResourceId: " + resource_id)13    print("Resourcetype: " + resource_type)14    compliance_type = 'NON_COMPLIANT'15    annotation = 'No Resources should be deployed in this Region'16    if (17            (context.function_name == resource_id and resource_type == 'AWS::Lambda::Function') or18            (os.environ['StackName'] in resource_id and resource_type == CLOUDFORMATION_TYPE) or19            ('StackSet-' in resource_id and resource_type == CLOUDFORMATION_TYPE)20    ):21        compliance_type = 'COMPLIANT'22        annotation = 'Compliant'23    print('ComplianceStatus: ' + compliance_type)24    config.put_evaluations(25        Evaluations=[26            {27                'ComplianceResourceType': resource_type,28                'ComplianceResourceId': resource_id,29                'ComplianceType': compliance_type,30                'Annotation': annotation,31                'OrderingTimestamp': configItem['configurationItemCaptureTime']32            }33        ],34        ResultToken=event['resultToken']...

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.