# Copyright (c) 2015 Infoblox Inc. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import mock import netaddr from neutron_lib import constants from neutron_lib import context as ncontext from neutron_lib import exceptions as n_exc from oslo_config import cfg from oslo_db import exception as db_exc from oslo_utils import netutils from oslo_utils import uuidutils import webob.exc from neutron.db import ipam_backend_mixin from neutron.db import ipam_pluggable_backend from neutron.ipam import requests as ipam_req from neutron.objects import ports as port_obj from neutron.objects import subnet as obj_subnet from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_base class UseIpamMixin(object): def setUp(self): cfg.CONF.set_override("ipam_driver", 'internal') super(UseIpamMixin, self).setUp() class TestIpamHTTPResponse(UseIpamMixin, test_db_base.TestV2HTTPResponse): pass class TestIpamPorts(UseIpamMixin, test_db_base.TestPortsV2): pass class TestIpamNetworks(UseIpamMixin, test_db_base.TestNetworksV2): pass class TestIpamSubnets(UseIpamMixin, test_db_base.TestSubnetsV2): pass class TestIpamSubnetPool(UseIpamMixin, test_db_base.TestSubnetPoolsV2): pass class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): def setUp(self): cfg.CONF.set_override("ipam_driver", 'internal') super(TestDbBasePluginIpam, self).setUp() self.tenant_id = uuidutils.generate_uuid() self.subnet_id = uuidutils.generate_uuid() self.admin_context = ncontext.get_admin_context() def _prepare_mocks(self, address_factory=None, subnet_factory=None): if address_factory is None: address_factory = ipam_req.AddressRequestFactory if subnet_factory is None: subnet_factory = ipam_req.SubnetRequestFactory mocks = { 'driver': mock.Mock(), 'subnet': mock.Mock(), 'subnets': mock.Mock(), 'port': { 'device_owner': constants.DEVICE_OWNER_COMPUTE_PREFIX + 'None' }, 'subnet_request': ipam_req.SpecificSubnetRequest( self.tenant_id, self.subnet_id, '10.0.0.0/24', '10.0.0.1', [netaddr.IPRange('10.0.0.2', '10.0.0.254')]), } mocks['driver'].get_subnet.return_value = mocks['subnet'] mocks['driver'].allocate_subnet.return_value = mocks['subnet'] mocks['driver'].get_allocator.return_value = mocks['subnets'] mocks['subnets'].allocate.return_value = ( '127.0.0.1', uuidutils.generate_uuid()) mocks['driver'].get_subnet_request_factory.return_value = ( subnet_factory) mocks['driver'].get_address_request_factory.return_value = ( address_factory) mocks['subnet'].get_details.return_value = mocks['subnet_request'] return mocks def _prepare_ipam(self): mocks = self._prepare_mocks() mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend() return mocks def _prepare_mocks_with_pool_mock(self, pool_mock, address_factory=None, subnet_factory=None): mocks = self._prepare_mocks(address_factory=address_factory, subnet_factory=subnet_factory) pool_mock.get_instance.return_value = mocks['driver'] return mocks def _get_allocate_mock(self, subnet_id, auto_ip='10.0.0.2', fail_ip='127.0.0.1', exception=n_exc.InvalidInput( error_message='SomeError')): def allocate_mock(request): if isinstance(request, ipam_req.SpecificAddressRequest): if request.address == netaddr.IPAddress(fail_ip): raise exception else: return str(request.address), subnet_id else: return auto_ip, subnet_id return allocate_mock def _get_deallocate_mock(self, fail_ip='127.0.0.1', exception=n_exc.InvalidInput( error_message='SomeError')): def deallocate_mock(ip): if str(ip) == fail_ip: raise exception return deallocate_mock def _validate_allocate_calls(self, expected_calls, mocks): self.assertTrue(mocks['subnets'].allocate.called) actual_calls = mocks['subnets'].allocate.call_args_list self.assertEqual(len(expected_calls), len(actual_calls)) i = 0 for call in expected_calls: if call['ip_address']: self.assertIsInstance(actual_calls[i][0][0], ipam_req.SpecificAddressRequest) self.assertEqual(netaddr.IPAddress(call['ip_address']), actual_calls[i][0][0].address) else: self.assertIsInstance(actual_calls[i][0][0], ipam_req.AnyAddressRequest) i += 1 def _convert_to_ips(self, data): ips = [{'ip_address': ip, 'subnet_id': data[ip][1], 'subnet_cidr': data[ip][0]} for ip in data] return sorted(ips, key=lambda t: t['subnet_cidr']) def _gen_subnet_id(self): return uuidutils.generate_uuid() def test_deallocate_single_ip(self): mocks = self._prepare_ipam() ip = '192.168.12.45' data = {ip: ['192.168.12.0/24', self._gen_subnet_id()]} ips = self._convert_to_ips(data) mocks['ipam']._ipam_deallocate_ips(mock.ANY, mocks['driver'], mock.ANY, ips) mocks['driver'].get_subnet.assert_called_once_with(data[ip][1]) mocks['subnet'].deallocate.assert_called_once_with(ip) def test_deallocate_multiple_ips(self): mocks = self._prepare_ipam() data = {'192.168.43.15': ['192.168.43.0/24', self._gen_subnet_id()], '172.23.158.84': ['172.23.128.0/17', self._gen_subnet_id()], '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]} ips = self._convert_to_ips(data) mocks['ipam']._ipam_deallocate_ips(mock.ANY, mocks['driver'], mock.ANY, ips) get_calls = [mock.call(data[ip][1]) for ip in data] mocks['driver'].get_subnet.assert_has_calls(get_calls, any_order=True) ip_calls = [mock.call(ip) for ip in data] mocks['subnet'].deallocate.assert_has_calls(ip_calls, any_order=True) def _single_ip_allocate_helper(self, mocks, ip, network, subnet): ips = [{'subnet_cidr': network, 'subnet_id': subnet}] if ip: ips[0]['ip_address'] = ip allocated_ips = mocks['ipam']._ipam_allocate_ips( mock.ANY, mocks['driver'], mocks['port'], ips) mocks['driver'].get_allocator.assert_called_once_with([subnet]) self.assertTrue(mocks['subnets'].allocate.called) request = mocks['subnets'].allocate.call_args[0][0] return {'ips': allocated_ips, 'request': request} def test_allocate_single_fixed_ip(self): mocks = self._prepare_ipam() ip = '192.168.15.123' subnet_id = self._gen_subnet_id() mocks['subnets'].allocate.return_value = ip, subnet_id results = self._single_ip_allocate_helper(mocks, ip, '192.168.15.0/24', subnet_id) self.assertIsInstance(results['request'], ipam_req.SpecificAddressRequest) self.assertEqual(netaddr.IPAddress(ip), results['request'].address) self.assertEqual(ip, results['ips'][0]['ip_address'], 'Should allocate the same ip as passed') def test_allocate_single_any_ip(self): mocks = self._prepare_ipam() network = '192.168.15.0/24' ip = '192.168.15.83' subnet_id = self._gen_subnet_id() mocks['subnets'].allocate.return_value = ip, subnet_id results = self._single_ip_allocate_helper(mocks, '', network, subnet_id) self.assertIsInstance(results['request'], ipam_req.AnyAddressRequest) self.assertEqual(ip, results['ips'][0]['ip_address']) def test_allocate_eui64_ip(self): mocks = self._prepare_ipam() ip = {'subnet_id': self._gen_subnet_id(), 'subnet_cidr': '2001:470:abcd::/64', 'mac': '6c:62:6d:de:cf:49', 'eui64_address': True} eui64_ip = netutils.get_ipv6_addr_by_EUI64(ip['subnet_cidr'], ip['mac']) mocks['ipam']._ipam_allocate_ips(mock.ANY, mocks['driver'], mock.ANY, [ip]) request = mocks['subnets'].allocate.call_args[0][0] self.assertIsInstance(request, ipam_req.AutomaticAddressRequest) self.assertEqual(eui64_ip, request.address) def test_allocate_multiple_eui64_ips(self): mocks = self._prepare_ipam() ips = [{'subnet_id': self._gen_subnet_id(), 'subnet_cidr': '2001:470:abcd::/64', 'mac': '6c:62:6d:de:cf:49', 'eui64_address': True}, {'subnet_id': self._gen_subnet_id(), 'subnet_cidr': '2001:360:abcd::/64', 'mac': '6c:62:6d:de:cf:49', 'eui64_address': True}] mocks['ipam']._ipam_allocate_ips(mock.ANY, mocks['driver'], mock.ANY, ips) eui64_ips = [] request_ips = [] i = 0 requests = mocks['subnets'].allocate.call_args_list for ip in ips: eui64_ip = netutils.get_ipv6_addr_by_EUI64(ip['subnet_cidr'], ip['mac']) self.assertIsInstance(requests[i][0][0], ipam_req.AutomaticAddressRequest) self.assertEqual(eui64_ip, requests[i][0][0].address) request_ips.append(requests[i][0][0].address) eui64_ips.append(eui64_ip) i += 1 self.assertEqual(request_ips, eui64_ips) def test_allocate_multiple_ips(self): mocks = self._prepare_ipam() subnet_id = self._gen_subnet_id() data = {'': ['172.23.128.0/17', subnet_id], '192.168.43.15': ['192.168.43.0/24', self._gen_subnet_id()], '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]} ips = self._convert_to_ips(data) mocks['subnets'].allocate.side_effect = self._get_allocate_mock( subnet_id, auto_ip='172.23.128.94') mocks['ipam']._ipam_allocate_ips( mock.ANY, mocks['driver'], mocks['port'], ips) get_calls = [mock.call([data[ip][1]]) for ip in data] mocks['driver'].get_allocator.assert_has_calls( get_calls, any_order=True) self._validate_allocate_calls(ips, mocks) def _test_allocate_multiple_ips_with_exception(self, exc_on_deallocate=False): mocks = self._prepare_ipam() fail_ip = '192.168.43.15' auto_ip = '172.23.128.94' subnet_id = self._gen_subnet_id() data = {'': ['172.23.128.0/17', subnet_id], fail_ip: ['192.168.43.0/24', self._gen_subnet_id()], '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]} ips = self._convert_to_ips(data) mocks['subnets'].allocate.side_effect = self._get_allocate_mock( subnet_id, auto_ip=auto_ip, fail_ip=fail_ip, exception=db_exc.DBDeadlock()) # Exception should be raised on attempt to allocate second ip. # Revert action should be performed for the already allocated ips, # In this test case only one ip should be deallocated # and original error should be reraised self.assertRaises(db_exc.DBDeadlock, mocks['ipam']._ipam_allocate_ips, mock.ANY, mocks['driver'], mocks['port'], ips) # get_subnet should be called only for the first two networks get_calls = [mock.call([data[ip][1]]) for ip in ['', fail_ip]] mocks['driver'].get_allocator.assert_has_calls( get_calls, any_order=True) # Allocate should be called for the first two ips only self._validate_allocate_calls(ips[:-1], mocks) # Deallocate should be called for the first ip only mocks['subnet'].deallocate.assert_called_once_with(auto_ip) def test_allocate_multiple_ips_with_exception(self): self._test_allocate_multiple_ips_with_exception() def test_allocate_multiple_ips_with_exception_on_rollback(self): # Validate that original exception is not replaced with one raised on # rollback (during deallocate) self._test_allocate_multiple_ips_with_exception(exc_on_deallocate=True) def test_deallocate_multiple_ips_with_exception(self): mocks = self._prepare_ipam() fail_ip = '192.168.43.15' data = {fail_ip: ['192.168.43.0/24', self._gen_subnet_id()], '0.10.8.8': ['0.10.0.0/8', self._gen_subnet_id()]} ips = self._convert_to_ips(data) mocks['subnet'].deallocate.side_effect = self._get_deallocate_mock( fail_ip=fail_ip, exception=db_exc.DBDeadlock()) mocks['subnet'].allocate.side_effect = ValueError('Some-error') # Validate that exception from deallocate (DBDeadlock) is not replaced # by exception from allocate (ValueError) in rollback block, # so original exception is not changed self.assertRaises(db_exc.DBDeadlock, mocks['ipam']._ipam_deallocate_ips, mock.ANY, mocks['driver'], mock.ANY, ips) mocks['subnets'].allocate.assert_called_once_with(mock.ANY) def test_test_fixed_ips_for_port_pd_gateway(self): context = mock.Mock() pluggable_backend = ipam_pluggable_backend.IpamPluggableBackend() with self.subnet(cidr=constants.PROVISIONAL_IPV6_PD_PREFIX, ip_version=6) as subnet: subnet = subnet['subnet'] fixed_ips = [{'subnet_id': subnet['id'], 'ip_address': '::1'}] filtered_ips = (pluggable_backend. _test_fixed_ips_for_port(context, subnet['network_id'], fixed_ips, constants.DEVICE_OWNER_ROUTER_INTF, [subnet])) # Assert that ports created on prefix delegation subnets # will be returned without an ip address. This prevents router # interfaces being given the ::1 gateway address. self.assertEqual(1, len(filtered_ips)) self.assertEqual(subnet['id'], filtered_ips[0]['subnet_id']) self.assertNotIn('ip_address', filtered_ips[0]) @mock.patch('neutron.ipam.driver.Pool') def test_create_subnet_over_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) cidr = '192.168.0.0/24' allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}] with self.subnet(allocation_pools=allocation_pools, cidr=cidr): pool_mock.get_instance.assert_called_once_with(None, mock.ANY) self.assertTrue(mocks['driver'].allocate_subnet.called) request = mocks['driver'].allocate_subnet.call_args[0][0] self.assertIsInstance(request, ipam_req.SpecificSubnetRequest) self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr) @mock.patch('neutron.ipam.driver.Pool') def test_create_ipv6_pd_subnet_over_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) cfg.CONF.set_override('ipv6_pd_enabled', True) cidr = constants.PROVISIONAL_IPV6_PD_PREFIX allocation_pools = [netaddr.IPRange('::2', '::ffff:ffff:ffff:ffff')] with self.subnet(cidr=None, ip_version=6, subnetpool_id=constants.IPV6_PD_POOL_ID, ipv6_ra_mode=constants.IPV6_SLAAC, ipv6_address_mode=constants.IPV6_SLAAC): self.assertEqual(2, pool_mock.get_instance.call_count) self.assertTrue(mocks['driver'].allocate_subnet.called) request = mocks['driver'].allocate_subnet.call_args[0][0] self.assertIsInstance(request, ipam_req.SpecificSubnetRequest) self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr) self.assertEqual(allocation_pools, request.allocation_pools) @mock.patch('neutron.ipam.driver.Pool') def test_create_subnet_over_ipam_with_rollback(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) mocks['driver'].allocate_subnet.side_effect = ValueError cidr = '10.0.2.0/24' with self.network() as network: self._create_subnet(self.fmt, network['network']['id'], cidr, expected_res_status=500) pool_mock.get_instance.assert_called_once_with(None, mock.ANY) self.assertTrue(mocks['driver'].allocate_subnet.called) request = mocks['driver'].allocate_subnet.call_args[0][0] self.assertIsInstance(request, ipam_req.SpecificSubnetRequest) self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr) # Verify no subnet was created for network req = self.new_show_request('networks', network['network']['id']) res = req.get_response(self.api) net = self.deserialize(self.fmt, res) self.assertEqual(0, len(net['network']['subnets'])) def _test_rollback_on_subnet_creation(self, pool_mock, driver_mocks): cidr = '10.0.2.0/24' with mock.patch.object( ipam_backend_mixin.IpamBackendMixin, '_save_subnet', side_effect=ValueError), self.network() as network: self._create_subnet(self.fmt, network['network']['id'], cidr, expected_res_status=500) pool_mock.get_instance.assert_any_call(None, mock.ANY) self.assertEqual(2, pool_mock.get_instance.call_count) self.assertTrue(driver_mocks['driver'].allocate_subnet.called) request = driver_mocks['driver'].allocate_subnet.call_args[0][0] self.assertIsInstance(request, ipam_req.SpecificSubnetRequest) self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr) # Verify remove ipam subnet was called driver_mocks['driver'].remove_subnet.assert_called_once_with( self.subnet_id) @mock.patch('neutron.ipam.driver.Pool') def test_ipam_subnet_deallocated_if_create_fails(self, pool_mock): driver_mocks = self._prepare_mocks_with_pool_mock(pool_mock) self._test_rollback_on_subnet_creation(pool_mock, driver_mocks) @mock.patch('neutron.ipam.driver.Pool') def test_ipam_subnet_create_and_rollback_fails(self, pool_mock): driver_mocks = self._prepare_mocks_with_pool_mock(pool_mock) # remove_subnet is called on rollback stage and n_exc.NotFound # typically produces 404 error. Validate that exception from # rollback stage is silenced and main exception (ValueError in this # case) is reraised. So resulting http status should be 500. driver_mocks['driver'].remove_subnet.side_effect = n_exc.NotFound self._test_rollback_on_subnet_creation(pool_mock, driver_mocks) @mock.patch('neutron.ipam.driver.Pool') def test_update_subnet_over_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) cidr = '10.0.0.0/24' allocation_pools = [{'start': '10.0.0.2', 'end': '10.0.0.254'}] with self.subnet(allocation_pools=allocation_pools, cidr=cidr) as subnet: data = {'subnet': {'allocation_pools': [ {'start': '10.0.0.10', 'end': '10.0.0.20'}, {'start': '10.0.0.30', 'end': '10.0.0.40'}]}} req = self.new_update_request('subnets', data, subnet['subnet']['id']) res = req.get_response(self.api) self.assertEqual(200, res.status_code) pool_mock.get_instance.assert_any_call(None, mock.ANY) self.assertEqual(2, pool_mock.get_instance.call_count) self.assertTrue(mocks['driver'].update_subnet.called) request = mocks['driver'].update_subnet.call_args[0][0] self.assertIsInstance(request, ipam_req.SpecificSubnetRequest) self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr) ip_ranges = [netaddr.IPRange(p['start'], p['end']) for p in data['subnet']['allocation_pools']] self.assertEqual(ip_ranges, request.allocation_pools) @mock.patch('neutron.ipam.driver.Pool') def test_delete_subnet_over_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) gateway_ip = '10.0.0.1' cidr = '10.0.0.0/24' res = self._create_network(fmt=self.fmt, name='net', admin_state_up=True) network = self.deserialize(self.fmt, res) subnet = self._make_subnet(self.fmt, network, gateway_ip, cidr, ip_version=4) req = self.new_delete_request('subnets', subnet['subnet']['id']) res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int) pool_mock.get_instance.assert_any_call(None, mock.ANY) self.assertEqual(2, pool_mock.get_instance.call_count) mocks['driver'].remove_subnet.assert_called_once_with( subnet['subnet']['id']) @mock.patch('neutron.ipam.driver.Pool') def test_delete_subnet_over_ipam_with_rollback(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) mocks['driver'].remove_subnet.side_effect = ValueError gateway_ip = '10.0.0.1' cidr = '10.0.0.0/24' res = self._create_network(fmt=self.fmt, name='net', admin_state_up=True) network = self.deserialize(self.fmt, res) subnet = self._make_subnet(self.fmt, network, gateway_ip, cidr, ip_version=4) req = self.new_delete_request('subnets', subnet['subnet']['id']) res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPServerError.code, res.status_int) pool_mock.get_instance.assert_any_call(None, mock.ANY) self.assertEqual(2, pool_mock.get_instance.call_count) mocks['driver'].remove_subnet.assert_called_once_with( subnet['subnet']['id']) # Verify subnet was recreated after failed ipam call subnet_req = self.new_show_request('subnets', subnet['subnet']['id']) raw_res = subnet_req.get_response(self.api) sub_res = self.deserialize(self.fmt, raw_res) self.assertIn(sub_res['subnet']['cidr'], cidr) self.assertIn(sub_res['subnet']['gateway_ip'], gateway_ip) @mock.patch('neutron.ipam.driver.Pool') def test_create_port_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) auto_ip = '10.0.0.2' expected_calls = [{'ip_address': ''}] with self.subnet() as subnet: mocks['subnets'].allocate.side_effect = self._get_allocate_mock( subnet['subnet']['id'], auto_ip=auto_ip) with self.port(subnet=subnet) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(ips[0]['ip_address'], auto_ip) self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id']) self._validate_allocate_calls(expected_calls, mocks) @mock.patch('neutron.ipam.driver.Pool') def test_create_port_ipam_with_rollback(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) mocks['subnet'].allocate.side_effect = ValueError with self.network() as network: with self.subnet(network=network): net_id = network['network']['id'] data = { 'port': {'network_id': net_id, 'tenant_id': network['network']['tenant_id']}} port_req = self.new_create_request('ports', data) res = port_req.get_response(self.api) self.assertEqual(webob.exc.HTTPServerError.code, res.status_int) # verify no port left after failure req = self.new_list_request('ports', self.fmt, "network_id=%s" % net_id) res = self.deserialize(self.fmt, req.get_response(self.api)) self.assertEqual(0, len(res['ports'])) @mock.patch('neutron.ipam.driver.Pool') def test_update_port_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) auto_ip = '10.0.0.2' new_ip = '10.0.0.15' expected_calls = [{'ip_address': ip} for ip in ['', new_ip]] with self.subnet() as subnet: mocks['subnets'].allocate.side_effect = self._get_allocate_mock( subnet['subnet']['id'], auto_ip=auto_ip) with self.port(subnet=subnet) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(auto_ip, ips[0]['ip_address']) # Update port with another new ip data = {"port": {"fixed_ips": [{ 'subnet_id': subnet['subnet']['id'], 'ip_address': new_ip}]}} req = self.new_update_request('ports', data, port['port']['id']) res = self.deserialize(self.fmt, req.get_response(self.api)) ips = res['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(new_ip, ips[0]['ip_address']) # Allocate should be called for the first two networks self._validate_allocate_calls(expected_calls, mocks) # Deallocate should be called for the first ip only mocks['subnet'].deallocate.assert_called_once_with(auto_ip) @mock.patch('neutron.ipam.driver.Pool') def test_delete_port_ipam(self, pool_mock): mocks = self._prepare_mocks_with_pool_mock(pool_mock) auto_ip = '10.0.0.2' with self.subnet() as subnet: mocks['subnets'].allocate.side_effect = self._get_allocate_mock( subnet['subnet']['id'], auto_ip=auto_ip) with self.port(subnet=subnet) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(auto_ip, ips[0]['ip_address']) req = self.new_delete_request('ports', port['port']['id']) res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int) mocks['subnet'].deallocate.assert_called_once_with(auto_ip) def test_recreate_port_ipam(self): with self.subnet() as subnet: subnet_cidr = subnet['subnet']['cidr'] with self.port(subnet=subnet) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) orig_ip = ips[0]['ip_address'] self.assertIn(netaddr.IPAddress(ips[0]['ip_address']), netaddr.IPSet(netaddr.IPNetwork(subnet_cidr))) req = self.new_delete_request('ports', port['port']['id']) res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int) with self.port(subnet=subnet, fixed_ips=ips) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(orig_ip, ips[0]['ip_address']) def test_recreate_port_ipam_specific_ip(self): with self.subnet() as subnet: ip = '10.0.0.2' fixed_ip_data = [{'subnet_id': subnet['subnet']['id'], 'ip_address': ip}] with self.port(subnet=subnet, fixed_ips=fixed_ip_data) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(ip, ips[0]['ip_address']) req = self.new_delete_request('ports', port['port']['id']) res = req.get_response(self.api) self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int) with self.port(subnet=subnet, fixed_ips=ips) as port: ips = port['port']['fixed_ips'] self.assertEqual(1, len(ips)) self.assertEqual(ip, ips[0]['ip_address']) @mock.patch('neutron.ipam.driver.Pool') def test_update_ips_for_port_passes_port_dict_to_factory(self, pool_mock): address_factory = mock.Mock() mocks = self._prepare_mocks_with_pool_mock( pool_mock, address_factory=address_factory) context = mock.Mock() new_ips = mock.Mock() original_ips = mock.Mock() mac = mock.Mock() ip_dict = {'ip_address': '192.1.1.10', 'subnet_id': uuidutils.generate_uuid()} changes = ipam_pluggable_backend.IpamPluggableBackend.Changes( add=[ip_dict], original=[], remove=[]) changes_mock = mock.Mock(return_value=changes) fixed_ips_mock = mock.Mock(return_value=changes.add) mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend() mocks['ipam']._get_changed_ips_for_port = changes_mock mocks['ipam']._ipam_get_subnets = mock.Mock(return_value=[]) mocks['ipam']._test_fixed_ips_for_port = fixed_ips_mock mocks['ipam']._update_ips_for_pd_subnet = mock.Mock(return_value=[]) port_dict = {'device_owner': uuidutils.generate_uuid(), 'network_id': uuidutils.generate_uuid()} mocks['ipam']._update_ips_for_port(context, port_dict, None, original_ips, new_ips, mac) mocks['driver'].get_address_request_factory.assert_called_once_with() mocks['ipam']._ipam_get_subnets.assert_called_once_with( context, network_id=port_dict['network_id'], host=None, service_type=port_dict['device_owner']) # Validate port_dict is passed into address_factory address_factory.get_request.assert_called_once_with(context, port_dict, ip_dict) @mock.patch('neutron.ipam.driver.Pool') def test_update_ips_for_port_passes_port_id_to_factory(self, pool_mock): port_id = uuidutils.generate_uuid() network_id = uuidutils.generate_uuid() address_factory = mock.Mock() mocks = self._prepare_mocks_with_pool_mock( pool_mock, address_factory=address_factory) context = mock.Mock() ip_dict = {'ip_address': '192.1.1.10', 'subnet_id': uuidutils.generate_uuid()} port_dict = {'port': {'device_owner': uuidutils.generate_uuid(), 'network_id': network_id, 'fixed_ips': [ip_dict]}} subnets = [{'id': ip_dict['subnet_id'], 'network_id': network_id, 'cidr': '192.1.1.0/24', 'ip_version': 4, 'ipv6_address_mode': None, 'ipv6_ra_mode': None}] get_subnets_mock = mock.Mock(return_value=subnets) get_subnet_mock = mock.Mock(return_value=subnets[0]) mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend() mocks['ipam']._ipam_get_subnets = get_subnets_mock mocks['ipam']._get_subnet = get_subnet_mock with mock.patch.object(port_obj.IPAllocation, 'create'): mocks['ipam'].allocate_ips_for_port_and_store(context, port_dict, port_id) mocks['driver'].get_address_request_factory.assert_called_once_with() port_dict_with_id = port_dict['port'].copy() port_dict_with_id['id'] = port_id # Validate port id is added to port dict before address_factory call ip_dict.pop('device_owner') address_factory.get_request.assert_called_once_with(context, port_dict_with_id, ip_dict) # Verify incoming port dict is not changed ('id' is not added to it) self.assertIsNone(port_dict['port'].get('id')) def _test_update_db_subnet(self, pool_mock, subnet, expected_subnet, old_pools): subnet_factory = mock.Mock() context = self.admin_context if 'cidr' in subnet: subnet['cidr'] = netaddr.IPNetwork(subnet['cidr']) if 'cidr' in expected_subnet: expected_subnet['cidr'] = netaddr.IPNetwork( expected_subnet['cidr']) mocks = self._prepare_mocks_with_pool_mock( pool_mock, subnet_factory=subnet_factory) mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend() mocks['ipam'].update_db_subnet( context, subnet['id'], subnet, old_pools) mocks['driver'].get_subnet_request_factory.assert_called_once_with() subnet_factory.get_request.assert_called_once_with(context, expected_subnet, None) @mock.patch('neutron.ipam.driver.Pool') def test_update_db_subnet_unchanged_pools(self, pool_mock): old_pools = [{'start': '192.1.1.2', 'end': '192.1.1.254'}] context = self.admin_context subnet = {'id': uuidutils.generate_uuid(), 'ip_version': constants.IP_VERSION_4, 'cidr': netaddr.IPNetwork('192.1.1.0/24'), 'ipv6_address_mode': None, 'ipv6_ra_mode': None} subnet_with_pools = subnet.copy() subnet_obj = obj_subnet.Subnet(context, **subnet_with_pools) subnet_obj.create() subnet_with_pools['allocation_pools'] = old_pools # if subnet has no allocation pools set, then old pools has to # be added to subnet dict passed to request factory self._test_update_db_subnet(pool_mock, subnet, subnet_with_pools, old_pools) @mock.patch('neutron.ipam.driver.Pool') def test_update_db_subnet_new_pools(self, pool_mock): old_pools = [{'start': '192.1.1.2', 'end': '192.1.1.254'}] context = self.admin_context subnet = {'id': uuidutils.generate_uuid(), 'ip_version': constants.IP_VERSION_4, 'cidr': netaddr.IPNetwork('192.1.1.0/24'), 'ipv6_address_mode': None, 'ipv6_ra_mode': None} # make a copy of subnet for validation, since update_subnet changes # incoming subnet dict expected_subnet = subnet.copy() subnet_obj = obj_subnet.Subnet(context, **subnet) subnet_obj.create() subnet['allocation_pools'] = [ netaddr.IPRange('192.1.1.10', '192.1.1.254')] expected_subnet = subnet.copy() obj_subnet.IPAllocationPool(context, subnet_id=subnet['id'], start='192.1.1.10', end='192.1.1.254').create() # validate that subnet passed to request factory is the same as # incoming one, i.e. new pools in it are not overwritten by old pools self._test_update_db_subnet(pool_mock, subnet, expected_subnet, old_pools) @mock.patch('neutron.ipam.driver.Pool') def test_update_db_subnet_new_pools_exception(self, pool_mock): context = mock.Mock() mocks = self._prepare_mocks_with_pool_mock(pool_mock) mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend() new_port = {'fixed_ips': [{'ip_address': '192.168.1.20', 'subnet_id': uuidutils.generate_uuid()}, {'ip_address': '192.168.1.50', 'subnet_id': uuidutils.generate_uuid()}]} db_port = port_obj.Port(context, id=uuidutils.generate_uuid(), network_id=uuidutils.generate_uuid()) old_port = {'fixed_ips': [{'ip_address': '192.168.1.10', 'subnet_id': uuidutils.generate_uuid()}, {'ip_address': '192.168.1.50', 'subnet_id': uuidutils.generate_uuid()}]} changes = mocks['ipam'].Changes( add=[{'ip_address': '192.168.1.20', 'subnet_id': uuidutils.generate_uuid()}], original=[{'ip_address': '192.168.1.50', 'subnet_id': uuidutils.generate_uuid()}], remove=[{'ip_address': '192.168.1.10', 'subnet_id': uuidutils.generate_uuid()}]) mocks['ipam']._delete_ip_allocation = mock.Mock() mocks['ipam']._make_port_dict = mock.Mock(return_value=old_port) mocks['ipam']._update_ips_for_port = mock.Mock(return_value=changes) mocks['ipam']._update_db_port = mock.Mock( side_effect=db_exc.DBDeadlock) # emulate raising exception on rollback actions mocks['ipam']._ipam_deallocate_ips = mock.Mock(side_effect=ValueError) mocks['ipam']._ipam_allocate_ips = mock.Mock(side_effect=ValueError) # Validate original exception (DBDeadlock) is not overridden by # exception raised on rollback (ValueError) with mock.patch.object(port_obj.IPAllocation, 'create'): self.assertRaises(db_exc.DBDeadlock, mocks['ipam'].update_port_with_ips, context, None, db_port, new_port, mock.Mock()) mocks['ipam']._ipam_deallocate_ips.assert_called_once_with( context, mocks['driver'], db_port, changes.add, revert_on_fail=False) mocks['ipam']._ipam_allocate_ips.assert_called_once_with( context, mocks['driver'], db_port, changes.remove, revert_on_fail=False) class TestRollback(test_db_base.NeutronDbPluginV2TestCase): def setUp(self): cfg.CONF.set_override('ipam_driver', 'internal') super(TestRollback, self).setUp() def test_ipam_rollback_not_broken_on_session_rollback(self): """Triggers an error that calls rollback on session.""" with self.network() as net: with self.subnet(network=net, cidr='10.0.1.0/24') as subnet1: with self.subnet(network=net, cidr='10.0.2.0/24') as subnet2: pass # If this test fails and this method appears in the server side stack # trace then IPAM rollback was likely tried using a session which had # already been rolled back by the DB exception. def rollback(func, *args, **kwargs): func(*args, **kwargs) # Ensure DBDuplicate exception is raised in the context where IPAM # rollback is triggered. It "breaks" the session because it triggers DB # rollback. Inserting a flush in _store_ip_allocation does this. orig = ipam_pluggable_backend.IpamPluggableBackend._store_ip_allocation def store(context, ip_address, *args, **kwargs): try: return orig(context, ip_address, *args, **kwargs) finally: context.session.flush() # Create a port to conflict with later. Simulates a race for addresses. result = self._create_port( self.fmt, net_id=net['network']['id'], fixed_ips=[{'subnet_id': subnet1['subnet']['id']}, {'subnet_id': subnet2['subnet']['id']}]) port = self.deserialize(self.fmt, result) fixed_ips = port['port']['fixed_ips'] # Hands out the same 2nd IP to create conflict and trigger rollback ips = [{'subnet_id': fixed_ips[0]['subnet_id'], 'ip_address': fixed_ips[0]['ip_address']}, {'subnet_id': fixed_ips[1]['subnet_id'], 'ip_address': fixed_ips[1]['ip_address']}] def alloc(*args, **kwargs): def increment_address(a): a['ip_address'] = str(netaddr.IPAddress(a['ip_address']) + 1) # Increment 1st address to return a free address on the first call increment_address(ips[0]) try: return copy.deepcopy(ips) finally: # Increment 2nd address to return free address on the 2nd call increment_address(ips[1]) Backend = ipam_pluggable_backend.IpamPluggableBackend with mock.patch.object(Backend, '_store_ip_allocation', wraps=store),\ mock.patch.object(Backend, '_safe_rollback', wraps=rollback),\ mock.patch.object(Backend, '_allocate_ips_for_port', wraps=alloc): # Create port with two addresses. The wrapper lets one succeed # then simulates race for the second to trigger IPAM rollback. response = self._create_port( self.fmt, net_id=net['network']['id'], fixed_ips=[{'subnet_id': subnet1['subnet']['id']}, {'subnet_id': subnet2['subnet']['id']}]) # When all goes well, retry kicks in and the operation is successful. self.assertEqual(webob.exc.HTTPCreated.code, response.status_int)