How to use delete_transit_gateway_vpc_attachment method in localstack

Best Python code snippet using localstack_python

test_transit_gateway.py

Source:test_transit_gateway.py

1# Copyright (c) 2018 Cloudify Platform Ltd. All rights reserved2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#        http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14# Standard imports15import unittest16# Third party imports17from mock import patch, MagicMock18from cloudify.exceptions import OperationRetry19# Local imports20from cloudify_aws.common._compat import reload_module21from cloudify_aws.ec2.resources import transit_gateway as mod22from cloudify_aws.common.tests.test_base import (23    TestBase,24    mock_decorator25)26class TestEC2TransitGateway(TestBase):27    def setUp(self):28        self.transit_gateway = mod.EC2TransitGateway(29            "ctx_node",30            resource_id='test_name',31            client=True,32            logger=None)33        mock1 = patch('cloudify_aws.common.decorators.aws_resource',34                      mock_decorator)35        mock2 = patch('cloudify_aws.common.decorators.wait_for_status',36                      mock_decorator)37        mock1.start()38        mock2.start()39        reload_module(mod)40    def test_class_properties(self):41        effect = self.get_client_error_exception(name='EC2 Transit Gateway')42        self.transit_gateway.client = self.make_client_function(43            'describe_transit_gateways', side_effect=effect)44        res = self.transit_gateway.properties45        self.assertEqual(res, {})46        value = {}47        self.transit_gateway.client = self.make_client_function(48            'describe_transit_gateways', return_value=value)49        res = self.transit_gateway.properties50        self.assertEqual(res, value)51        value = {mod.TGS: [{mod.TG_ID: 'test_name'}]}52        self.transit_gateway.client = self.make_client_function(53            'describe_transit_gateways', return_value=value)54        res = self.transit_gateway.properties55        self.assertEqual(res[mod.TG_ID], 'test_name')56    def test_class_status(self):57        value = {58            mod.TGS: [59                {mod.TG_ID: 'test_name', 'State': None}]60        }61        self.transit_gateway.client = self.make_client_function(62            'describe_transit_gateways', return_value=value)63        res = self.transit_gateway.status64        self.assertIsNone(res)65    def test_class_status_positive(self):66        value = {67            mod.TGS: [68                {mod.TG_ID: 'test_name', 'State': 'available'}]69        }70        se = [value, value, value]71        self.transit_gateway.client = self.make_client_function(72            'describe_transit_gateways', side_effect=se)73        res = self.transit_gateway.status74        self.assertEqual(res, 'available')75    def test_class_create(self):76        value = {mod.TG: 'test'}77        self.transit_gateway.client = self.make_client_function(78            'create_transit_gateway', return_value=value)79        res = self.transit_gateway.create(value)80        self.assertEqual(res[mod.TG], value[mod.TG])81    def test_class_delete(self):82        params = {}83        self.transit_gateway.client = self.make_client_function(84            'delete_transit_gateway')85        self.transit_gateway.delete(params)86        self.assertTrue(87            self.transit_gateway.client.delete_transit_gateway.called)88        params = {mod.TG: 'transit gateway'}89        self.transit_gateway.delete(params)90        self.assertEqual(params[mod.TG], 'transit gateway')91    def test_prepare(self):92        ctx = self.get_mock_ctx(mod.TG)93        config = {mod.TG_ID: 'transit gateway'}94        mod.prepare(ctx, mod.EC2TransitGateway, config)95        self.assertEqual(ctx.instance.runtime_properties['resource_config'],96                         config)97    def test_create(self):98        ctx = self.get_mock_ctx(mod.TG)99        config = {mod.TG_ID: 'transit gateway'}100        self.transit_gateway.resource_id = config[mod.TG_ID]101        iface = MagicMock()102        iface.create = self.mock_return({mod.TG: config})103        mod.create(ctx=ctx, iface=iface, resource_config=config)104        self.assertEqual(self.transit_gateway.resource_id, 'transit gateway')105    def test_delete(self):106        ctx = self.get_mock_ctx(mod.TG)107        iface = MagicMock()108        mod.delete(ctx=ctx, iface=iface, resource_config={})109        self.assertTrue(iface.delete.called)110class TestEC2TransitGatewayAttachment(TestBase):111    def setUp(self):112        self.transit_gateway_attachment = mod.EC2TransitGatewayAttachment(113            "ctx_node",114            resource_id=True,115            client=True,116            logger=None)117        mock1 = patch('cloudify_aws.common.decorators.aws_resource',118                      mock_decorator)119        mock2 = patch('cloudify_aws.common.decorators.wait_for_status',120                      mock_decorator)121        mock1.start()122        mock2.start()123        reload_module(mod)124    def test_class_properties(self):125        effect = self.get_client_error_exception(126            name='EC2 Transit Gateway Attachment')127        self.transit_gateway_attachment.client = self.make_client_function(128            'describe_transit_gateway_vpc_attachments', side_effect=effect)129        res = self.transit_gateway_attachment.properties130        self.assertIsNone(res)131        value = {}132        self.transit_gateway_attachment.client = self.make_client_function(133            'describe_transit_gateway_vpc_attachments', return_value=value)134        res = self.transit_gateway_attachment.properties135        self.assertIsNone(res)136        value = {137            mod.TG_ATTACHMENTS: [138                {mod.TG_ATTACHMENT_ID: 'test_name'}]139        }140        self.transit_gateway_attachment.client = self.make_client_function(141            'describe_transit_gateway_vpc_attachments', return_value=value)142        res = self.transit_gateway_attachment.properties143        self.assertEqual(res[mod.TG_ATTACHMENT_ID], 'test_name')144    def test_class_status(self):145        value = {}146        self.transit_gateway_attachment.client = self.make_client_function(147            'describe_transit_gateway_vpc_attachments', return_value=value)148        res = self.transit_gateway_attachment.status149        self.assertIsNone(res)150        value = {151            mod.TG_ATTACHMENTS: [152                {153                    mod.TG_ATTACHMENT_ID: 'test_name',154                    'State': 'available'155                }156            ]157        }158        self.transit_gateway_attachment.client = self.make_client_function(159            'describe_transit_gateway_vpc_attachments', return_value=value)160        res = self.transit_gateway_attachment.status161        self.assertEqual(res, 'available')162    def test_class_create(self):163        value = {mod.TG_ATTACHMENT: 'test'}164        self.transit_gateway_attachment.client = self.make_client_function(165            'create_transit_gateway_vpc_attachment', return_value=value)166        res = self.transit_gateway_attachment.create(value)167        self.assertEqual(res[mod.TG_ATTACHMENT],168                         value[mod.TG_ATTACHMENT])169    def test_class_accept(self):170        value = {mod.TG_ATTACHMENT: 'test'}171        self.transit_gateway_attachment.client = self.make_client_function(172            'accept_transit_gateway_vpc_attachment', return_value=value)173        res = self.transit_gateway_attachment.accept(value)174        self.assertEqual(res[mod.TG_ATTACHMENT],175                         value[mod.TG_ATTACHMENT])176    def test_class_delete(self):177        params = {}178        self.transit_gateway_attachment.client = self.make_client_function(179            'delete_transit_gateway_vpc_attachment')180        self.transit_gateway_attachment.delete(params)181        self.assertTrue(self.transit_gateway_attachment.182                        client.delete_transit_gateway_vpc_attachment.called)183        params = {mod.TG_ATTACHMENT_ID: 'transit gateway'}184        self.transit_gateway_attachment.delete(params)185        self.assertEqual(params[mod.TG_ATTACHMENT_ID],186                         'transit gateway')187    def test_create(self):188        source_ctx = self.get_mock_ctx(mod.TG)189        target_ctx = self.get_mock_ctx(mod.TG)190        ctx = self.get_mock_relationship_ctx(191            mod.TG_ATTACHMENT, test_source=source_ctx,192            test_target=target_ctx)193        config = {mod.TG_ATTACHMENT_ID: 'transit gateway'}194        self.transit_gateway_attachment.resource_id = \195            config[mod.TG_ATTACHMENT_ID]196        iface = MagicMock()197        iface.create = self.mock_return({mod.TG: config})198        mod.request_vpc_attachment(199            ctx=ctx,200            iface=iface,201            transit_gateway_id='transit gateway',202            vpc_id='vpc',203            subnet_ids=['subnet'])204        self.assertIn(mod.TG_ATTACHMENTS,205                      source_ctx.instance.runtime_properties)206    def test_delete(self):207        source_ctx = self.get_mock_ctx(208            mod.TG,209            test_runtime_properties={'aws_resource_id': 'transit gateway'})210        ctx = self.get_mock_relationship_ctx(211            mod.TG_ATTACHMENT, test_source=source_ctx)212        iface = MagicMock(client=MagicMock())213        with self.assertRaises(OperationRetry):214            mod.delete_vpc_attachment(215                ctx=ctx,216                iface=iface,217                transit_gateway_attachment_id='transit gateway')218            self.assertTrue(iface.delete.called)219if __name__ == '__main__':...

