KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
