How to use accept_transit_gateway_vpc_attachment method in localstack

Best Python code snippet using localstack_python

client.py

Source:client.py

...8from typing import List9class Client(BaseClient):10    def accept_reserved_instances_exchange_quote(self, ReservedInstanceIds: List, DryRun: bool = None, TargetConfigurations: List = None) -> Dict:11        pass12    def accept_transit_gateway_vpc_attachment(self, TransitGatewayAttachmentId: str, DryRun: bool = None) -> Dict:13        pass14    def accept_vpc_endpoint_connections(self, ServiceId: str, VpcEndpointIds: List, DryRun: bool = None) -> Dict:15        pass16    def accept_vpc_peering_connection(self, DryRun: bool = None, VpcPeeringConnectionId: str = None) -> Dict:17        pass18    def advertise_byoip_cidr(self, Cidr: str, DryRun: bool = None) -> Dict:19        pass20    def allocate_address(self, Domain: str = None, Address: str = None, PublicIpv4Pool: str = None, DryRun: bool = None) -> Dict:21        pass22    def allocate_hosts(self, AvailabilityZone: str, InstanceType: str, Quantity: int, AutoPlacement: str = None, ClientToken: str = None, TagSpecifications: List = None) -> Dict:23        pass24    def apply_security_groups_to_client_vpn_target_network(self, ClientVpnEndpointId: str, VpcId: str, SecurityGroupIds: List, DryRun: bool = None) -> Dict:25        pass26    def assign_ipv6_addresses(self, NetworkInterfaceId: str, Ipv6AddressCount: int = None, Ipv6Addresses: List = None) -> Dict:...

transit_gateway.py