transit-gateway.py

Source:transit-gateway.py

...81        DryRun=False82    )83    return response84#Delete transit gateway vpc attachment85def delete_transit_gateway_vpc_attachment(transit_gateway_attachment_id):86    client = boto3.client('ec2', region_name='us-east-1')87    response = client.delete_transit_gateway_vpc_attachment(88        TransitGatewayAttachmentId=transit_gateway_attachment_id,89        DryRun=False90    )91    return response92#Delete transit gateway93def delete_transit_gateway(transit_gateway_id):94    client = boto3.client('ec2', region_name='us-east-1')95    response = client.delete_transit_gateway(96        TransitGatewayId=transit_gateway_id,97        DryRun=False98    )99# dissacotiate transit gateway multicast domain100def disassociate_transit_gateway_multicast_domain(tgw_multicast_domain_id, tgw_attachment_id, subnet_id):101    client = boto3.client('ec2', region_name='us-east-1')102    response = client.disassociate_transit_gateway_multicast_domain(103        TransitGatewayMulticastDomainId=tgw_multicast_domain_id,104        TransitGatewayAttachmentId=tgw_attachment_id,105        SubnetIds=[106            subnet_id,107        ],108        DryRun=False109    )110#register transit gateway multi-cast members111def register_transit_gateway_multicast_group_members(tgw_multicast_domain_id, group_address, network_interface2, network_interface3, network_interface4):112    client = boto3.client('ec2', region_name='us-east-1')113    response = client.register_transit_gateway_multicast_group_members(114    TransitGatewayMulticastDomainId = tgw_multicast_domain_id,115    GroupIpAddress = group_address,116    NetworkInterfaceIds = [117                              network_interface2, network_interface3, network_interface4118                          ],119    DryRun = False120    )121    return response122def register_transit_gateway_multicast_group_sources(tgw_multicast_domain_id, nic1id):123    client = boto3.client('ec2', region_name='us-east-1')124    response = client.register_transit_gateway_multicast_group_sources(125        TransitGatewayMulticastDomainId=tgw_multicast_domain_id,126        GroupIpAddress='224.0.0.0',127        NetworkInterfaceIds=[128            nic1id,129        ],130        DryRun=False131    )132    return response133if __name__ == '__main__':134    action = sys.argv[1]135    if action == "apply":136        #Get variables from cmd line137        vpc_id = sys.argv[2]138        subnet_id = sys.argv[3]139        i1nic = sys.argv[4]140        i2nic = sys.argv[5]141        i3nic = sys.argv[6]142        i4nic = sys.argv[7]143        #Create transit gateway144        tgw_response = create_transit_gateway()145        #Collect transit gateway id from response146        tgw_id = tgw_response['TransitGateway']['TransitGatewayId']147        time.sleep(180)148        #Create transit gateway attachment149        tgw_attachment_response = create_transit_gateway_vpc_attachment(tgw_id, vpc_id, subnet_id)150        #get tgw attachment id response151        tgw_attachment_id = tgw_attachment_response['TransitGatewayVpcAttachment']['TransitGatewayAttachmentId']152        time.sleep(60)153        #Create transit gateway route table and get response154        tgw_rt_response = create_transit_gateway_rt(tgw_id)155        #Get transit gateway route table id from the response156        tgw_rt_id = tgw_rt_response ['TransitGatewayRouteTable']['TransitGatewayRouteTableId']157        time.sleep(90)158        #Create transit gateway route159        create_transit_gateway_route(tgw_rt_id, "10.123.111.0/24", tgw_attachment_id)160        #Create transit gateway multi cast domain161        tgw_multicast_domain_response = create_transit_gateway_multicast_domain(tgw_id)162        #Get transit gateway multi cast domain id163        tgw_multicast_domain_id = tgw_multicast_domain_response['TransitGatewayMulticastDomain']['TransitGatewayMulticastDomainId']164        time.sleep(90)165        #Associate transit gateway multi-cast domain166        associate_transit_gateway_multicast_domain(tgw_multicast_domain_id, tgw_attachment_id, subnet_id)167        time.sleep(30)168        #register transit gateway multicast group members169        register_transit_gateway_multicast_group_members(tgw_multicast_domain_id, "224.0.0.0", i2nic, i3nic, i4nic)170        time.sleep(30)171        #register transit gateway multicast group sources172        register_transit_gateway_multicast_group_sources(tgw_multicast_domain_id,i1nic)173    if action == "destroy":174        #Get values from input175        transit_gateway_multicast_domain_id = sys.argv[2]176        transit_gateway_attachment_id = sys.argv[3]177        transit_gateway_id = sys.argv[4]178        subnet_id = sys.argv[5]179        #Dissacotiate transit gate way multicast domain with subnet180        disassociate_transit_gateway_multicast_domain(transit_gateway_multicast_domain_id, transit_gateway_attachment_id, subnet_id)181        time.sleep(120)182        #Delete transit gateway multicast domain183        delete_transit_gateway_multicast_domain(transit_gateway_multicast_domain_id)184        time.sleep(120)185        #Delete transit gateway VPC attachment186        delete_transit_gateway_vpc_attachment(transit_gateway_attachment_id)187        time.sleep(120)188        #Delete transit gateway...

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.