Merge "Reset MAC on unbinding direct-physical port" into stable/rocky

neutron/plugins/ml2/plugin.py

@@ -340,6 +340,18 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
                                 new_mac=port['mac_address'])
         return mac_change
 
+    def _reset_mac_for_direct_physical(self, orig_port, port, binding):
+        # when unbinding direct-physical port we need to free
+        # physical device MAC address so that other ports may reuse it
+        if (binding.vnic_type == portbindings.VNIC_DIRECT_PHYSICAL and
+                port.get('device_id') == '' and
+                port.get('device_owner') == '' and
+                orig_port['device_id'] != ''):
+            port['mac_address'] = self._generate_mac()
+            return True
+        else:
+            return False
+
     @registry.receives(resources.AGENT, [events.AFTER_UPDATE])
     def _retry_binding_revived_agents(self, resource, event, trigger,
                                       **kwargs):
@@ -1392,6 +1404,8 @@ class Ml2Plugin(db_base_plugin_v2.NeutronDbPluginV2,
                 raise exc.PortNotFound(port_id=id)
             mac_address_updated = self._check_mac_update_allowed(
                 port_db, attrs, binding)
+            mac_address_updated |= self._reset_mac_for_direct_physical(
+                port_db, attrs, binding)
             need_port_update_notify |= mac_address_updated
             original_port = self._make_port_dict(port_db)
             updated_port = super(Ml2Plugin, self).update_port(context, id,

neutron/tests/unit/plugins/ml2/test_plugin.py

@@ -1633,6 +1633,37 @@ class TestMl2PluginOnly(Ml2PluginV2TestCase):
         with testtools.ExpectedException(exc.PortBound):
             self._test_check_mac_update_allowed(portbindings.VIF_TYPE_OVS)
 
+    def _test_reset_mac_for_direct_physical(self, direct_physical=True,
+                                            unbinding=True):
+        plugin = directory.get_plugin()
+        port = {'device_id': '123', 'device_owner': 'compute:nova'}
+        new_attrs = ({'device_id': '', 'device_owner': ''} if unbinding else
+            {'name': 'new'})
+        binding = mock.Mock()
+        binding.vnic_type = (
+            portbindings.VNIC_DIRECT_PHYSICAL if direct_physical else
+            portbindings.VNIC_NORMAL)
+        new_mac = plugin._reset_mac_for_direct_physical(
+            port, new_attrs, binding)
+        if direct_physical and unbinding:
+            self.assertTrue(new_mac)
+            self.assertIsNotNone(new_attrs.get('mac_address'))
+        else:
+            self.assertFalse(new_mac)
+            self.assertIsNone(new_attrs.get('mac_address'))
+
+    def test_reset_mac_for_direct_physical(self):
+        self._test_reset_mac_for_direct_physical()
+
+    def test_reset_mac_for_direct_physical_not_physycal(self):
+        self._test_reset_mac_for_direct_physical(False, True)
+
+    def test_reset_mac_for_direct_physical_no_unbinding(self):
+        self._test_reset_mac_for_direct_physical(True, False)
+
+    def test_reset_mac_for_direct_physical_no_unbinding_not_physical(self):
+        self._test_reset_mac_for_direct_physical(False, False)
+
     def test__device_to_port_id_prefix_names(self):
         input_output = [('sg-abcdefg', 'abcdefg'),
                         ('tap123456', '123456'),