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

How to use enable_vgw_route_propagation method in localstack

Best Python code snippet using localstack_python

vpc.py

Source:vpc.py

...99            ],100        )101    def enable_route_propagation(self, vgw_id, route_id):102        LOG.logger.info("Enabling Propagation .... " + route_id)103        return self._client.enable_vgw_route_propagation(GatewayId=vgw_id, RouteTableId=route_id)104    def disable_route_propagation(self, vgw_id, route_id):105        LOG.logger.info("Disabling Propagation .... " + route_id)106        return self._client.disable_vgw_route_propagation(GatewayId=vgw_id, RouteTableId=route_id)107    def add_route_to_igw_vgw(self, gw_id, rt_id, cidirip):108        LOG.logger.info(f"Adding IP Address {cidirip} and Gateway {gw_id} to  this Route Table {rt_id} ")109        return self._client.create_route(110            DestinationCidrBlock=cidirip,111            GatewayId=gw_id,112            RouteTableId=rt_id113        )114    def add_route_to_nat(self, ngw_id, rt_id, cidirip):115        LOG.logger.info(f"Adding IP Address {cidirip} and Gateway {ngw_id} to  this Route Table {rt_id} ")116        return self._client.create_route(117            DestinationCidrBlock=cidirip,118            NatGatewayId=ngw_id,119            RouteTableId=rt_id120        )121    def add_route_to_tgw(self, tgw_id, rt_id, cidirip):122        LOG.logger.info(f"Adding IP Address {cidirip} and Gateway {tgw_id} to  this Route Table {rt_id} ")123        return self._client.create_route(124            DestinationCidrBlock=cidirip,125            TransitGatewayId=tgw_id,126            RouteTableId=rt_id127        )128    def add_route_to_vpc_peering(self, peer_id, rt_id, cidirip):129        LOG.logger.info(f"Adding IP Address {cidirip} and Gateway {peer_id} to  this Route Table {rt_id} ")130        return self._client.create_route(131            DestinationCidrBlock=cidirip,132            VpcPeeringConnectionId=peer_id,133            RouteTableId=rt_id134        )135    def add_route_to_endpoint_gw(self, vpce_id, rm_rt, add_rt):136        LOG.logger.info("Adding Endpoint " + vpce_id + " Remove Route Table " + rm_rt + " Add Route table " + add_rt)137        return self._client.modify_vpc_endpoint(138            VpcEndpointId=vpce_id,139            RemoveRouteTableIds=[rm_rt],140            AddRouteTableIds=[add_rt],141        )142    def delete_route_record(self, rt_id, cidirip):143        LOG.logger.info("delete IP Address to " + cidirip + " This Route Table " + rt_id)144        return self._client.delete_route(145            DestinationCidrBlock=cidirip,146            RouteTableId=rt_id,147        )148    def enable_propagation_route_table(self, vgw_id, rt_id):149        LOG.logger.info(f"Enable Propagation This Route Table {rt_id}")150        return self._client.enable_vgw_route_propagation(151            GatewayId=vgw_id,152            RouteTableId=rt_id,153        )154    def disable_propagation_route_table(self, vgw_id, rt_id):155        LOG.logger.info(f"Disabling Propagation This Route Table {rt_id}")156        return self._client.disable_vgw_route_propagation(157            GatewayId=vgw_id,158            RouteTableId=rt_id,159        )160    def route_table_associate_with_subnet(self, route_id, subnet_id):161        LOG.logger.info("Associating Route Table " + route_id + " to Subnet " + subnet_id)162        return self._client.associate_route_table(RouteTableId=route_id, SubnetId=subnet_id)163    def route_table_disassociate_with_subnet(self, disassocate_id):164        LOG.logger.info("Disassociating Route " + disassocate_id)...

vpc_connection.py

Source:vpc_connection.py

