Join the 1-day Testing & QA Summit featuring 15+ Expert Speakers.Register for FREE! Join TestMu Conference

How to use describe_vpn_connections method in localstack

Best Python code snippet using localstack_python

spokevgw_function.py

Source:spokevgw_function.py

...71    #---------------------------------------------------------------------------------------------------------------------------------------72        for r in regions['Regions']:73                srcregion = r['RegionName']74                regionclient = boto3.client('ec2', region_name=srcregion)75                vpns=regionclient.describe_vpn_connections(Filters=[{'Name':'state','Values':['available', 'attached', 'detached']},{'Name':'tag:'+NameTag+'','Values':[''+NameValue+'']}])76                for l in vpns['VpnConnections']:77                        b = srcregion, l['VpnGatewayId']78                        vpnconnections.append(b)79    #-------------------------------------------------------------------------------------------------------------------------------------80        if (len(vpnconnections)) == 0:81                newvpcconnection = vpngatewayids182        else:83                newvpcconnection = list(set(vpngatewayids1) - set(vpnconnections))84                print (newvpcconnection)85    #-----------------------------------------------------------------------------------------------------------------------------------86        print "-----------------------------------------------------------------------------------------------------------------------------------"87        print "2. Listing all the Vitual Priavte Gatways with  Spoke tags: [('Region', 'VirtualPriavteGatwayID')] \n" +str(vpngatewayids1)88        print "-----------------------------------------------------------------------------------------------------------------------------------"89        print "3. Listing all the Vitual Priavte Gatways with  VPN Connections: [('Region', 'VirtualPriavteGatwayID')] \n" +str(vpnconnections)90        print "-----------------------------------------------------------------------------------------------------------------------------------"91        print "4. New VPN Connection that will be created for this Virtual Private Gateway: [('Region', 'VirtualPriavteGatwayID')])] \n" +str(newvpcconnection)92        print "-----------------------------------------------------------------------------------------------------------------------------------"93    #----------------------------------------------------------------------------------------------------------------------------------94        for n in newvpcconnection:95                nr = boto3.client('ec2', region_name=n[0])96                cg1=nr.create_customer_gateway(Type='ipsec.1',PublicIp=FortinetEIP1,BgpAsn=int(BGP))97                nr.create_tags(Resources=[cg1['CustomerGateway']['CustomerGatewayId']], Tags=[{'Key': 'Name','Value': 'Transit VPC Endpoint1' }])98                cg2=nr.create_customer_gateway(Type='ipsec.1',PublicIp=FortinetEIP2,BgpAsn=int(BGP))99                nr.create_tags(Resources=[cg1['CustomerGateway']['CustomerGatewayId']], Tags=[{'Key': 'Name','Value': 'Transit VPC Endpoint2' }])100                vpn1=nr.create_vpn_connection(Type='ipsec.1',CustomerGatewayId=cg1['CustomerGateway']['CustomerGatewayId'],VpnGatewayId=n[1],Options={'StaticRoutesOnly':False})101                nr.create_tags(Resources=[vpn1['VpnConnection']['VpnConnectionId']],Tags=[{'Key': NameTag,'Value': NameValue }])102                vpn_config1=nr.describe_vpn_connections(VpnConnectionIds=[vpn1['VpnConnection']['VpnConnectionId']])103                vpn_config1=vpn_config1['VpnConnections'][0]['CustomerGatewayConfiguration']104                s3_client = boto3.client('s3')105                s3_client.put_object(106                        Body=str.encode(vpn_config1),107                        Bucket=S3Bucket,108                        Key=''+S3Prefix+''+n[0]+'-'+vpn1['VpnConnection']['VpnConnectionId']+'.conf',109                        ACL='bucket-owner-full-control',110                        #ServerSideEncryption='aws:kms',111                        #SSEKMSKeyId=config['KMS_KEY']112                )113                processfilenames.append(n[0]+'-'+vpn1['VpnConnection']['VpnConnectionId']+'.conf')114                vpn2=nr.create_vpn_connection(Type='ipsec.1',CustomerGatewayId=cg2['CustomerGateway']['CustomerGatewayId'],VpnGatewayId=n[1],Options={'StaticRoutesOnly':False})115                nr.create_tags(Resources=[vpn1['VpnConnection']['VpnConnectionId']],Tags=[{'Key': NameTag,'Value': NameValue }])116                vpn_config2=nr.describe_vpn_connections(VpnConnectionIds=[vpn2['VpnConnection']['VpnConnectionId']])117                vpn_config2=vpn_config2['VpnConnections'][0]['CustomerGatewayConfiguration']118                s3_client = boto3.client('s3')119                s3_client.put_object(120                        Body=str.encode(vpn_config2),121                        Bucket=S3Bucket,122                        Key=''+S3Prefix+''+n[0]+'-'+vpn2['VpnConnection']['VpnConnectionId']+'.conf',123                        ACL='bucket-owner-full-control',124                        #ServerSideEncryption='aws:kms',125                        #SSEKMSKeyId=config['KMS_KEY']126                )127                processfilenames.append(n[0]+'-'+vpn2['VpnConnection']['VpnConnectionId']+'.conf')128        if len(processfilenames)==0:129                print "5. No process files available for Processing"130                print "-----------------------------------------------------------------------------------------------------------------------------------"...

