Merge "L3: enable plugin to decide if subnet is mandatory"

neutron/db/l3_db.py

@@ -391,6 +391,16 @@ class L3_NAT_dbonly_mixin(l3.RouterPluginBase):
         if new_valid_gw_port_attachment:
             subnets = self._core_plugin.get_subnets_by_network(context,
                                                                new_network_id)
+            try:
+                kwargs = {'context': context, 'router_id': router_id,
+                          'network_id': new_network_id, 'subnets': subnets}
+                registry.notify(
+                    resources.ROUTER_GATEWAY, events.BEFORE_CREATE, self,
+                    **kwargs)
+            except exceptions.CallbackFailure as e:
+                # raise the underlying exception
+                raise e.errors[0].error
+
             for subnet in subnets:
                 self._check_for_dup_router_subnet(context, router,
                                                   new_network_id, subnet['id'],

neutron/tests/unit/extensions/test_l3.py

@@ -1617,6 +1617,30 @@ class L3NatTestCaseBase(L3NatTestCaseMixin):
                 gw_info = body['router']['external_gateway_info']
                 self.assertIsNone(gw_info)
 
+    def test_router_add_gateway_no_subnet_forbidden(self):
+        with self.router() as r:
+            with self.network() as n:
+                self._set_net_external(n['network']['id'])
+                with mock.patch.object(registry, 'notify') as notify:
+                    errors = [
+                        exceptions.NotificationError(
+                            'foo_callback_id',
+                            n_exc.InvalidInput(error_message='forbidden')),
+                    ]
+                    notify.side_effect = exceptions.CallbackFailure(
+                        errors=errors)
+                    self._add_external_gateway_to_router(
+                        r['router']['id'], n['network']['id'],
+                        expected_code=exc.HTTPBadRequest.code)
+                    notify.assert_called_once_with(
+                        resources.ROUTER_GATEWAY,
+                        events.BEFORE_CREATE,
+                        mock.ANY,
+                        context=mock.ANY,
+                        router_id=r['router']['id'],
+                        network_id=n['network']['id'],
+                        subnets=[])
+
     def test_router_remove_interface_inuse_returns_409(self):
         with self.router() as r:
             with self.subnet() as s: