neutron/tests/unit/db/test_ipam_pluggable_backend.py

   1 # Copyright (c) 2015 Infoblox Inc.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17 import netaddr
  18 import webob.exc
  19
  20 from oslo_config import cfg
  21 from oslo_utils import uuidutils
  22
  23 from neutron.common import constants
  24 from neutron.common import exceptions as n_exc
  25 from neutron.common import ipv6_utils
  26 from neutron.db import ipam_backend_mixin
  27 from neutron.db import ipam_pluggable_backend
  28 from neutron.ipam import requests as ipam_req
  29 from neutron.tests.unit.db import test_db_base_plugin_v2 as test_db_base
  30
  31
  32 class UseIpamMixin(object):
  33
  34     def setUp(self):
  35         cfg.CONF.set_override("ipam_driver", 'internal')
  36         super(UseIpamMixin, self).setUp()
  37
  38
  39 class TestIpamHTTPResponse(UseIpamMixin, test_db_base.TestV2HTTPResponse):
  40     pass
  41
  42
  43 class TestIpamPorts(UseIpamMixin, test_db_base.TestPortsV2):
  44     pass
  45
  46
  47 class TestIpamNetworks(UseIpamMixin, test_db_base.TestNetworksV2):
  48     pass
  49
  50
  51 class TestIpamSubnets(UseIpamMixin, test_db_base.TestSubnetsV2):
  52     pass
  53
  54
  55 class TestIpamSubnetPool(UseIpamMixin, test_db_base.TestSubnetPoolsV2):
  56     pass
  57
  58
  59 class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
  60     def setUp(self):
  61         cfg.CONF.set_override("ipam_driver", 'internal')
  62         super(TestDbBasePluginIpam, self).setUp()
  63         self.tenant_id = uuidutils.generate_uuid()
  64         self.subnet_id = uuidutils.generate_uuid()
  65
  66     def _prepare_mocks(self, address_factory=None):
  67         if address_factory is None:
  68             address_factory = ipam_req.AddressRequestFactory
  69
  70         mocks = {
  71             'driver': mock.Mock(),
  72             'subnet': mock.Mock(),
  73             'subnet_request': ipam_req.SpecificSubnetRequest(
  74                 self.tenant_id,
  75                 self.subnet_id,
  76                 '10.0.0.0/24',
  77                 '10.0.0.1',
  78                 [netaddr.IPRange('10.0.0.2', '10.0.0.254')]),
  79         }
  80         mocks['driver'].get_subnet.return_value = mocks['subnet']
  81         mocks['driver'].allocate_subnet.return_value = mocks['subnet']
  82         mocks['driver'].get_subnet_request_factory.return_value = (
  83             ipam_req.SubnetRequestFactory)
  84         mocks['driver'].get_address_request_factory.return_value = (
  85             address_factory)
  86         mocks['subnet'].get_details.return_value = mocks['subnet_request']
  87         return mocks
  88
  89     def _prepare_ipam(self):
  90         mocks = self._prepare_mocks()
  91         mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend()
  92         return mocks
  93
  94     def _prepare_mocks_with_pool_mock(self, pool_mock, address_factory=None):
  95         mocks = self._prepare_mocks(address_factory=address_factory)
  96         pool_mock.get_instance.return_value = mocks['driver']
  97         return mocks
  98
  99     def _get_allocate_mock(self, auto_ip='10.0.0.2',
 100                            fail_ip='127.0.0.1',
 101                            error_message='SomeError'):
 102         def allocate_mock(request):
 103             if type(request) == ipam_req.SpecificAddressRequest:
 104                 if request.address == netaddr.IPAddress(fail_ip):
 105                     raise n_exc.InvalidInput(error_message=error_message)
 106                 else:
 107                     return str(request.address)
 108             else:
 109                 return auto_ip
 110
 111         return allocate_mock
 112
 113     def _validate_allocate_calls(self, expected_calls, mocks):
 114         self.assertTrue(mocks['subnet'].allocate.called)
 115
 116         actual_calls = mocks['subnet'].allocate.call_args_list
 117         self.assertEqual(len(expected_calls), len(actual_calls))
 118
 119         i = 0
 120         for call in expected_calls:
 121             if call['ip_address']:
 122                 self.assertIsInstance(actual_calls[i][0][0],
 123                                       ipam_req.SpecificAddressRequest)
 124                 self.assertEqual(netaddr.IPAddress(call['ip_address']),
 125                                  actual_calls[i][0][0].address)
 126             else:
 127                 self.assertIsInstance(actual_calls[i][0][0],
 128                                       ipam_req.AnyAddressRequest)
 129             i += 1
 130
 131     def _convert_to_ips(self, data):
 132         ips = [{'ip_address': ip,
 133                 'subnet_id': data[ip][1],
 134                 'subnet_cidr': data[ip][0]} for ip in data]
 135         return sorted(ips, key=lambda t: t['subnet_cidr'])
 136
 137     def _gen_subnet_id(self):
 138         return uuidutils.generate_uuid()
 139
 140     def test_deallocate_single_ip(self):
 141         mocks = self._prepare_ipam()
 142         ip = '192.168.12.45'
 143         data = {ip: ['192.168.12.0/24', self._gen_subnet_id()]}
 144         ips = self._convert_to_ips(data)
 145
 146         mocks['ipam']._ipam_deallocate_ips(mock.ANY, mocks['driver'],
 147                                            mock.ANY, ips)
 148
 149         mocks['driver'].get_subnet.assert_called_once_with(data[ip][1])
 150         mocks['subnet'].deallocate.assert_called_once_with(ip)
 151
 152     def test_deallocate_multiple_ips(self):
 153         mocks = self._prepare_ipam()
 154         data = {'192.168.43.15': ['192.168.43.0/24', self._gen_subnet_id()],
 155                 '172.23.158.84': ['172.23.128.0/17', self._gen_subnet_id()],
 156                 '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]}
 157         ips = self._convert_to_ips(data)
 158
 159         mocks['ipam']._ipam_deallocate_ips(mock.ANY, mocks['driver'],
 160                                            mock.ANY, ips)
 161
 162         get_calls = [mock.call(data[ip][1]) for ip in data]
 163         mocks['driver'].get_subnet.assert_has_calls(get_calls, any_order=True)
 164
 165         ip_calls = [mock.call(ip) for ip in data]
 166         mocks['subnet'].deallocate.assert_has_calls(ip_calls, any_order=True)
 167
 168     def _single_ip_allocate_helper(self, mocks, ip, network, subnet):
 169         ips = [{'subnet_cidr': network,
 170                 'subnet_id': subnet}]
 171         if ip:
 172             ips[0]['ip_address'] = ip
 173
 174         allocated_ips = mocks['ipam']._ipam_allocate_ips(
 175             mock.ANY, mocks['driver'], mock.ANY, ips)
 176
 177         mocks['driver'].get_subnet.assert_called_once_with(subnet)
 178
 179         self.assertTrue(mocks['subnet'].allocate.called)
 180         request = mocks['subnet'].allocate.call_args[0][0]
 181
 182         return {'ips': allocated_ips,
 183                 'request': request}
 184
 185     def test_allocate_single_fixed_ip(self):
 186         mocks = self._prepare_ipam()
 187         ip = '192.168.15.123'
 188         mocks['subnet'].allocate.return_value = ip
 189
 190         results = self._single_ip_allocate_helper(mocks,
 191                                                   ip,
 192                                                   '192.168.15.0/24',
 193                                                   self._gen_subnet_id())
 194
 195         self.assertIsInstance(results['request'],
 196                               ipam_req.SpecificAddressRequest)
 197         self.assertEqual(netaddr.IPAddress(ip), results['request'].address)
 198
 199         self.assertEqual(ip, results['ips'][0]['ip_address'],
 200                          'Should allocate the same ip as passed')
 201
 202     def test_allocate_single_any_ip(self):
 203         mocks = self._prepare_ipam()
 204         network = '192.168.15.0/24'
 205         ip = '192.168.15.83'
 206         mocks['subnet'].allocate.return_value = ip
 207
 208         results = self._single_ip_allocate_helper(mocks, '', network,
 209                                                   self._gen_subnet_id())
 210
 211         self.assertIsInstance(results['request'], ipam_req.AnyAddressRequest)
 212         self.assertEqual(ip, results['ips'][0]['ip_address'])
 213
 214     def test_allocate_eui64_ip(self):
 215         mocks = self._prepare_ipam()
 216         ip = {'subnet_id': self._gen_subnet_id(),
 217               'subnet_cidr': '2001:470:abcd::/64',
 218               'mac': '6c:62:6d:de:cf:49',
 219               'eui64_address': True}
 220         eui64_ip = ipv6_utils.get_ipv6_addr_by_EUI64(ip['subnet_cidr'],
 221                                                      ip['mac'])
 222         mocks['ipam']._ipam_allocate_ips(mock.ANY, mocks['driver'],
 223                                          mock.ANY, [ip])
 224
 225         request = mocks['subnet'].allocate.call_args[0][0]
 226         self.assertIsInstance(request, ipam_req.AutomaticAddressRequest)
 227         self.assertEqual(eui64_ip, request.address)
 228
 229     def test_allocate_multiple_ips(self):
 230         mocks = self._prepare_ipam()
 231         data = {'': ['172.23.128.0/17', self._gen_subnet_id()],
 232                 '192.168.43.15': ['192.168.43.0/24', self._gen_subnet_id()],
 233                 '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]}
 234         ips = self._convert_to_ips(data)
 235         mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
 236             auto_ip='172.23.128.94')
 237
 238         mocks['ipam']._ipam_allocate_ips(
 239             mock.ANY, mocks['driver'], mock.ANY, ips)
 240         get_calls = [mock.call(data[ip][1]) for ip in data]
 241         mocks['driver'].get_subnet.assert_has_calls(get_calls, any_order=True)
 242
 243         self._validate_allocate_calls(ips, mocks)
 244
 245     def test_allocate_multiple_ips_with_exception(self):
 246         mocks = self._prepare_ipam()
 247
 248         auto_ip = '172.23.128.94'
 249         fail_ip = '192.168.43.15'
 250         data = {'': ['172.23.128.0/17', self._gen_subnet_id()],
 251                 fail_ip: ['192.168.43.0/24', self._gen_subnet_id()],
 252                 '8.8.8.8': ['8.0.0.0/8', self._gen_subnet_id()]}
 253         ips = self._convert_to_ips(data)
 254         mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
 255             auto_ip=auto_ip, fail_ip=fail_ip)
 256
 257         # Exception should be raised on attempt to allocate second ip.
 258         # Revert action should be performed for the already allocated ips,
 259         # In this test case only one ip should be deallocated
 260         # and original error should be reraised
 261         self.assertRaises(n_exc.InvalidInput,
 262                           mocks['ipam']._ipam_allocate_ips,
 263                           mock.ANY,
 264                           mocks['driver'],
 265                           mock.ANY,
 266                           ips)
 267
 268         # get_subnet should be called only for the first two networks
 269         get_calls = [mock.call(data[ip][1]) for ip in ['', fail_ip]]
 270         mocks['driver'].get_subnet.assert_has_calls(get_calls, any_order=True)
 271
 272         # Allocate should be called for the first two ips only
 273         self._validate_allocate_calls(ips[:-1], mocks)
 274         # Deallocate should be called for the first ip only
 275         mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
 276
 277     @mock.patch('neutron.ipam.driver.Pool')
 278     def test_create_subnet_over_ipam(self, pool_mock):
 279         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 280         cidr = '192.168.0.0/24'
 281         allocation_pools = [{'start': '192.168.0.2', 'end': '192.168.0.254'}]
 282         with self.subnet(allocation_pools=allocation_pools,
 283                          cidr=cidr):
 284             pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
 285             self.assertTrue(mocks['driver'].allocate_subnet.called)
 286             request = mocks['driver'].allocate_subnet.call_args[0][0]
 287             self.assertIsInstance(request, ipam_req.SpecificSubnetRequest)
 288             self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
 289
 290     @mock.patch('neutron.ipam.driver.Pool')
 291     def test_create_ipv6_pd_subnet_over_ipam(self, pool_mock):
 292         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 293         cfg.CONF.set_override('ipv6_pd_enabled', True)
 294         cidr = constants.PROVISIONAL_IPV6_PD_PREFIX
 295         allocation_pools = [netaddr.IPRange('::2', '::ffff:ffff:ffff:ffff')]
 296         with self.subnet(cidr=None, ip_version=6,
 297                          ipv6_ra_mode=constants.IPV6_SLAAC,
 298                          ipv6_address_mode=constants.IPV6_SLAAC):
 299             pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
 300             self.assertTrue(mocks['driver'].allocate_subnet.called)
 301             request = mocks['driver'].allocate_subnet.call_args[0][0]
 302             self.assertIsInstance(request, ipam_req.SpecificSubnetRequest)
 303             self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
 304             self.assertEqual(allocation_pools, request.allocation_pools)
 305
 306     @mock.patch('neutron.ipam.driver.Pool')
 307     def test_create_subnet_over_ipam_with_rollback(self, pool_mock):
 308         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 309         mocks['driver'].allocate_subnet.side_effect = ValueError
 310         cidr = '10.0.2.0/24'
 311         with self.network() as network:
 312             self._create_subnet(self.fmt, network['network']['id'],
 313                                 cidr, expected_res_status=500)
 314
 315             pool_mock.get_instance.assert_called_once_with(None, mock.ANY)
 316             self.assertTrue(mocks['driver'].allocate_subnet.called)
 317             request = mocks['driver'].allocate_subnet.call_args[0][0]
 318             self.assertIsInstance(request, ipam_req.SpecificSubnetRequest)
 319             self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
 320             # Verify no subnet was created for network
 321             req = self.new_show_request('networks', network['network']['id'])
 322             res = req.get_response(self.api)
 323             net = self.deserialize(self.fmt, res)
 324             self.assertEqual(0, len(net['network']['subnets']))
 325
 326     @mock.patch('neutron.ipam.driver.Pool')
 327     def test_ipam_subnet_deallocated_if_create_fails(self, pool_mock):
 328         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 329         cidr = '10.0.2.0/24'
 330         with mock.patch.object(
 331                 ipam_backend_mixin.IpamBackendMixin, '_save_subnet',
 332                 side_effect=ValueError), self.network() as network:
 333             self._create_subnet(self.fmt, network['network']['id'],
 334                                 cidr, expected_res_status=500)
 335             pool_mock.get_instance.assert_any_call(None, mock.ANY)
 336             self.assertEqual(2, pool_mock.get_instance.call_count)
 337             self.assertTrue(mocks['driver'].allocate_subnet.called)
 338             request = mocks['driver'].allocate_subnet.call_args[0][0]
 339             self.assertIsInstance(request, ipam_req.SpecificSubnetRequest)
 340             self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
 341             # Verify remove ipam subnet was called
 342             mocks['driver'].remove_subnet.assert_called_once_with(
 343                 self.subnet_id)
 344
 345     @mock.patch('neutron.ipam.driver.Pool')
 346     def test_update_subnet_over_ipam(self, pool_mock):
 347         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 348         cidr = '10.0.0.0/24'
 349         allocation_pools = [{'start': '10.0.0.2', 'end': '10.0.0.254'}]
 350         with self.subnet(allocation_pools=allocation_pools,
 351                          cidr=cidr) as subnet:
 352             data = {'subnet': {'allocation_pools': [
 353                     {'start': '10.0.0.10', 'end': '10.0.0.20'},
 354                     {'start': '10.0.0.30', 'end': '10.0.0.40'}]}}
 355             req = self.new_update_request('subnets', data,
 356                                           subnet['subnet']['id'])
 357             res = req.get_response(self.api)
 358             self.assertEqual(200, res.status_code)
 359
 360             pool_mock.get_instance.assert_any_call(None, mock.ANY)
 361             self.assertEqual(2, pool_mock.get_instance.call_count)
 362             self.assertTrue(mocks['driver'].update_subnet.called)
 363             request = mocks['driver'].update_subnet.call_args[0][0]
 364             self.assertIsInstance(request, ipam_req.SpecificSubnetRequest)
 365             self.assertEqual(netaddr.IPNetwork(cidr), request.subnet_cidr)
 366
 367             ip_ranges = [netaddr.IPRange(p['start'],
 368                 p['end']) for p in data['subnet']['allocation_pools']]
 369             self.assertEqual(ip_ranges, request.allocation_pools)
 370
 371     @mock.patch('neutron.ipam.driver.Pool')
 372     def test_delete_subnet_over_ipam(self, pool_mock):
 373         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 374         gateway_ip = '10.0.0.1'
 375         cidr = '10.0.0.0/24'
 376         res = self._create_network(fmt=self.fmt, name='net',
 377                                    admin_state_up=True)
 378         network = self.deserialize(self.fmt, res)
 379         subnet = self._make_subnet(self.fmt, network, gateway_ip,
 380                                    cidr, ip_version=4)
 381         req = self.new_delete_request('subnets', subnet['subnet']['id'])
 382         res = req.get_response(self.api)
 383         self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
 384
 385         pool_mock.get_instance.assert_any_call(None, mock.ANY)
 386         self.assertEqual(2, pool_mock.get_instance.call_count)
 387         mocks['driver'].remove_subnet.assert_called_once_with(
 388             subnet['subnet']['id'])
 389
 390     @mock.patch('neutron.ipam.driver.Pool')
 391     def test_delete_subnet_over_ipam_with_rollback(self, pool_mock):
 392         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 393         mocks['driver'].remove_subnet.side_effect = ValueError
 394         gateway_ip = '10.0.0.1'
 395         cidr = '10.0.0.0/24'
 396         res = self._create_network(fmt=self.fmt, name='net',
 397                                    admin_state_up=True)
 398         network = self.deserialize(self.fmt, res)
 399         subnet = self._make_subnet(self.fmt, network, gateway_ip,
 400                                    cidr, ip_version=4)
 401         req = self.new_delete_request('subnets', subnet['subnet']['id'])
 402         res = req.get_response(self.api)
 403         self.assertEqual(webob.exc.HTTPServerError.code, res.status_int)
 404
 405         pool_mock.get_instance.assert_any_call(None, mock.ANY)
 406         self.assertEqual(2, pool_mock.get_instance.call_count)
 407         mocks['driver'].remove_subnet.assert_called_once_with(
 408             subnet['subnet']['id'])
 409         # Verify subnet was recreated after failed ipam call
 410         subnet_req = self.new_show_request('subnets',
 411                                            subnet['subnet']['id'])
 412         raw_res = subnet_req.get_response(self.api)
 413         sub_res = self.deserialize(self.fmt, raw_res)
 414         self.assertIn(sub_res['subnet']['cidr'], cidr)
 415         self.assertIn(sub_res['subnet']['gateway_ip'],
 416                       gateway_ip)
 417
 418     @mock.patch('neutron.ipam.driver.Pool')
 419     def test_create_port_ipam(self, pool_mock):
 420         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 421         auto_ip = '10.0.0.2'
 422         expected_calls = [{'ip_address': ''}]
 423         mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
 424             auto_ip=auto_ip)
 425         with self.subnet() as subnet:
 426             with self.port(subnet=subnet) as port:
 427                 ips = port['port']['fixed_ips']
 428                 self.assertEqual(1, len(ips))
 429                 self.assertEqual(ips[0]['ip_address'], auto_ip)
 430                 self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
 431                 self._validate_allocate_calls(expected_calls, mocks)
 432
 433     @mock.patch('neutron.ipam.driver.Pool')
 434     def test_create_port_ipam_with_rollback(self, pool_mock):
 435         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 436         mocks['subnet'].allocate.side_effect = ValueError
 437         with self.network() as network:
 438             with self.subnet(network=network):
 439                 net_id = network['network']['id']
 440                 data = {
 441                     'port': {'network_id': net_id,
 442                              'tenant_id': network['network']['tenant_id']}}
 443                 port_req = self.new_create_request('ports', data)
 444                 res = port_req.get_response(self.api)
 445                 self.assertEqual(webob.exc.HTTPServerError.code,
 446                                  res.status_int)
 447
 448                 # verify no port left after failure
 449                 req = self.new_list_request('ports', self.fmt,
 450                                             "network_id=%s" % net_id)
 451                 res = self.deserialize(self.fmt, req.get_response(self.api))
 452                 self.assertEqual(0, len(res['ports']))
 453
 454     @mock.patch('neutron.ipam.driver.Pool')
 455     def test_update_port_ipam(self, pool_mock):
 456         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 457         auto_ip = '10.0.0.2'
 458         new_ip = '10.0.0.15'
 459         expected_calls = [{'ip_address': ip} for ip in ['', new_ip]]
 460         mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
 461             auto_ip=auto_ip)
 462         with self.subnet() as subnet:
 463             with self.port(subnet=subnet) as port:
 464                 ips = port['port']['fixed_ips']
 465                 self.assertEqual(1, len(ips))
 466                 self.assertEqual(ips[0]['ip_address'], auto_ip)
 467                 # Update port with another new ip
 468                 data = {"port": {"fixed_ips": [{
 469                         'subnet_id': subnet['subnet']['id'],
 470                         'ip_address': new_ip}]}}
 471                 req = self.new_update_request('ports', data,
 472                                               port['port']['id'])
 473                 res = self.deserialize(self.fmt, req.get_response(self.api))
 474                 ips = res['port']['fixed_ips']
 475                 self.assertEqual(1, len(ips))
 476                 self.assertEqual(new_ip, ips[0]['ip_address'])
 477
 478                 # Allocate should be called for the first two networks
 479                 self._validate_allocate_calls(expected_calls, mocks)
 480                 # Deallocate should be called for the first ip only
 481                 mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
 482
 483     @mock.patch('neutron.ipam.driver.Pool')
 484     def test_delete_port_ipam(self, pool_mock):
 485         mocks = self._prepare_mocks_with_pool_mock(pool_mock)
 486         auto_ip = '10.0.0.2'
 487         mocks['subnet'].allocate.side_effect = self._get_allocate_mock(
 488             auto_ip=auto_ip)
 489         with self.subnet() as subnet:
 490             with self.port(subnet=subnet) as port:
 491                 ips = port['port']['fixed_ips']
 492                 self.assertEqual(1, len(ips))
 493                 self.assertEqual(ips[0]['ip_address'], auto_ip)
 494                 req = self.new_delete_request('ports', port['port']['id'])
 495                 res = req.get_response(self.api)
 496
 497                 self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
 498                 mocks['subnet'].deallocate.assert_called_once_with(auto_ip)
 499
 500     def test_recreate_port_ipam(self):
 501         ip = '10.0.0.2'
 502         with self.subnet() as subnet:
 503             with self.port(subnet=subnet) as port:
 504                 ips = port['port']['fixed_ips']
 505                 self.assertEqual(1, len(ips))
 506                 self.assertEqual(ips[0]['ip_address'], ip)
 507                 req = self.new_delete_request('ports', port['port']['id'])
 508                 res = req.get_response(self.api)
 509                 self.assertEqual(webob.exc.HTTPNoContent.code, res.status_int)
 510                 with self.port(subnet=subnet, fixed_ips=ips) as port:
 511                     ips = port['port']['fixed_ips']
 512                     self.assertEqual(1, len(ips))
 513                     self.assertEqual(ips[0]['ip_address'], ip)
 514
 515     @mock.patch('neutron.ipam.driver.Pool')
 516     def test_update_ips_for_port_passes_port_dict_to_factory(self, pool_mock):
 517         address_factory = mock.Mock()
 518         mocks = self._prepare_mocks_with_pool_mock(
 519             pool_mock, address_factory=address_factory)
 520         context = mock.Mock()
 521         new_ips = mock.Mock()
 522         original_ips = mock.Mock()
 523         mac = mock.Mock()
 524
 525         ip_dict = {'ip_address': '192.1.1.10',
 526                    'subnet_id': uuidutils.generate_uuid()}
 527         changes = ipam_pluggable_backend.IpamPluggableBackend.Changes(
 528             add=[ip_dict], original=[], remove=[])
 529         changes_mock = mock.Mock(return_value=changes)
 530         fixed_ips_mock = mock.Mock(return_value=changes.add)
 531         mocks['ipam'] = ipam_pluggable_backend.IpamPluggableBackend()
 532         mocks['ipam']._get_changed_ips_for_port = changes_mock
 533         mocks['ipam']._test_fixed_ips_for_port = fixed_ips_mock
 534
 535         port_dict = {'device_owner': uuidutils.generate_uuid(),
 536                      'network_id': uuidutils.generate_uuid()}
 537
 538         mocks['ipam']._update_ips_for_port(context, port_dict,
 539                                            original_ips, new_ips, mac)
 540         mocks['driver'].get_address_request_factory.assert_called_once_with()
 541         # Validate port_dict is passed into address_factory
 542         address_factory.get_request.assert_called_once_with(context,
 543                                                             port_dict,
 544                                                             ip_dict)