diff --git a/neutron/db/ipam_pluggable_backend.py b/neutron/db/ipam_pluggable_backend.py index 1c7417921fe..dc0dcb15a5a 100644 --- a/neutron/db/ipam_pluggable_backend.py +++ b/neutron/db/ipam_pluggable_backend.py @@ -283,7 +283,8 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin): context, original_ips, new_ips, port['device_owner']) try: subnets = self._ipam_get_subnets( - context, network_id=port['network_id'], host=host) + context, network_id=port['network_id'], host=host, + service_type=port.get('device_owner')) except ipam_exc.DeferIpam: subnets = [] diff --git a/neutron/tests/unit/db/test_ipam_pluggable_backend.py b/neutron/tests/unit/db/test_ipam_pluggable_backend.py index 70efd58e87d..4100e3e98fb 100644 --- a/neutron/tests/unit/db/test_ipam_pluggable_backend.py +++ b/neutron/tests/unit/db/test_ipam_pluggable_backend.py @@ -659,7 +659,8 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase): 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) + 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, diff --git a/neutron/tests/unit/extensions/test_subnet_service_types.py b/neutron/tests/unit/extensions/test_subnet_service_types.py index bd0d7d60084..a13fdaeade9 100644 --- a/neutron/tests/unit/extensions/test_subnet_service_types.py +++ b/neutron/tests/unit/extensions/test_subnet_service_types.py @@ -296,6 +296,31 @@ class SubnetServiceTypesExtensionTestCase( def test_create_dhcp_port_compute_subnet_no_dhcp(self): self.test_create_dhcp_port_compute_subnet(enable_dhcp=False) + def test_update_port_fixed_ips(self): + with self.network() as network: + pass + service_type = 'compute:foo' + # Create a subnet with a service_type + res = self._create_service_subnet([service_type], + cidr=self.CIDRS[1], + network=network) + service_subnet = self.deserialize('json', res)['subnet'] + # Create a port with a matching device owner + network = network['network'] + port = self._create_port(self.fmt, + net_id=network['id'], + tenant_id=network['tenant_id'], + device_owner=service_type) + port = self.deserialize('json', port)['port'] + # Update the port's fixed_ips. It's ok to reuse the same IP it already + # has. + ip_address = port['fixed_ips'][0]['ip_address'] + data = {'port': {'fixed_ips': [{'subnet_id': service_subnet['id'], + 'ip_address': ip_address}]}} + # self._update will fail with a MismatchError if the update cannot be + # applied + port = self._update('ports', port['id'], data) + class SubnetServiceTypesExtensionTestCasev6( SubnetServiceTypesExtensionTestCase):