Fix port update for service subnets

Updating a port's fixed_ips will currently fail for a subnet that has
a service type, regardless of whether the port has a matching
device_owner. This is because _update_ips_for_port in
ipam_pluggable_backend.py does not take the device_owner into account
when calling _ipam_get_subnets.

This patch fixes the bug by passing the port's device_owner to
_ipam_get_subnets in _update_ips_for_port, and includes a new unit
test to prevent future regressions.

Change-Id: I09ac773ae27eebe104d5c39e1e8863c02880f85d
Closes-Bug: #1660607
(cherry picked from commit b61b630f0b)

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 = []
 

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,

neutron/tests/unit/extensions/test_subnet_service_types.py

@@ -293,6 +293,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):