neutron/tests/unit/extensions/test_external_net.py

   1 # Copyright (c) 2013 OpenStack Foundation.
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17 from oslo_utils import uuidutils
  18 import testtools
  19 from webob import exc
  20
  21 from neutron import context
  22 from neutron.db import models_v2
  23 from neutron.extensions import external_net as external_net
  24 from neutron import manager
  25 from neutron.tests.unit.api.v2 import test_base
  26 from neutron.tests.unit.db import test_db_base_plugin_v2
  27
  28
  29 _uuid = uuidutils.generate_uuid
  30 _get_path = test_base._get_path
  31
  32
  33 class ExtNetTestExtensionManager(object):
  34
  35     def get_resources(self):
  36         return []
  37
  38     def get_actions(self):
  39         return []
  40
  41     def get_request_extensions(self):
  42         return []
  43
  44
  45 class ExtNetDBTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
  46
  47     def _create_network(self, fmt, name, admin_state_up, **kwargs):
  48         """Override the routine for allowing the router:external attribute."""
  49         # attributes containing a colon should be passed with
  50         # a double underscore
  51         new_args = dict(zip(map(lambda x: x.replace('__', ':'), kwargs),
  52                             kwargs.values()))
  53         arg_list = new_args.pop('arg_list', ()) + (external_net.EXTERNAL,)
  54         return super(ExtNetDBTestCase, self)._create_network(
  55             fmt, name, admin_state_up, arg_list=arg_list, **new_args)
  56
  57     def setUp(self):
  58         plugin = 'neutron.tests.unit.extensions.test_l3.TestNoL3NatPlugin'
  59         ext_mgr = ExtNetTestExtensionManager()
  60         super(ExtNetDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
  61
  62     def _set_net_external(self, net_id):
  63         self._update('networks', net_id,
  64                      {'network': {external_net.EXTERNAL: True}})
  65
  66     def test_list_nets_external(self):
  67         with self.network() as n1:
  68             self._set_net_external(n1['network']['id'])
  69             with self.network():
  70                 body = self._list('networks')
  71                 self.assertEqual(len(body['networks']), 2)
  72
  73                 body = self._list('networks',
  74                                   query_params="%s=True" %
  75                                                external_net.EXTERNAL)
  76                 self.assertEqual(len(body['networks']), 1)
  77
  78                 body = self._list('networks',
  79                                   query_params="%s=False" %
  80                                                external_net.EXTERNAL)
  81                 self.assertEqual(len(body['networks']), 1)
  82
  83     def test_list_nets_external_pagination(self):
  84         if self._skip_native_pagination:
  85             self.skipTest("Skip test for not implemented pagination feature")
  86         with self.network(name='net1') as n1, self.network(name='net3') as n3:
  87             self._set_net_external(n1['network']['id'])
  88             self._set_net_external(n3['network']['id'])
  89             with self.network(name='net2') as n2:
  90                 self._test_list_with_pagination(
  91                     'network', (n1, n3), ('name', 'asc'), 1, 3,
  92                     query_params='router:external=True')
  93                 self._test_list_with_pagination(
  94                     'network', (n2, ), ('name', 'asc'), 1, 2,
  95                     query_params='router:external=False')
  96
  97     def test_get_network_succeeds_without_filter(self):
  98         plugin = manager.NeutronManager.get_plugin()
  99         ctx = context.Context(None, None, is_admin=True)
 100         result = plugin.get_networks(ctx, filters=None)
 101         self.assertEqual([], result)
 102
 103     def test_update_network_set_external_non_admin_fails(self):
 104         # Assert that a non-admin user cannot update the
 105         # router:external attribute
 106         with self.network(tenant_id='noadmin') as network:
 107             data = {'network': {'router:external': True}}
 108             req = self.new_update_request('networks',
 109                                           data,
 110                                           network['network']['id'])
 111             req.environ['neutron.context'] = context.Context('', 'noadmin')
 112             res = req.get_response(self.api)
 113             self.assertEqual(exc.HTTPForbidden.code, res.status_int)
 114
 115     def test_network_filter_hook_admin_context(self):
 116         plugin = manager.NeutronManager.get_plugin()
 117         ctx = context.Context(None, None, is_admin=True)
 118         model = models_v2.Network
 119         conditions = plugin._network_filter_hook(ctx, model, [])
 120         self.assertEqual([], conditions)
 121
 122     def test_network_filter_hook_nonadmin_context(self):
 123         plugin = manager.NeutronManager.get_plugin()
 124         ctx = context.Context('edinson', 'cavani')
 125         model = models_v2.Network
 126         txt = "externalnetworks.network_id IS NOT NULL"
 127         conditions = plugin._network_filter_hook(ctx, model, [])
 128         self.assertEqual(conditions.__str__(), txt)
 129         # Try to concatenate conditions
 130         conditions = plugin._network_filter_hook(ctx, model, conditions)
 131         self.assertEqual(conditions.__str__(), "%s OR %s" % (txt, txt))
 132
 133     def test_create_port_external_network_non_admin_fails(self):
 134         with self.network(router__external=True) as ext_net:
 135             with self.subnet(network=ext_net) as ext_subnet:
 136                 with testtools.ExpectedException(
 137                         exc.HTTPClientError) as ctx_manager:
 138                     with self.port(subnet=ext_subnet,
 139                                    set_context='True',
 140                                    tenant_id='noadmin'):
 141                         pass
 142                     self.assertEqual(ctx_manager.exception.code, 403)
 143
 144     def test_create_port_external_network_admin_succeeds(self):
 145         with self.network(router__external=True) as ext_net:
 146             with self.subnet(network=ext_net) as ext_subnet:
 147                 with self.port(subnet=ext_subnet) as port:
 148                     self.assertEqual(port['port']['network_id'],
 149                                      ext_net['network']['id'])
 150
 151     def test_create_external_network_non_admin_fails(self):
 152         with testtools.ExpectedException(exc.HTTPClientError) as ctx_manager:
 153             with self.network(router__external=True,
 154                               set_context='True',
 155                               tenant_id='noadmin'):
 156                 pass
 157             self.assertEqual(ctx_manager.exception.code, 403)
 158
 159     def test_create_external_network_admin_succeeds(self):
 160         with self.network(router__external=True) as ext_net:
 161             self.assertTrue(ext_net['network'][external_net.EXTERNAL])
 162
 163     def test_delete_network_check_disassociated_floatingips(self):
 164         with mock.patch.object(manager.NeutronManager,
 165                                'get_service_plugins') as srv_plugins:
 166             l3_mock = mock.Mock()
 167             srv_plugins.return_value = {'L3_ROUTER_NAT': l3_mock}
 168             with self.network() as net:
 169                 req = self.new_delete_request('networks', net['network']['id'])
 170                 res = req.get_response(self.api)
 171                 self.assertEqual(res.status_int, exc.HTTPNoContent.code)
 172                 (l3_mock.delete_disassociated_floatingips
 173                  .assert_called_once_with(mock.ANY, net['network']['id']))