Prohibit deletion of ports currently in use by a trunk

A port that is currently in use as a trunk port or a subport cannot be
deleted. The trunk or subport that the port is owned by must be deleted
first.

Partially-implements: blueprint vlan-aware-vms
Change-Id: I35aec716b44b85a686c9d0f8f591d7d1ec21a794

neutron/services/trunk/exceptions.py

@@ -41,6 +41,16 @@ class ParentPortInUse(n_exc.InUse):
                 "eligible for use as a parent port.")
 
 
+class PortInUseAsTrunkParent(n_exc.InUse):
+    message = _("Port %(port_id)s is currently a parent port "
+                "for trunk %(trunk_id)s.")
+
+
+class PortInUseAsSubPort(n_exc.InUse):
+    message = _("Port %(port_id)s is currently a subport for "
+                "trunk %(trunk_id)s.")
+
+
 class TrunkInUse(n_exc.InUse):
     message = _("Trunk %(trunk_id)s is currently in use.")
 

neutron/services/trunk/plugin.py

@@ -18,6 +18,7 @@ from oslo_utils import uuidutils
 from neutron.api.v2 import attributes
 from neutron.callbacks import events
 from neutron.callbacks import registry
+from neutron.callbacks import resources
 from neutron.db import api as db_api
 from neutron.db import common_db_mixin
 from neutron.db import db_base_plugin_common
@@ -58,6 +59,8 @@ class TrunkPlugin(service_base.ServicePluginBase,
         db_base_plugin_v2.NeutronDbPluginV2.register_dict_extend_funcs(
             attributes.PORTS, [_extend_port_trunk_details])
         self._segmentation_types = {}
+        registry.subscribe(rules.enforce_port_deletion_rules,
+                           resources.PORT, events.BEFORE_DELETE)
         registry.notify(constants.TRUNK_PLUGIN, events.AFTER_INIT, self)
         LOG.debug('Trunk plugin loaded')
 

neutron/services/trunk/rules.py

@@ -33,6 +33,24 @@ def trunk_can_be_managed(context, trunk):
         raise trunk_exc.TrunkDisabled(trunk_id=trunk.id)
 
 
+def enforce_port_deletion_rules(resource, event, trigger, **kwargs):
+    """Prohibit the deletion of a port that's used in a trunk."""
+    # NOTE: the ML2 plugin properly catches these exceptions when raised, but
+    # non-ML2 plugins might not. To address this we should move the callback
+    # registry notification emitted in the ML2 plugin's delete_port() higher
+    # up in the plugin hierarchy.
+    context = kwargs['context']
+    port_id = kwargs['port_id']
+    subport_obj = trunk_objects.SubPort.get_object(context, port_id=port_id)
+    if subport_obj:
+        raise trunk_exc.PortInUseAsSubPort(port_id=port_id,
+                                           trunk_id=subport_obj.trunk_id)
+    trunk_obj = trunk_objects.Trunk.get_object(context, port_id=port_id)
+    if trunk_obj:
+        raise trunk_exc.PortInUseAsTrunkParent(port_id=port_id,
+                                               trunk_id=trunk_obj.id)
+
+
 class TrunkPortValidator(object):
 
     def __init__(self, port_id):

neutron/tests/tempest/api/test_trunk_negative.py

@@ -216,3 +216,22 @@ class TrunkTestJSON(test_trunk.TrunkTestJSONBase):
         trunk = self._create_trunk_with_network_and_parent([])
         self.assertRaises(lib_exc.NotFound, self.client.remove_subports,
                           trunk['trunk']['id'], [subport_data])
+
+    @test.attr(type='negative')
+    @test.idempotent_id('6c9c5126-4f61-11e6-8248-40a8f063c891')
+    def test_delete_port_in_use_by_trunk(self):
+        trunk = self._create_trunk_with_network_and_parent(None)
+        self.assertRaises(lib_exc.Conflict, self.client.delete_port,
+                          trunk['trunk']['port_id'])
+
+    @test.attr(type='negative')
+    @test.idempotent_id('343a03d0-4f7c-11e6-97fa-40a8f063c891')
+    def test_delete_port_in_use_by_subport(self):
+        network = self.create_network()
+        port = self.create_port(network)
+        subports = [{'port_id': port['id'],
+                     'segmentation_type': 'vlan',
+                     'segmentation_id': 2}]
+        self._create_trunk_with_network_and_parent(subports)
+        self.assertRaises(lib_exc.Conflict, self.client.delete_port,
+                          port['id'])

neutron/tests/unit/services/trunk/test_plugin.py

@@ -58,6 +58,26 @@ class TrunkPluginTestCase(test_plugin.Ml2PluginV2TestCase):
             self.context, port_id=port_id)
         return subports[0]
 
+    def _test_delete_port_raise_in_use(self, parent_port, child_port, port_id,
+                                       exception):
+        subport = create_subport_dict(child_port['port']['id'])
+        self._create_test_trunk(parent_port, [subport])
+        core_plugin = manager.NeutronManager.get_plugin()
+        self.assertRaises(exception, core_plugin.delete_port,
+                          self.context, port_id)
+
+    def test_delete_port_raise_in_use_by_trunk(self):
+        with self.port() as parent_port, self.port() as child_port:
+            self._test_delete_port_raise_in_use(
+                parent_port, child_port, parent_port['port']['id'],
+                trunk_exc.PortInUseAsTrunkParent)
+
+    def test_delete_port_raise_in_use_by_subport(self):
+        with self.port() as parent_port, self.port() as child_port:
+            self._test_delete_port_raise_in_use(
+                parent_port, child_port, child_port['port']['id'],
+                trunk_exc.PortInUseAsSubPort)
+
     def test_delete_trunk_raise_in_use(self):
         with self.port() as port:
             trunk = self._create_test_trunk(port)