Merge "Emit registry events on subport addition/removal"

neutron/services/trunk/constants.py

@@ -13,7 +13,7 @@
 #    under the License.
 
 PARENT_PORT = 'parent_port'
-SUBPORT = 'subport'
+SUBPORTS = 'subports'
 TRUNK = 'trunk'
 TRUNK_PLUGIN = 'trunk_plugin'
 

neutron/services/trunk/plugin.py

@@ -126,6 +126,7 @@ class TrunkPlugin(service_base.ServicePluginBase,
         subports_validator = rules.SubPortsValidator(
             self._segmentation_types, subports)
         subports = subports_validator.validate(context, basic_validation=True)
+        added_subports = []
 
         with db_api.autonested_transaction(context.session):
             for subport in subports:
@@ -137,7 +138,11 @@ class TrunkPlugin(service_base.ServicePluginBase,
                                segmentation_id=subport['segmentation_id'])
                 obj.create()
                 trunk['sub_ports'].append(obj)
+                added_subports.append(obj)
 
+        registry.notify(
+            constants.SUBPORTS, events.AFTER_CREATE, self,
+            added_subports=added_subports)
         return trunk
 
     @db_base_plugin_common.convert_result_to_dict
@@ -158,6 +163,7 @@ class TrunkPlugin(service_base.ServicePluginBase,
                 trunk_validation=False)
 
             current_subports = {p.port_id: p for p in trunk.sub_ports}
+            removed_subports = []
 
             for subport in subports:
                 subport_obj = current_subports.pop(subport['port_id'], None)
@@ -166,8 +172,12 @@ class TrunkPlugin(service_base.ServicePluginBase,
                     raise trunk_exc.SubPortNotFound(trunk_id=trunk_id,
                                                     port_id=subport['port_id'])
                 subport_obj.delete()
+                removed_subports.append(subport_obj)
 
             trunk.sub_ports = list(current_subports.values())
+            registry.notify(
+                constants.SUBPORTS, events.AFTER_DELETE, self,
+                removed_subports=removed_subports)
             return trunk
 
     @db_base_plugin_common.filter_fields

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

@@ -13,28 +13,79 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import mock
+
+from neutron.callbacks import events
+from neutron.callbacks import registry
 from neutron import manager
+from neutron.objects import trunk as trunk_objects
+from neutron.services.trunk import constants
 from neutron.services.trunk import exceptions as trunk_exc
 from neutron.services.trunk import plugin as trunk_plugin
 from neutron.tests.unit.plugins.ml2 import test_plugin
 
 
+def create_subport_dict(port_id):
+    return {'segmentation_type': 'vlan',
+            'segmentation_id': 123,
+            'port_id': port_id}
+
+
+def register_mock_callback(resource, event):
+    callback = mock.Mock()
+    registry.subscribe(callback, resource, event)
+    return callback
+
+
 class TrunkPluginTestCase(test_plugin.Ml2PluginV2TestCase):
 
     def setUp(self):
         super(TrunkPluginTestCase, self).setUp()
         self.trunk_plugin = trunk_plugin.TrunkPlugin()
+        self.trunk_plugin.add_segmentation_type('vlan', lambda x: True)
+
+    def _create_test_trunk(self, port, subports=None):
+        subports = subports if subports else []
+        trunk = {'port_id': port['port']['id'],
+                 'tenant_id': 'test_tenant',
+                 'sub_ports': subports}
+        response = (
+            self.trunk_plugin.create_trunk(self.context, {'trunk': trunk}))
+        return response
+
+    def _get_subport_obj(self, port_id):
+        subports = trunk_objects.SubPort.get_objects(
+            self.context, port_id=port_id)
+        return subports[0]
 
     def test_delete_trunk_raise_in_use(self):
         with self.port() as port:
-            trunk = {'port_id': port['port']['id'],
-                     'tenant_id': 'test_tenant',
-                     'sub_ports': []}
-            response = (
-                self.trunk_plugin.create_trunk(self.context, {'trunk': trunk}))
+            trunk = self._create_test_trunk(port)
             core_plugin = manager.NeutronManager.get_plugin()
             port['port']['binding:host_id'] = 'host'
             core_plugin.update_port(self.context, port['port']['id'], port)
             self.assertRaises(trunk_exc.TrunkInUse,
                               self.trunk_plugin.delete_trunk,
-                              self.context, response['id'])
+                              self.context, trunk['id'])
+
+    def _test_subports_action_notify(self, event, payload_key):
+        with self.port() as parent_port, self.port() as child_port:
+            trunk = self._create_test_trunk(parent_port)
+            subport = create_subport_dict(child_port['port']['id'])
+            callback = register_mock_callback(constants.SUBPORTS, event)
+            self.trunk_plugin.add_subports(
+                self.context, trunk['id'], [subport])
+            subport_obj = self._get_subport_obj(subport['port_id'])
+            self.trunk_plugin.remove_subports(
+                self.context, trunk['id'], [subport])
+            payload = {payload_key: [subport_obj]}
+            callback.assert_called_once_with(
+                constants.SUBPORTS, event, self.trunk_plugin, **payload)
+
+    def test_add_subports_notify(self):
+        self._test_subports_action_notify(events.AFTER_CREATE,
+                                          'added_subports')
+
+    def test_remove_subports_notify(self):
+        self._test_subports_action_notify(events.AFTER_DELETE,
+                                          'removed_subports')