Handle auto-address subnets on port update

The patch handles the case when a port is updated
with an IP from a new IPv6 auto-address subnet

See bug for details on how the issue can affect DHCP
functionality.

Closes-Bug: #1678104
Change-Id: If2473f2db3ca16b5f46d3280e79a49756d1c098a

neutron/db/ipam_pluggable_backend.py

@@ -221,19 +221,28 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
                     ips.append([{'subnet_id': s['id']}
                                 for s in subnets])
 
-        is_router_port = (
+        ips.extend(self._get_auto_address_ips(v6_stateless, p))
-            p['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
+
-        if not is_router_port:
-            for subnet in v6_stateless:
-                # IP addresses for IPv6 SLAAC and DHCPv6-stateless subnets
-                # are implicitly included.
-                ips.append({'subnet_id': subnet['id'],
-                            'subnet_cidr': subnet['cidr'],
-                            'eui64_address': True,
-                            'mac': p['mac_address']})
         ipam_driver = driver.Pool.get_instance(None, context)
         return self._ipam_allocate_ips(context, ipam_driver, p, ips)
 
+    def _get_auto_address_ips(self, v6_stateless_subnets, port,
+                              exclude_subnet_ids=None):
+        exclude_subnet_ids = exclude_subnet_ids or []
+        ips = []
+        is_router_port = (
+            port['device_owner'] in constants.ROUTER_INTERFACE_OWNERS_SNAT)
+        if not is_router_port:
+            for subnet in v6_stateless_subnets:
+                if subnet['id'] not in exclude_subnet_ids:
+                    # IP addresses for IPv6 SLAAC and DHCPv6-stateless subnets
+                    # are implicitly included.
+                    ips.append({'subnet_id': subnet['id'],
+                                'subnet_cidr': subnet['cidr'],
+                                'eui64_address': True,
+                                'mac': port['mac_address']})
+        return ips
+
     def _test_fixed_ips_for_port(self, context, network_id, fixed_ips,
                                  device_owner, subnets):
         """Test fixed IPs for port.
@@ -297,6 +306,14 @@ class IpamPluggableBackend(ipam_backend_mixin.IpamBackendMixin):
         if changes.remove:
             removed = self._ipam_deallocate_ips(context, ipam_driver, port,
                                                 changes.remove)
+
+        v6_stateless = self._classify_subnets(
+            context, subnets)[2]
+        handled_subnet_ids = [ip['subnet_id'] for ip in
+                              to_add + changes.original + changes.remove]
+        to_add.extend(self._get_auto_address_ips(
+            v6_stateless, port, handled_subnet_ids))
+
         if to_add:
             added = self._ipam_allocate_ips(context, ipam_driver,
                                             port, to_add)

neutron/tests/unit/db/test_db_base_plugin_v2.py

@@ -2178,6 +2178,44 @@ fixed_ips=ip_address%%3D%s&fixed_ips=ip_address%%3D%s&fixed_ips=subnet_id%%3D%s
         self.assertEqual(self._calc_ipv6_addr_by_EUI64(port, subnet_v6),
                          ips[0]['ip_address'])
 
+    def test_update_port_with_new_ipv6_slaac_subnet_in_fixed_ips(self):
+        """Test port update with a new IPv6 SLAAC subnet in fixed IPs."""
+        res = self._create_network(fmt=self.fmt, name='net',
+                                   admin_state_up=True)
+        network = self.deserialize(self.fmt, res)
+        # Create a port using an IPv4 subnet and an IPv6 SLAAC subnet
+        subnet_v4 = self._make_subnet(self.fmt, network, gateway='10.0.0.1',
+                                      cidr='10.0.0.0/24', ip_version=4)
+        subnet_v6 = self._make_v6_subnet(network, constants.IPV6_SLAAC)
+        res = self._create_port(self.fmt, net_id=network['network']['id'])
+        port = self.deserialize(self.fmt, res)
+        self.assertEqual(2, len(port['port']['fixed_ips']))
+        # Update port to have only IPv4 address
+        ips = [{'subnet_id': subnet_v4['subnet']['id']},
+               {'subnet_id': subnet_v6['subnet']['id'],
+                'delete_subnet': True}]
+        data = {'port': {'fixed_ips': ips}}
+        req = self.new_update_request('ports', data,
+                                      port['port']['id'])
+        res = self.deserialize(self.fmt, req.get_response(self.api))
+        # Port should only have an address corresponding to IPv4 subnet
+        ips = res['port']['fixed_ips']
+        self.assertEqual(1, len(ips))
+        self.assertEqual(subnet_v4['subnet']['id'], ips[0]['subnet_id'])
+        # Now update port and request an additional address on the IPv6 SLAAC
+        # subnet.
+        ips.append({'subnet_id': subnet_v6['subnet']['id']})
+        data = {'port': {'fixed_ips': ips}}
+        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']
+        # Port should have IPs on both IPv4 and IPv6 subnets
+        self.assertEqual(2, len(ips))
+        self.assertEqual(set([subnet_v4['subnet']['id'],
+                              subnet_v6['subnet']['id']]),
+                         set([ip['subnet_id'] for ip in ips]))
+
     def test_update_port_excluding_ipv6_slaac_subnet_from_fixed_ips(self):
         """Test port update excluding IPv6 SLAAC subnet from fixed ips."""
         res = self._create_network(fmt=self.fmt, name='net',

neutron/tests/unit/db/test_ipam_pluggable_backend.py

@@ -648,7 +648,7 @@ class TestDbBasePluginIpam(test_db_base.NeutronDbPluginV2TestCase):
         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()
+        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=[])