TGW-Static-Route-VPN-Failover.py

Source:TGW-Static-Route-VPN-Failover.py

...12ActiveVPN = os.environ['ActiveVPN']13StandbyVPN = os.environ['StandbyVPN']14TransitGatewayID = os.environ['TransitGatewayID']15def lambda_handler(event, context):16    response_primary_vpn = client.describe_vpn_connections(17        VpnConnectionIds=[18            ActiveVPN,19        ]20    )21    response_secondary_vpn = client.describe_vpn_connections(22        VpnConnectionIds=[23            StandbyVPN,24        ]25    )26    if TransitGatewayID == response_primary_vpn['VpnConnections'][0]['TransitGatewayId'] == \27            response_secondary_vpn['VpnConnections'][0]['TransitGatewayId']:28        print('VPN IDs are valid')29    else:30        print('VPN IDs are not associated with TGW ID ' + TransitGatewayID)31        sys.exit('Exiting the code. Please enter the correct VPN ID which are associated with same TGW ID')32    # fetch TGW Attachment ID and check the TGW RTB for it33    response_primary_vpn = client.describe_transit_gateway_attachments(34        Filters=[35            {36                'Name': 'resource-id',37                'Values': [38                    ActiveVPN,39                ]40            },41        ],42    )43    ActiveVPN_Ass_Id = response_primary_vpn['TransitGatewayAttachments'][0]['TransitGatewayAttachmentId']44    response_secondary_vpn = client.describe_transit_gateway_attachments(45        Filters=[46            {47                'Name': 'resource-id',48                'Values': [49                    StandbyVPN,50                ]51            },52        ],53    )54    PassiveVPN_Ass_Id = response_secondary_vpn['TransitGatewayAttachments'][0]['TransitGatewayAttachmentId']55    TGWRouteTableId = response_primary_vpn['TransitGatewayAttachments'][0]['Association']['TransitGatewayRouteTableId']56    if response_primary_vpn['TransitGatewayAttachments'][0]['Association']['TransitGatewayRouteTableId'] != \57            response_secondary_vpn['TransitGatewayAttachments'][0]['Association']['TransitGatewayRouteTableId']:58        print('Both VPN ' + ActiveVPN + ' ' + StandbyVPN + ' are not associated to same TGW RTB')59        sys.exit('Exiting code. Please associate both VPN to the same TGW RTB')60    # check if Primary VPN is UP or not61    response = client.describe_vpn_connections(VpnConnectionIds=[ActiveVPN])62    for i in range(0, 2):63        TunnelDetails = response['VpnConnections'][0]['VgwTelemetry'][i]64        PrimaryVPNStatus.append(TunnelDetails['Status'])65    # print(PrimaryVPNStatus)66    if PrimaryVPNStatus[0] == PrimaryVPNStatus[1] == 'DOWN':67        PrimaryVPN = 'DOWN'68        print('Primary VPN ' + ActiveVPN + ' is down')69    else:70        PrimaryVPN = 'UP'71        print('Primary VPN ' + ActiveVPN + ' is UP')72    # check if Secondary VPN is UP73    if PrimaryVPNStatus[0] == PrimaryVPNStatus[1] == 'DOWN':74        response = client.describe_vpn_connections(VpnConnectionIds=[StandbyVPN])75        for i in range(0, 2):76            TunnelDetails = response['VpnConnections'][0]['VgwTelemetry'][i]77            SecondaryVPNStatus.append(TunnelDetails['Status'])78        if SecondaryVPNStatus[0] == SecondaryVPNStatus[1] == 'DOWN':79            SecondaryVPN = 'DOWN'80            print('Secondary VPN ' + StandbyVPN + ' is down')81        else:82            SecondaryVPN = 'UP'83            print('Secondary VPN ' + StandbyVPN + ' is UP')84    # check the existing routes which are for primary VPN store the CIDR85    response_route = client.search_transit_gateway_routes(TransitGatewayRouteTableId=TGWRouteTableId,86                                                          Filters=[87                                                              {88                                                                  'Name': 'attachment.transit-gateway-attachment-id',...

vpnConnectionStatus.py

Source:vpnConnectionStatus.py

...11import argparse12import boto313def get_parameters(args):14  client = boto3.client('ec2')15  response = client.describe_vpn_connections(16    VpnConnectionIds=args.vpnconnectionid17  )18  19  for vpncon in response['VpnConnections']:20    if vpncon['State'] == 'available':21      print("OK")22    else:23      print("KO")24def main():25  parser = argparse.ArgumentParser(description="Description: Monitors Site-to-Site VPN Connection status")26  parser.add_argument('-v', '--vpnconnectionid', help="VPN Connection ID to monitor", required=True)27  args = parser.parse_args()28  get_parameters(args)29if __name__ == "__main__":...

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.