<s> import mock <EOL> import netaddr <EOL> from neutron_lib import constants <EOL> from oslo_config import cfg <EOL> from oslo_db import exception as db_exc <EOL> from oslo_utils import uuidutils <EOL> from neutron . common import exceptions as n_exc <EOL> from neutron import context <EOL> from neutron . ipam import requests as ipam_req <EOL> from neutron . ipam import subnet_alloc <EOL> from neutron import manager <EOL> from neutron . tests . unit . db import test_db_base_plugin_v2 <EOL> from neutron . tests . unit import testlib_api <EOL> class TestSubnetAllocation ( testlib_api . SqlTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestSubnetAllocation , self ) . setUp ( ) <EOL> self . _tenant_id = '<STR_LIT>' <EOL> self . setup_coreplugin ( test_db_base_plugin_v2 . DB_PLUGIN_KLASS ) <EOL> self . plugin = manager . NeutronManager . get_plugin ( ) <EOL> self . ctx = context . get_admin_context ( ) <EOL> cfg . CONF . set_override ( '<STR_LIT>' , True ) <EOL> def _create_subnet_pool ( self , plugin , ctx , name , prefix_list , <EOL> min_prefixlen , ip_version , <EOL> max_prefixlen = constants . ATTR_NOT_SPECIFIED , <EOL> default_prefixlen = constants . ATTR_NOT_SPECIFIED , <EOL> default_quota = constants . ATTR_NOT_SPECIFIED , <EOL> shared = False , is_default = False ) : <EOL> subnetpool = { '<STR_LIT>' : { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : self . _tenant_id , <EOL> '<STR_LIT>' : prefix_list , <EOL> '<STR_LIT>' : min_prefixlen , <EOL> '<STR_LIT>' : max_prefixlen , <EOL> '<STR_LIT>' : default_prefixlen , <EOL> '<STR_LIT>' : shared , <EOL> '<STR_LIT>' : is_default , <EOL> '<STR_LIT>' : default_quota } } <EOL> return plugin . create_subnetpool ( ctx , subnetpool ) <EOL> def _get_subnetpool ( self , ctx , plugin , id ) : <EOL> return plugin . get_subnetpool ( ctx , id ) <EOL> def test_allocate_any_subnet ( self ) : <EOL> prefix_list = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> prefix_list , <NUM_LIT> , <NUM_LIT:4> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . AnySubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> constants . IPv4 , <NUM_LIT> ) <EOL> res = sa . allocate_subnet ( req ) <EOL> detail = res . get_details ( ) <EOL> prefix_set = netaddr . IPSet ( iterable = prefix_list ) <EOL> allocated_set = netaddr . IPSet ( iterable = [ detail . subnet_cidr ] ) <EOL> self . assertTrue ( allocated_set . issubset ( prefix_set ) ) <EOL> self . assertEqual ( <NUM_LIT> , detail . prefixlen ) <EOL> def test_allocate_specific_subnet ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' ) <EOL> res = sa . allocate_subnet ( req ) <EOL> detail = res . get_details ( ) <EOL> sp = self . _get_subnetpool ( self . ctx , self . plugin , sp [ '<STR_LIT:id>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , str ( detail . subnet_cidr ) ) <EOL> self . assertEqual ( <NUM_LIT> , detail . prefixlen ) <EOL> def test_insufficient_prefix_space_for_any_allocation ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . AnySubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> constants . IPv4 , <EOL> <NUM_LIT> ) <EOL> self . assertRaises ( n_exc . SubnetAllocationError , <EOL> sa . allocate_subnet , req ) <EOL> def test_insufficient_prefix_space_for_specific_allocation ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( n_exc . SubnetAllocationError , <EOL> sa . allocate_subnet , req ) <EOL> def test_allocate_any_subnet_gateway ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . AnySubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> constants . IPv4 , <NUM_LIT> ) <EOL> res = sa . allocate_subnet ( req ) <EOL> detail = res . get_details ( ) <EOL> self . assertEqual ( detail . gateway_ip , <EOL> detail . subnet_cidr . network + <NUM_LIT:1> ) <EOL> def test_allocate_specific_subnet_specific_gateway ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' , <EOL> gateway_ip = '<STR_LIT>' ) <EOL> res = sa . allocate_subnet ( req ) <EOL> detail = res . get_details ( ) <EOL> self . assertEqual ( netaddr . IPAddress ( '<STR_LIT>' ) , <EOL> detail . gateway_ip ) <EOL> def test_allocate_specific_ipv6_subnet_specific_gateway ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , <EOL> <NUM_LIT:64> , <NUM_LIT:6> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> with self . ctx . session . begin ( subtransactions = True ) : <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> res = sa . allocate_subnet ( req ) <EOL> detail = res . get_details ( ) <EOL> self . assertEqual ( netaddr . IPAddress ( '<STR_LIT>' ) , <EOL> detail . gateway_ip ) <EOL> def test__allocation_value_for_tenant_no_allocations ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:4> ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> value = sa . _allocations_used_by_tenant ( <NUM_LIT:32> ) <EOL> self . assertEqual ( <NUM_LIT:0> , value ) <EOL> def test_subnetpool_default_quota_exceeded ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:6> , default_quota = <NUM_LIT:1> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( n_exc . SubnetPoolQuotaExceeded , <EOL> sa . allocate_subnet , <EOL> req ) <EOL> def test_subnetpool_concurrent_allocation_exception ( self ) : <EOL> sp = self . _create_subnet_pool ( self . plugin , self . ctx , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , <EOL> <NUM_LIT> , <NUM_LIT:6> , default_quota = <NUM_LIT:1> ) <EOL> sp = self . plugin . _get_subnetpool ( self . ctx , sp [ '<STR_LIT:id>' ] ) <EOL> sa = subnet_alloc . SubnetAllocator ( sp , self . ctx ) <EOL> req = ipam_req . SpecificSubnetRequest ( self . _tenant_id , <EOL> uuidutils . generate_uuid ( ) , <EOL> '<STR_LIT>' ) <EOL> with mock . patch ( "<STR_LIT>" , return_value = <NUM_LIT:0> ) : <EOL> self . assertRaises ( db_exc . RetryRequest , sa . allocate_subnet , req ) </s>