1########2# Copyright (c) 2015 GigaSpaces Technologies Ltd. All rights reserved3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#        http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12#    * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13#    * See the License for the specific language governing permissions and14#    * limitations under the License.15import xmltodict16# Cloudify imports17from cloudify_aws.base import AwsBaseRelationship18from cloudify_aws import connection, constants, utils19from cloudify import ctx20from cloudify.decorators import operation21@operation22def establish(args=None, **_):23    target_aws_config = ctx.target.node.properties['aws_config']24    client = \25        connection.VPCConnectionClient().client(aws_config=target_aws_config)26    return VpnConnectionExtension(client=client).associate(args)27@operation28def propagate_routes(args=None, **_):29    target_aws_config = ctx.target.node.properties['aws_config']30    client = \31        connection.VPCConnectionClient().client(aws_config=target_aws_config)32    return VpnConnectionExtension(client).propagate_routes(args)33class VpnConnectionExtension(AwsBaseRelationship):34    def __init__(self, client=None):35        super(AwsBaseRelationship, self).__init__(36            client=connection.VPCConnectionClient().client()37        )38        self.source_resource_id = \39            ctx.target.instance.runtime_properties.get('vpn_connection', None)40        self.target_resource_id = \41            ctx.target.instance.runtime_properties.get('vpn_connection', None)42        self.source_get_all_handler = {43            'function': self.client.get_all_vpn_connections,44            'argument':45                '{0}_ids'.format('vpn_connection')46        }47    def associate(self, args):48        vpc_connection = self.filter_for_single_resource(49            self.source_get_all_handler['function'],50            {'vpn_connection_ids': self.source_resource_id}51        )52        if 'available' not in vpc_connection.state:53            return ctx.operation.retry(54                'VPN Connection not ready. Usually more than 2 minutes wait. '55            )56        ctx.logger.debug(57            'CONFIGURATION: {0}'.format(vpc_connection.customer_gateway_configuration)58        )59        customer_gateway_configuration = xmltodict.parse(vpc_connection.customer_gateway_configuration)60        ctx.source.instance.runtime_properties['customer_gateway_configuration'] = customer_gateway_configuration61        vpn_config = customer_gateway_configuration.get('vpn_connection')62        vpn_id = vpn_config.get('@id')63        loop_id = 064        ctx.source.instance.runtime_properties['ipsec_tunnel'] = {}65        for tunnel_config in vpn_config.get('ipsec_tunnel'):66            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)] = {}67            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['id'] = '{0}-{1}'.format(vpn_id,68                                                                                                          loop_id)69            cgw_config = tunnel_config.get('customer_gateway')70            vgw_config = tunnel_config.get('vpn_gateway')71            ike_config = tunnel_config.get('ike')72            ipsec_config = tunnel_config.get('ipsec')73            key_lifetime = ike_config.get('lifetime')74            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['keylife'] = key_lifetime75            raw_encryption_protocol = ike_config.get('encryption_protocol').split('-')76            authentication_protocol = ike_config.get('authentication_protocol')77            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['authentication_protocol'] = \78                authentication_protocol79            proposal = '{0}{1}-{2}'.format(raw_encryption_protocol[0],80                                           raw_encryption_protocol[1],81                                           authentication_protocol)82            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['proposal'] = proposal83            vgw_external_ip = vgw_config.get('tunnel_outside_address').get('ip_address')84            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['remote-gw'] = vgw_external_ip85            preshared_key = ike_config.get('pre_shared_key')86            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['psksecret'] = preshared_key87            dh_group = ike_config.get('perfect_forward_secrecy')88            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['dhgrp'] = dh_group.split('group')[1]89            ipsec_lifetime = ipsec_config.get('lifetime')90            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['keylifeseconds'] = ipsec_lifetime91            cgw_internal_ip = cgw_config.get('tunnel_inside_address').get('ip_address')92            cgw_netmask = cgw_config.get('tunnel_inside_address').get('network_mask')93            cgw_cidr = cgw_config.get('tunnel_inside_address').get('network_cidr')94            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['cgw_internal_ip'] = cgw_internal_ip95            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['cgw_netmask'] = cgw_netmask96            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['cgw_cidr'] = cgw_cidr97            vgw_internal_ip = vgw_config.get('tunnel_inside_address').get('ip_address')98            vgw_netmask = vgw_config.get('tunnel_inside_address').get('network_mask')99            vgw_cidr = vgw_config.get('tunnel_inside_address').get('network_cidr')100            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['vgw_internal_ip'] = vgw_internal_ip101            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['vgw_netmask'] = vgw_netmask102            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['vgw_cidr'] = vgw_cidr103            cgw_external_ip = cgw_config.get('tunnel_outside_address').get('ip_address')104            ctx.source.instance.runtime_properties['ipsec_tunnel'][str(loop_id)]['cgw_external_ip'] = cgw_external_ip105            loop_id += 1106        return True107    def propagate_routes(self, args):108        attachment_args = {}109        attachment_args = utils.update_args(attachment_args, args)110        return self.execute(self.client.enable_vgw_route_propagation,...

vpn.py

Source:vpn.py

...52            raise AwsError(boto3Api="attach_vpn_gateway", error=e)53        54        self.log.info("Attached vpn gateway to vpc: {0}".format(vpcId))55        try:56            enableResp = self.ec2Client.enable_vgw_route_propagation(RouteTableId=routeTableId, 57                                        GatewayId=vpnGwId)58        except Exception as e:59            raise AwsError(boto3Api="enable_vgw_route_propagation", error=e)60        61        62        self.log.info("Enabled virtual gateway route propogation")63        retValue = {'vpnGatewayId': vpnGwId}64        return retValue65    def updateSg(self, **kwargs):66        _requiredArgs = ['sgId', 'custCidrIp']67        try:68            sgId = kwargs['sgId']69            customerCidrBlock = kwargs['custCidrIp']70        except KeyError as e:...

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.