<s> import time <EOL> import botocore . exceptions <EOL> from oslo_log import log <EOL> import testtools <EOL> from ec2api . tests . functional import base <EOL> from ec2api . tests . functional import config <EOL> CONF = config . CONF <EOL> LOG = log . getLogger ( __name__ ) <EOL> class AddressTest ( base . EC2TestCase ) : <EOL> @ base . skip_without_vpc ( ) <EOL> def test_create_delete_vpc_address ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> data = self . client . allocate_address ( * [ ] , ** kwargs ) <EOL> id = data [ '<STR_LIT>' ] <EOL> res_clean = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = id ) <EOL> self . assertEqual ( '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> data = self . client . release_address ( AllocationId = id ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> def test_create_delete_standard_address ( self ) : <EOL> data = self . client . allocate_address ( ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> res_clean = self . addResourceCleanUp ( self . client . release_address , <EOL> PublicIp = ip ) <EOL> data = self . client . release_address ( PublicIp = ip ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> @ base . skip_without_vpc ( ) <EOL> def test_invalid_delete_vpc_address ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> data = self . client . allocate_address ( * [ ] , ** kwargs ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> id = data [ '<STR_LIT>' ] <EOL> res_clean = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = id ) <EOL> self . assertEqual ( '<STR_LIT>' , data [ '<STR_LIT>' ] ) <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> PublicIp = ip , AllocationId = id ) <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> PublicIp = ip ) <EOL> data = self . client . release_address ( AllocationId = id ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> if CONF . aws . run_incompatible_tests : <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> PublicIp = ip ) <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> AllocationId = id ) <EOL> kwargs = { <EOL> "<STR_LIT>" : '<STR_LIT>' , <EOL> } <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> ** kwargs ) <EOL> if CONF . aws . run_incompatible_tests : <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . release_address , <EOL> PublicIp = '<STR_LIT>' ) <EOL> def test_invalid_create_address ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> try : <EOL> data = self . client . allocate_address ( * [ ] , ** kwargs ) <EOL> allocation_id = data . get ( '<STR_LIT>' ) <EOL> if allocation_id : <EOL> self . client . release_address ( AllocationId = allocation_id ) <EOL> else : <EOL> public_ip = data . get ( '<STR_LIT>' ) <EOL> self . client . release_address ( PublicIp = public_ip ) <EOL> except botocore . exceptions . ClientError as e : <EOL> self . assertEqual ( '<STR_LIT>' , <EOL> e . response [ '<STR_LIT>' ] [ '<STR_LIT>' ] ) <EOL> @ base . skip_without_vpc ( ) <EOL> def test_describe_vpc_addresses ( self ) : <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> data = self . client . allocate_address ( * [ ] , ** kwargs ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> id = data [ '<STR_LIT>' ] <EOL> res_clean = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = id ) <EOL> data = self . client . describe_addresses ( * [ ] , ** { } ) <EOL> for address in data [ '<STR_LIT>' ] : <EOL> if address . get ( '<STR_LIT>' ) == id : <EOL> self . assertEqual ( '<STR_LIT>' , address [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( ip , address [ '<STR_LIT>' ] ) <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ ip ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( id , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ id ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( ip , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> } <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . describe_addresses , <EOL> ** kwargs ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> } <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . describe_addresses , <EOL> ** kwargs ) <EOL> kwargs = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> data = self . client . allocate_address ( * [ ] , ** kwargs ) <EOL> id2 = data [ '<STR_LIT>' ] <EOL> res_clean2 = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = id2 ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ ip ] , <EOL> '<STR_LIT>' : [ id2 ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( data [ '<STR_LIT>' ] ) ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> self . client . release_address ( AllocationId = id ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> self . client . release_address ( AllocationId = id2 ) <EOL> self . cancelResourceCleanUp ( res_clean2 ) <EOL> def test_describe_standard_addresses ( self ) : <EOL> data = self . client . allocate_address ( * [ ] , ** { } ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> res_clean = self . addResourceCleanUp ( self . client . release_address , <EOL> PublicIp = ip ) <EOL> data = self . client . describe_addresses ( * [ ] , ** { } ) <EOL> for address in data [ '<STR_LIT>' ] : <EOL> if address [ '<STR_LIT>' ] == ip : <EOL> self . assertEqual ( '<STR_LIT>' , address [ '<STR_LIT>' ] ) <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ ip ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( data [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( ip , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ '<STR_LIT>' ] , <EOL> } <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . describe_addresses , <EOL> PublicIps = [ '<STR_LIT>' ] ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> self . client . release_address ( PublicIp = ip ) <EOL> self . cancelResourceCleanUp ( res_clean ) <EOL> @ base . skip_without_vpc ( ) <EOL> @ testtools . skipUnless ( CONF . aws . image_id , "<STR_LIT>" ) <EOL> def test_associate_disassociate_vpc_addresses ( self ) : <EOL> aws_zone = CONF . aws . aws_zone <EOL> base_net = '<STR_LIT>' <EOL> data = self . client . create_vpc ( CidrBlock = base_net + '<STR_LIT>' ) <EOL> vpc_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> clean_vpc = self . addResourceCleanUp ( self . client . delete_vpc , <EOL> VpcId = vpc_id ) <EOL> self . get_vpc_waiter ( ) . wait_available ( vpc_id ) <EOL> cidr = base_net + '<STR_LIT>' <EOL> data = self . client . create_subnet ( VpcId = vpc_id , CidrBlock = cidr , <EOL> AvailabilityZone = aws_zone ) <EOL> subnet_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> clean_subnet = self . addResourceCleanUp ( self . client . delete_subnet , <EOL> SubnetId = subnet_id ) <EOL> instance_id = self . run_instance ( SubnetId = subnet_id ) <EOL> data = self . client . allocate_address ( Domain = '<STR_LIT>' ) <EOL> alloc_id = data [ '<STR_LIT>' ] <EOL> clean_a = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = alloc_id ) <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . associate_address , <EOL> InstanceId = instance_id , AllocationId = alloc_id ) <EOL> data = self . client . create_internet_gateway ( ) <EOL> gw_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> clean_ig = self . addResourceCleanUp ( self . client . delete_internet_gateway , <EOL> InternetGatewayId = gw_id ) <EOL> data = self . client . attach_internet_gateway ( VpcId = vpc_id , <EOL> InternetGatewayId = gw_id ) <EOL> clean_aig = self . addResourceCleanUp ( <EOL> self . client . detach_internet_gateway , <EOL> VpcId = vpc_id , <EOL> InternetGatewayId = gw_id ) <EOL> self . prepare_route ( vpc_id , gw_id ) <EOL> data = self . client . associate_address ( InstanceId = instance_id , <EOL> AllocationId = alloc_id ) <EOL> assoc_id = data [ '<STR_LIT>' ] <EOL> clean_aa = self . addResourceCleanUp ( self . client . disassociate_address , <EOL> AssociationId = assoc_id ) <EOL> self . get_address_assoc_waiter ( ) . wait_available ( <EOL> { '<STR_LIT>' : alloc_id } ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ alloc_id ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( instance_id , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> data = self . client . disassociate_address ( AssociationId = assoc_id ) <EOL> self . cancelResourceCleanUp ( clean_aa ) <EOL> self . get_address_assoc_waiter ( ) . wait_delete ( { '<STR_LIT>' : alloc_id } ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> self . client . detach_internet_gateway ( VpcId = vpc_id , <EOL> InternetGatewayId = gw_id ) <EOL> self . cancelResourceCleanUp ( clean_aig ) <EOL> self . client . delete_internet_gateway ( InternetGatewayId = gw_id ) <EOL> self . cancelResourceCleanUp ( clean_ig ) <EOL> self . client . release_address ( AllocationId = alloc_id ) <EOL> self . cancelResourceCleanUp ( clean_a ) <EOL> self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> self . client . delete_subnet ( SubnetId = subnet_id ) <EOL> self . cancelResourceCleanUp ( clean_subnet ) <EOL> self . get_subnet_waiter ( ) . wait_delete ( subnet_id ) <EOL> self . client . delete_vpc ( VpcId = vpc_id ) <EOL> self . cancelResourceCleanUp ( clean_vpc ) <EOL> self . get_vpc_waiter ( ) . wait_delete ( vpc_id ) <EOL> @ testtools . skipUnless ( CONF . aws . image_id , "<STR_LIT>" ) <EOL> def test_associate_disassociate_standard_addresses ( self ) : <EOL> instance_id = self . run_instance ( ) <EOL> data = self . client . allocate_address ( * [ ] , ** { } ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> clean_a = self . addResourceCleanUp ( self . client . release_address , <EOL> PublicIp = ip ) <EOL> data = self . client . associate_address ( InstanceId = instance_id , <EOL> PublicIp = ip ) <EOL> clean_aa = self . addResourceCleanUp ( self . client . disassociate_address , <EOL> PublicIp = ip ) <EOL> self . get_address_assoc_waiter ( ) . wait_available ( { '<STR_LIT>' : ip } ) <EOL> kwargs = { <EOL> '<STR_LIT>' : [ ip ] , <EOL> } <EOL> data = self . client . describe_addresses ( * [ ] , ** kwargs ) <EOL> self . assertEqual ( instance_id , data [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> data = self . client . disassociate_address ( PublicIp = ip ) <EOL> self . cancelResourceCleanUp ( clean_aa ) <EOL> self . get_address_assoc_waiter ( ) . wait_delete ( { '<STR_LIT>' : ip } ) <EOL> time . sleep ( <NUM_LIT:3> ) <EOL> data = self . client . release_address ( PublicIp = ip ) <EOL> self . cancelResourceCleanUp ( clean_a ) <EOL> data = self . client . terminate_instances ( InstanceIds = [ instance_id ] ) <EOL> self . get_instance_waiter ( ) . wait_delete ( instance_id ) <EOL> @ base . skip_without_vpc ( ) <EOL> def test_disassociate_not_associated_vpc_addresses ( self ) : <EOL> aws_zone = CONF . aws . aws_zone <EOL> base_net = '<STR_LIT>' <EOL> data = self . client . create_vpc ( CidrBlock = base_net + '<STR_LIT>' ) <EOL> vpc_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> clean_vpc = self . addResourceCleanUp ( self . client . delete_vpc , <EOL> VpcId = vpc_id ) <EOL> self . get_vpc_waiter ( ) . wait_available ( vpc_id ) <EOL> cidr = base_net + '<STR_LIT>' <EOL> data = self . client . create_subnet ( VpcId = vpc_id , CidrBlock = cidr , <EOL> AvailabilityZone = aws_zone ) <EOL> subnet_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> clean_subnet = self . addResourceCleanUp ( self . client . delete_subnet , <EOL> SubnetId = subnet_id ) <EOL> data = self . client . allocate_address ( Domain = '<STR_LIT>' ) <EOL> alloc_id = data [ '<STR_LIT>' ] <EOL> ip = data [ '<STR_LIT>' ] <EOL> clean_a = self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = alloc_id ) <EOL> assoc_id = '<STR_LIT>' <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . disassociate_address , <EOL> AssociationId = assoc_id ) <EOL> self . assertRaises ( '<STR_LIT>' , <EOL> self . client . disassociate_address , <EOL> PublicIp = ip ) <EOL> self . client . release_address ( AllocationId = alloc_id ) <EOL> self . cancelResourceCleanUp ( clean_a ) <EOL> self . client . delete_subnet ( SubnetId = subnet_id ) <EOL> self . cancelResourceCleanUp ( clean_subnet ) <EOL> self . get_subnet_waiter ( ) . wait_delete ( subnet_id ) <EOL> self . client . delete_vpc ( VpcId = vpc_id ) <EOL> self . cancelResourceCleanUp ( clean_vpc ) <EOL> self . get_vpc_waiter ( ) . wait_delete ( vpc_id ) <EOL> def test_disassociate_not_associated_standard_addresses ( self ) : <EOL> data = self . client . allocate_address ( Domain = '<STR_LIT>' ) <EOL> ip = data [ '<STR_LIT>' ] <EOL> clean_a = self . addResourceCleanUp ( self . client . release_address , <EOL> PublicIp = ip ) <EOL> data = self . client . disassociate_address ( PublicIp = ip ) <EOL> data = self . client . release_address ( PublicIp = ip ) <EOL> self . cancelResourceCleanUp ( clean_a ) <EOL> @ base . skip_without_vpc ( ) <EOL> @ testtools . skipUnless ( CONF . aws . run_incompatible_tests , <EOL> '<STR_LIT>' ) <EOL> def test_preliminary_associate_address ( self ) : <EOL> vpc_id , subnet_id = self . create_vpc_and_subnet ( '<STR_LIT>' ) <EOL> self . create_and_attach_internet_gateway ( vpc_id ) <EOL> data = self . client . allocate_address ( Domain = '<STR_LIT>' ) <EOL> alloc_id = data [ '<STR_LIT>' ] <EOL> self . addResourceCleanUp ( self . client . release_address , <EOL> AllocationId = alloc_id ) <EOL> data = self . client . create_network_interface ( SubnetId = subnet_id ) <EOL> ni_id = data [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> self . addResourceCleanUp ( self . client . delete_network_interface , <EOL> NetworkInterfaceId = ni_id ) <EOL> self . get_network_interface_waiter ( ) . wait_available ( ni_id ) <EOL> data = self . client . associate_address ( <EOL> AllocationId = alloc_id , NetworkInterfaceId = ni_id ) <EOL> assoc_id = data [ '<STR_LIT>' ] <EOL> self . addResourceCleanUp ( self . client . disassociate_address , <EOL> AssociationId = assoc_id ) </s>
