<s> import mock <EOL> from neutron_lib import constants as n_const <EOL> from neutron_lib import exceptions as n_exc <EOL> from oslo_config import cfg <EOL> from neutron . common import constants <EOL> from neutron . common import ipv6_utils <EOL> from neutron . db import db_base_plugin_common <EOL> from neutron . db import db_base_plugin_v2 <EOL> from neutron . db import ipam_non_pluggable_backend as non_ipam <EOL> from neutron . db import models_v2 <EOL> from neutron . tests import base <EOL> class TestIpamNonPluggableBackend ( base . BaseTestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_generate_ip ( self ) : <EOL> with mock . patch . object ( non_ipam . IpamNonPluggableBackend , <EOL> '<STR_LIT>' ) as generate : <EOL> with mock . patch . object ( non_ipam . IpamNonPluggableBackend , <EOL> '<STR_LIT>' ) as rebuild : <EOL> non_ipam . IpamNonPluggableBackend . _generate_ip ( '<STR_LIT:c>' , '<STR_LIT:s>' ) <EOL> generate . assert_called_once_with ( '<STR_LIT:c>' , '<STR_LIT:s>' ) <EOL> self . assertEqual ( <NUM_LIT:0> , rebuild . call_count ) <EOL> def test_generate_ip_exhausted_pool ( self ) : <EOL> with mock . patch . object ( non_ipam . IpamNonPluggableBackend , <EOL> '<STR_LIT>' ) as generate : <EOL> with mock . patch . object ( non_ipam . IpamNonPluggableBackend , <EOL> '<STR_LIT>' ) as rebuild : <EOL> exception = n_exc . IpAddressGenerationFailure ( net_id = '<STR_LIT:n>' ) <EOL> generate . side_effect = [ exception , None ] <EOL> non_ipam . IpamNonPluggableBackend . _generate_ip ( '<STR_LIT:c>' , '<STR_LIT:s>' ) <EOL> self . assertEqual ( <NUM_LIT:2> , generate . call_count ) <EOL> rebuild . assert_called_once_with ( '<STR_LIT:c>' , '<STR_LIT:s>' ) <EOL> def _validate_rebuild_availability_ranges ( self , pools , allocations , <EOL> expected ) : <EOL> ip_qry = mock . Mock ( ) <EOL> ip_qry . with_lockmode . return_value = ip_qry <EOL> ip_qry . filter_by . return_value = allocations <EOL> pool_qry = mock . Mock ( ) <EOL> pool_qry . options . return_value = pool_qry <EOL> pool_qry . with_lockmode . return_value = pool_qry <EOL> pool_qry . filter_by . return_value = pools <EOL> def return_queries_side_effect ( * args , ** kwargs ) : <EOL> if args [ <NUM_LIT:0> ] == models_v2 . IPAllocation : <EOL> return ip_qry <EOL> if args [ <NUM_LIT:0> ] == models_v2 . IPAllocationPool : <EOL> return pool_qry <EOL> context = mock . Mock ( ) <EOL> context . session . query . side_effect = return_queries_side_effect <EOL> subnets = [ mock . MagicMock ( ) ] <EOL> non_ipam . IpamNonPluggableBackend . _rebuild_availability_ranges ( <EOL> context , subnets ) <EOL> actual = [ [ args [ <NUM_LIT:0> ] . allocation_pool_id , <EOL> args [ <NUM_LIT:0> ] . first_ip , args [ <NUM_LIT:0> ] . last_ip ] <EOL> for _name , args , _kwargs in context . session . add . mock_calls ] <EOL> self . assertEqual ( expected , actual ) <EOL> def test_rebuild_availability_ranges ( self ) : <EOL> pools = [ { '<STR_LIT:id>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT:b>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> allocations = [ { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> expected = [ [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT>' ] ] <EOL> self . _validate_rebuild_availability_ranges ( pools , allocations , <EOL> expected ) <EOL> def test_rebuild_ipv6_availability_ranges ( self ) : <EOL> pools = [ { '<STR_LIT:id>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT:id>' : '<STR_LIT:b>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> allocations = [ { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ] <EOL> expected = [ [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:a>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] , <EOL> [ '<STR_LIT:b>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] ] <EOL> self . _validate_rebuild_availability_ranges ( pools , allocations , <EOL> expected ) <EOL> def _test__allocate_ips_for_port ( self , subnets , port , expected ) : <EOL> cfg . CONF . set_override ( "<STR_LIT>" , None ) <EOL> plugin = db_base_plugin_v2 . NeutronDbPluginV2 ( ) <EOL> with mock . patch . object ( db_base_plugin_common . DbBasePluginCommon , <EOL> '<STR_LIT>' ) as get_subnets : <EOL> with mock . patch . object ( non_ipam . IpamNonPluggableBackend , <EOL> '<STR_LIT>' ) as check_unique : <EOL> context = mock . Mock ( ) <EOL> get_subnets . return_value = subnets <EOL> check_unique . return_value = True <EOL> actual = plugin . ipam . _allocate_ips_for_port ( context , port ) <EOL> self . assertEqual ( expected , actual ) <EOL> def test__allocate_ips_for_port_2_slaac_subnets ( self ) : <EOL> subnets = [ <EOL> { <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:id>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:6> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : u'<STR_LIT>' } , <EOL> { <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : u'<STR_LIT>' , <EOL> '<STR_LIT:id>' : u'<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:6> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : u'<STR_LIT>' } ] <EOL> port = { '<STR_LIT:port>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : n_const . ATTR_NOT_SPECIFIED , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> expected = [ ] <EOL> for subnet in subnets : <EOL> addr = str ( ipv6_utils . get_ipv6_addr_by_EUI64 ( <EOL> subnet [ '<STR_LIT>' ] , port [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] ) ) <EOL> expected . append ( { '<STR_LIT>' : addr , '<STR_LIT>' : subnet [ '<STR_LIT:id>' ] } ) <EOL> self . _test__allocate_ips_for_port ( subnets , port , expected ) <EOL> def test__allocate_ips_for_port_2_slaac_pd_subnets ( self ) : <EOL> subnets = [ <EOL> { <EOL> '<STR_LIT>' : constants . PROVISIONAL_IPV6_PD_PREFIX , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:6> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' } , <EOL> { <EOL> '<STR_LIT>' : constants . PROVISIONAL_IPV6_PD_PREFIX , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:6> , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' } ] <EOL> port = { '<STR_LIT:port>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : n_const . ATTR_NOT_SPECIFIED , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } } <EOL> expected = [ ] <EOL> for subnet in subnets : <EOL> addr = str ( ipv6_utils . get_ipv6_addr_by_EUI64 ( <EOL> subnet [ '<STR_LIT>' ] , port [ '<STR_LIT:port>' ] [ '<STR_LIT>' ] ) ) <EOL> expected . append ( { '<STR_LIT>' : addr , '<STR_LIT>' : subnet [ '<STR_LIT:id>' ] } ) <EOL> self . _test__allocate_ips_for_port ( subnets , port , expected ) </s>