Source: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'''15    EC2.TransitGateway16    ~~~~~~~~~~~~~~17    AWS EC2 Transit Gateway interface18'''19# Third Party imports20from botocore.exceptions import ClientError, ParamValidationError21from cloudify.exceptions import NonRecoverableError, OperationRetry22# Local imports23from cloudify_aws.ec2 import EC2Base24from cloudify_aws.common import constants, decorators, utils25RESOURCE_TYPE = 'EC2 Transit Gateway'26TG = 'TransitGateway'27TGS = 'TransitGateways'28TG_ID = 'TransitGatewayId'29TG_IDS = 'TransitGatewayIds'30TG_ATTACHMENT = 'TransitGatewayVpcAttachment'31TG_ATTACHMENTS = 'TransitGatewayVpcAttachments'32TG_ATTACHMENT_ID = 'TransitGatewayAttachmentId'33TG_ATTACHMENT_IDS = 'TransitGatewayAttachmentIds'34FAILED = ['failed', 'failing']35AVAILABLE = ['available', 'pendingAcceptance']36PENDING = ['initiatingRequest', 'pending', 'modifying']37UNAVAILABLE = ['deleted',38               'deleting',39               'rollingBack',40               'rejected',41               'rejecting']42class EC2TransitGateway(EC2Base):43    '''44        EC2 Transit Gateway45    '''46    def __init__(self, ctx_node, resource_id=None, client=None, logger=None):47        EC2Base.__init__(self, ctx_node, resource_id, client, logger)48        self.type_name = RESOURCE_TYPE49        self._describe_call = 'describe_transit_gateways'50        self._type_key = TGS51        self._id_key = TG_ID52        self._ids_key = TG_IDS53    @property54    def status(self):55        '''Gets the status of an external resource'''56        props = self.properties57        if not props:58            return None59        return props['State']60    def create(self, params):61        '''62            Create a new AWS EC2 Transit Gateway.63        '''64        return self.make_client_call('create_transit_gateway', params)65    def delete(self, params=None):66        '''67            Deletes an existing AWS EC2 Transit Gateway.68        '''69        self.logger.debug('Deleting %s with parameters: %s'70                          % (self.type_name, params))71        res = self.client.delete_transit_gateway(**params)72        self.logger.debug('Response: %s' % res)73        return res74class EC2TransitGatewayAttachment(EC2Base):75    '''76        EC2 Transit Gateway Attachment77    '''78    def __init__(self, ctx_node, resource_id=None, client=None, logger=None):79        EC2Base.__init__(self, ctx_node, resource_id, client, logger)80        self.type_name = RESOURCE_TYPE81    @property82    def properties(self):83        '''Gets the properties of an external resource'''84        params = {TG_ATTACHMENT_IDS: [self.resource_id]}85        try:86            resources = \87                self.client.describe_transit_gateway_vpc_attachments(**params)88        except (ParamValidationError, ClientError):89            pass90        else:91            return None if not resources else resources.get(92                TG_ATTACHMENTS, [None])[0]93        return None94    @property95    def status(self):96        '''Gets the status of an external resource'''97        props = self.properties98        if not props:99            return None100        return props['State']101    def create(self, params):102        '''103            Create a new AWS EC2 Transit Gateway Attachment.104        '''105        return self.make_client_call(106            'create_transit_gateway_vpc_attachment', params)107    def accept(self, params):108        '''109            Create a new AWS EC2 Transit Gateway Attachment.110        '''111        return self.make_client_call(112            'accept_transit_gateway_vpc_attachment', params)113    def delete(self, params=None):114        '''115            Deletes an existing AWS EC2 Transit Gateway Attachment.116        '''117        return self.make_client_call(118            'delete_transit_gateway_vpc_attachment', params)119@decorators.aws_resource(EC2TransitGateway,120                         resource_type=RESOURCE_TYPE,121                         waits_for_status=False)122def prepare(ctx, iface, resource_config, **_):123    '''Prepares an AWS EC2 Transit Gateway'''124    # Save the parameters125    ctx.instance.runtime_properties['resource_config'] = resource_config126@decorators.aws_resource(EC2TransitGateway, RESOURCE_TYPE)127@decorators.wait_for_status(status_good=['available'],128                            status_pending=['pending'])129@decorators.tag_resources130def create(ctx, iface, resource_config, **_):131    '''Creates an AWS EC2 Transit Gateway'''132    # Actually create the resource133    create_response = iface.create(resource_config)[TG]134    ctx.instance.runtime_properties['create_response'] = \135        utils.JsonCleanuper(create_response).to_dict()136    transit_gateway_id = create_response.get(TG_ID, '')137    iface.update_resource_id(transit_gateway_id)138    utils.update_resource_id(ctx.instance, transit_gateway_id)139@decorators.aws_resource(EC2TransitGateway,140                         RESOURCE_TYPE,141                         ignore_properties=True,142                         waits_for_status=False)143@decorators.untag_resources144def delete(iface, resource_config, **_):145    '''Deletes an AWS EC2 Transit Gateway'''146    if TG_ID not in resource_config:147        resource_config.update({TG_ID: iface.resource_id})148    iface.delete(resource_config)149@decorators.aws_relationship(EC2TransitGatewayAttachment, RESOURCE_TYPE)150def request_vpc_attachment(ctx,151                           iface,152                           transit_gateway_id=None,153                           vpc_id=None,154                           subnet_ids=None,155                           **_):156    transit_gateway_id = transit_gateway_id or \157        ctx.source.instance.runtime_properties.get(158            constants.EXTERNAL_RESOURCE_ID)159    vpc_id = vpc_id or ctx.target.instance.runtime_properties.get(160        constants.EXTERNAL_RESOURCE_ID)161    subnet_ids = subnet_ids or ctx.target.instance.runtime_properties.get(162        'subnets')163    transit_gateway_attachment_id = get_attachment_id_from_runtime_props(ctx)164    if not transit_gateway_id or not vpc_id:165        raise NonRecoverableError(166            'The "cloudify.relationships.aws.ec2.'167            'attach_transit_gateway_to_vpc" relationship operation did not '168            'receive a value for transit_gateway_id '169            '({tgi}) or for vpc_id ({vi}).'.format(170                tgi=transit_gateway_id, vi=vpc_id))171    # If we are retrying then we have this ID.172    # Normally, we could use the @decorators.wait_for_status decorator.173    # However, because this is a relationship neither the source nor the target174    # is an attachment type.175    if transit_gateway_attachment_id:176        iface = EC2TransitGatewayAttachment(177            ctx.source.node,178            transit_gateway_attachment_id,179            iface.client,180            ctx.logger)181        if iface.status in AVAILABLE:182            return183        if iface.status in PENDING:184            raise OperationRetry(185                'The {r} creation request '186                'has been received and is processing. State: {s}.'.format(187                    r=TG_ATTACHMENT, s=iface.status))188        elif iface.status in UNAVAILABLE + FAILED:189            raise NonRecoverableError(190                'The {r} creation request '191                'results in a fatal error: {s}'.format(192                    r=TG_ATTACHMENT,193                    s=iface.status))194        else:195            request = {TG_ATTACHMENT_ID: transit_gateway_attachment_id}196            try:197                iface.accept(request)198            except (NonRecoverableError, ClientError) as e:199                raise OperationRetry(200                    'Waiting for {t} to be in valid state: {s}. '201                    'Error={e}'.format(t=transit_gateway_attachment_id,202                                       s=iface.status,203                                       e=e))204    request = {205        TG_ID: transit_gateway_id,206        'VpcId': vpc_id,207        'SubnetIds': subnet_ids208    }209    try:210        response = iface.create(request)211    except (NonRecoverableError, ClientError) as e:212        raise OperationRetry(213            'Waiting for {t} to be in valid state: {s}. '214            'Error={e}'.format(t=transit_gateway_attachment_id,215                               s=iface.status,216                               e=e))217    ctx.logger.info('Sent the {r} creation request.'.format(218        r=TG_ATTACHMENT))219    ctx.source.instance.runtime_properties[TG_ATTACHMENTS][vpc_id] = \220        utils.JsonCleanuper(response).to_dict()221@decorators.aws_relationship(EC2TransitGatewayAttachment, RESOURCE_TYPE)222def delete_vpc_attachment(ctx, iface, transit_gateway_attachment_id=None, **_):223    transit_gateway_attachment_id = transit_gateway_attachment_id or \224        get_attachment_id_from_runtime_props(ctx)225    if not transit_gateway_attachment_id:226        ctx.logger.error('No transit_gateway_attachment_id was provided. '227                         'Skipping delete attachment.')228        return229    iface = EC2TransitGatewayAttachment(230        ctx.source.node,231        transit_gateway_attachment_id,232        iface.client,233        ctx.logger)234    request = {235        TG_ATTACHMENT_ID: transit_gateway_attachment_id236    }237    if iface.status == 'deleting':238        raise OperationRetry(239            'The {r} deletion request has been received and is processing. '240            'State: {s}.'.format(r=TG_ATTACHMENT, s=iface.status))241    elif iface.status in UNAVAILABLE:242        ctx.logger.info('The {r} has been deleted.'.format(243            r=TG_ATTACHMENT))244        return245    iface.delete(request)246    raise OperationRetry(247        'Sent the {r} deletion request.'.format(r=TG_ATTACHMENT))248def get_attachment_id(props):249    attachment = props.get(TG_ATTACHMENT, {})250    return attachment.get(TG_ATTACHMENT_ID)251def get_attachment_id_from_runtime_props(ctx):252    vpc_id = ctx.target.instance.runtime_properties.get(253        constants.EXTERNAL_RESOURCE_ID)254    if TG_ATTACHMENTS in ctx.source.instance.runtime_properties:255        if vpc_id in ctx.source.instance.runtime_properties[TG_ATTACHMENTS]:256            return ctx.source.instance.runtime_properties[257                TG_ATTACHMENTS][vpc_id][TG_ATTACHMENT][TG_ATTACHMENT_ID]258    else:...

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__':...

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.