| Server IP : 2.57.91.194 / Your IP : 216.73.217.32 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /./opt/gsutil/gslib/vendored/boto/tests/unit/vpc/ |
Upload File : |
# -*- coding: UTF-8 -*-
from tests.unit import unittest
from tests.unit import AWSMockServiceTestCase
from boto.vpc import VPCConnection, VPC
from boto.ec2.securitygroup import SecurityGroup
DESCRIBE_VPCS = b'''<?xml version="1.0" encoding="UTF-8"?>
<DescribeVpcsResponse xmlns="http://ec2.amazonaws.com/doc/2013-02-01/">
<requestId>623040d1-b51c-40bc-8080-93486f38d03d</requestId>
<vpcSet>
<item>
<vpcId>vpc-12345678</vpcId>
<state>available</state>
<cidrBlock>172.16.0.0/16</cidrBlock>
<dhcpOptionsId>dopt-12345678</dhcpOptionsId>
<instanceTenancy>default</instanceTenancy>
<isDefault>false</isDefault>
</item>
</vpcSet>
</DescribeVpcsResponse>'''
class TestDescribeVPCs(AWSMockServiceTestCase):
connection_class = VPCConnection
def default_body(self):
return DESCRIBE_VPCS
def test_get_vpcs(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.get_all_vpcs()
self.assertEqual(len(api_response), 1)
vpc = api_response[0]
self.assertFalse(vpc.is_default)
self.assertEqual(vpc.instance_tenancy, 'default')
class TestCreateVpc(AWSMockServiceTestCase):
connection_class = VPCConnection
def default_body(self):
return b"""
<CreateVpcResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-01/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<vpc>
<vpcId>vpc-1a2b3c4d</vpcId>
<state>pending</state>
<cidrBlock>10.0.0.0/16</cidrBlock>
<dhcpOptionsId>dopt-1a2b3c4d2</dhcpOptionsId>
<instanceTenancy>default</instanceTenancy>
<tagSet/>
</vpc>
</CreateVpcResponse>
"""
def test_create_vpc(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.create_vpc('10.0.0.0/16', 'default')
self.assert_request_parameters({
'Action': 'CreateVpc',
'InstanceTenancy': 'default',
'CidrBlock': '10.0.0.0/16'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp',
'Version'])
self.assertIsInstance(api_response, VPC)
self.assertEqual(api_response.id, 'vpc-1a2b3c4d')
self.assertEqual(api_response.state, 'pending')
self.assertEqual(api_response.cidr_block, '10.0.0.0/16')
self.assertEqual(api_response.dhcp_options_id, 'dopt-1a2b3c4d2')
self.assertEqual(api_response.instance_tenancy, 'default')
class TestDeleteVpc(AWSMockServiceTestCase):
connection_class = VPCConnection
def default_body(self):
return b"""
<DeleteVpcResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-01/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<return>true</return>
</DeleteVpcResponse>
"""
def test_delete_vpc(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.delete_vpc('vpc-1a2b3c4d')
self.assert_request_parameters({
'Action': 'DeleteVpc',
'VpcId': 'vpc-1a2b3c4d'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp',
'Version'])
self.assertEqual(api_response, True)
class TestModifyVpcAttribute(AWSMockServiceTestCase):
connection_class = VPCConnection
def default_body(self):
return b"""
<ModifyVpcAttributeResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-01/">
<requestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</requestId>
<return>true</return>
</ModifyVpcAttributeResponse>
"""
def test_modify_vpc_attribute_dns_support(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.modify_vpc_attribute(
'vpc-1a2b3c4d', enable_dns_support=True)
self.assert_request_parameters({
'Action': 'ModifyVpcAttribute',
'VpcId': 'vpc-1a2b3c4d',
'EnableDnsSupport.Value': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp',
'Version'])
self.assertEqual(api_response, True)
def test_modify_vpc_attribute_dns_hostnames(self):
self.set_http_response(status_code=200)
api_response = self.service_connection.modify_vpc_attribute(
'vpc-1a2b3c4d', enable_dns_hostnames=True)
self.assert_request_parameters({
'Action': 'ModifyVpcAttribute',
'VpcId': 'vpc-1a2b3c4d',
'EnableDnsHostnames.Value': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp',
'Version'])
self.assertEqual(api_response, True)
class TestGetAllClassicLinkVpc(AWSMockServiceTestCase):
connection_class = VPCConnection
def default_body(self):
return b"""
<DescribeVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>2484655d-d669-4950-bf55-7ba559805d36</requestId>
<vpcSet>
<item>
<vpcId>vpc-6226ab07</vpcId>
<classicLinkEnabled>false</classicLinkEnabled>
<tagSet>
<item>
<key>Name</key>
<value>hello</value>[
</item>
</tagSet>
</item>
<item>
<vpcId>vpc-9d24f8f8</vpcId>
<classicLinkEnabled>true</classicLinkEnabled>
<tagSet/>
</item>
</vpcSet>
</DescribeVpcClassicLinkResponse>
"""
def test_get_all_classic_link_vpcs(self):
self.set_http_response(status_code=200)
response = self.service_connection.get_all_classic_link_vpcs()
self.assertEqual(len(response), 2)
vpc = response[0]
self.assertEqual(vpc.id, 'vpc-6226ab07')
self.assertEqual(vpc.classic_link_enabled, 'false')
self.assertEqual(vpc.tags, {'Name': 'hello'})
def test_get_all_classic_link_vpcs_params(self):
self.set_http_response(status_code=200)
self.service_connection.get_all_classic_link_vpcs(
vpc_ids=['id1', 'id2'],
filters={'GroupId': 'sg-9b4343fe'},
dry_run=True,
)
self.assert_request_parameters({
'Action': 'DescribeVpcClassicLink',
'VpcId.1': 'id1',
'VpcId.2': 'id2',
'Filter.1.Name': 'GroupId',
'Filter.1.Value.1': 'sg-9b4343fe',
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
class TestVpcClassicLink(AWSMockServiceTestCase):
connection_class = VPCConnection
def setUp(self):
super(TestVpcClassicLink, self).setUp()
self.vpc = VPC(self.service_connection)
self.vpc_id = 'myid'
self.vpc.id = self.vpc_id
class TestAttachClassicLinkVpc(TestVpcClassicLink):
def default_body(self):
return b"""
<AttachClassicLinkVpcResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>88673bdf-cd16-40bf-87a1-6132fec47257</requestId>
<return>true</return>
</AttachClassicLinkVpcResponse>
"""
def test_attach_classic_link_instance_string_groups(self):
groups = ['sg-foo', 'sg-bar']
self.set_http_response(status_code=200)
response = self.vpc.attach_classic_instance(
instance_id='my_instance_id',
groups=groups,
dry_run=True
)
self.assertTrue(response)
self.assert_request_parameters({
'Action': 'AttachClassicLinkVpc',
'VpcId': self.vpc_id,
'InstanceId': 'my_instance_id',
'SecurityGroupId.1': 'sg-foo',
'SecurityGroupId.2': 'sg-bar',
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
def test_attach_classic_link_instance_object_groups(self):
sec_group_1 = SecurityGroup()
sec_group_1.id = 'sg-foo'
sec_group_2 = SecurityGroup()
sec_group_2.id = 'sg-bar'
groups = [sec_group_1, sec_group_2]
self.set_http_response(status_code=200)
response = self.vpc.attach_classic_instance(
instance_id='my_instance_id',
groups=groups,
dry_run=True
)
self.assertTrue(response)
self.assert_request_parameters({
'Action': 'AttachClassicLinkVpc',
'VpcId': self.vpc_id,
'InstanceId': 'my_instance_id',
'SecurityGroupId.1': 'sg-foo',
'SecurityGroupId.2': 'sg-bar',
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
class TestDetachClassicLinkVpc(TestVpcClassicLink):
def default_body(self):
return b"""
<DetachClassicLinkVpcResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>5565033d-1321-4eef-b121-6aa46f152ed7</requestId>
<return>true</return>
</DetachClassicLinkVpcResponse>
"""
def test_detach_classic_link_instance(self):
self.set_http_response(status_code=200)
response = self.vpc.detach_classic_instance(
instance_id='my_instance_id',
dry_run=True
)
self.assertTrue(response)
self.assert_request_parameters({
'Action': 'DetachClassicLinkVpc',
'VpcId': self.vpc_id,
'InstanceId': 'my_instance_id',
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
class TestEnableClassicLinkVpc(TestVpcClassicLink):
def default_body(self):
return b"""
<EnableVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>4ab2b2b3-a267-4366-a070-bab853b5927d</requestId>
<return>true</return>
</EnableVpcClassicLinkResponse>
"""
def test_enable_classic_link(self):
self.set_http_response(status_code=200)
response = self.vpc.enable_classic_link(
dry_run=True
)
self.assertTrue(response)
self.assert_request_parameters({
'Action': 'EnableVpcClassicLink',
'VpcId': self.vpc_id,
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
class TestDisableClassicLinkVpc(TestVpcClassicLink):
def default_body(self):
return b"""
<DisableVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>4ab2b2b3-a267-4366-a070-bab853b5927d</requestId>
<return>true</return>
</DisableVpcClassicLinkResponse>
"""
def test_enable_classic_link(self):
self.set_http_response(status_code=200)
response = self.vpc.disable_classic_link(
dry_run=True
)
self.assertTrue(response)
self.assert_request_parameters({
'Action': 'DisableVpcClassicLink',
'VpcId': self.vpc_id,
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
class TestUpdateClassicLinkVpc(TestVpcClassicLink):
def default_body(self):
return b"""
<DescribeVpcClassicLinkResponse xmlns="http://ec2.amazonaws.com/doc/2014-09-01/">
<requestId>2484655d-d669-4950-bf55-7ba559805d36</requestId>
<vpcSet>
<item>
<vpcId>myid</vpcId>
<classicLinkEnabled>true</classicLinkEnabled>
<tagSet/>
</item>
</vpcSet>
</DescribeVpcClassicLinkResponse>
"""
def test_vpc_update_classic_link_enabled(self):
self.vpc.classic_link_enabled = False
self.set_http_response(status_code=200)
self.vpc.update_classic_link_enabled(
dry_run=True,
validate=True
)
self.assert_request_parameters({
'Action': 'DescribeVpcClassicLink',
'VpcId.1': self.vpc_id,
'DryRun': 'true'},
ignore_params_values=['AWSAccessKeyId', 'SignatureMethod',
'SignatureVersion', 'Timestamp', 'Version'])
self.assertEqual(self.vpc.classic_link_enabled, 'true')
if __name__ == '__main__':
unittest.main()