← Back to branch summary

~ubuntu-branches/ubuntu/vivid/neutron/vivid-updates

« back to all changes in this revision

Viewing changes to neutron/tests/unit/metaplugin/test_metaplugin.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2015-03-30 11:17:19 UTC
  • mfrom: (1.1.21)
  • Revision ID: package-import@ubuntu.com-20150330111719-h0gx7233p4jkkgfh
Tags: 1:2015.1~b3-0ubuntu1
https://launchpad.net/bugs/1263539
* New upstream milestone release:
  - d/control: Align version requirements with upstream.
  - d/control: Add new dependency on oslo-log.
  - d/p/*: Rebase.
  - d/control,d/neutron-plugin-hyperv*: Dropped, decomposed into
    separate project upstream.
  - d/control,d/neutron-plugin-openflow*: Dropped, decomposed into
    separate project upstream.
  - d/neutron-common.install: Add neutron-rootwrap-daemon and 
    neutron-keepalived-state-change binaries.
  - d/rules: Ignore neutron-hyperv-agent when installing; only for Windows.
  - d/neutron-plugin-cisco.install: Drop neutron-cisco-cfg-agent as
    decomposed into separate project upstream.
  - d/neutron-plugin-vmware.install: Drop neutron-check-nsx-config and
    neutron-nsx-manage as decomposed into separate project upstream.
  - d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent.
* d/pydist-overrides: Add overrides for oslo packages.
* d/control: Fixup type in package description (LP: #1263539).
* d/p/fixup-driver-test-execution.patch: Cherry pick fix from upstream VCS
  to support unit test exection in out-of-tree vendor drivers.
* d/neutron-common.postinst: Allow general access to /etc/neutron but limit
  access to root/neutron to /etc/neutron/neutron.conf to support execution
  of unit tests in decomposed vendor drivers.
* d/control: Add dependency on python-neutron-fwaas to neutron-l3-agent
  package.

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/metaplugin/test_metaplugin.py
1
 
# Copyright 2012, Nachi Ueno, NTT MCL, Inc.
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.config import cfg
18
 
import testtools
19
 
 
20
 
from neutron.common import exceptions as exc
21
 
from neutron.common import topics
22
 
from neutron import context
23
 
from neutron.db import db_base_plugin_v2
24
 
from neutron.db import models_v2
25
 
from neutron.extensions import flavor as ext_flavor
26
 
from neutron.openstack.common import uuidutils
27
 
from neutron.plugins.metaplugin import meta_neutron_plugin
28
 
from neutron.tests.unit import testlib_api
29
 
from neutron.tests.unit import testlib_plugin
30
 
 
31
 
CONF_FILE = ""
32
 
META_PATH = "neutron.plugins.metaplugin"
33
 
FAKE_PATH = "neutron.tests.unit.metaplugin"
34
 
PROXY_PATH = "%s.proxy_neutron_plugin.ProxyPluginV2" % META_PATH
35
 
PLUGIN_LIST = """
36
 
fake1:%s.fake_plugin.Fake1,fake2:%s.fake_plugin.Fake2,proxy:%s
37
 
""".strip() % (FAKE_PATH, FAKE_PATH, PROXY_PATH)
38
 
L3_PLUGIN_LIST = """
39
 
fake1:%s.fake_plugin.Fake1,fake2:%s.fake_plugin.Fake2
40
 
""".strip() % (FAKE_PATH, FAKE_PATH)
41
 
 
42
 
 
43
 
def setup_metaplugin_conf(has_l3=True):
44
 
    cfg.CONF.set_override('auth_url', 'http://localhost:35357/v2.0',
45
 
                                      'PROXY')
46
 
    cfg.CONF.set_override('auth_region', 'RegionOne', 'PROXY')
47
 
    cfg.CONF.set_override('admin_user', 'neutron', 'PROXY')
48
 
    cfg.CONF.set_override('admin_password', 'password', 'PROXY')
49
 
    cfg.CONF.set_override('admin_tenant_name', 'service', 'PROXY')
50
 
    cfg.CONF.set_override('plugin_list', PLUGIN_LIST, 'META')
51
 
    if has_l3:
52
 
        cfg.CONF.set_override('l3_plugin_list', L3_PLUGIN_LIST, 'META')
53
 
    else:
54
 
        cfg.CONF.set_override('l3_plugin_list', "", 'META')
55
 
    cfg.CONF.set_override('default_flavor', 'fake2', 'META')
56
 
    cfg.CONF.set_override('default_l3_flavor', 'fake1', 'META')
57
 
    cfg.CONF.set_override('base_mac', "12:34:56:78:90:ab")
58
 
    #TODO(nati) remove this after subnet quota change is merged
59
 
    cfg.CONF.set_override('max_dns_nameservers', 10)
60
 
 
61
 
 
62
 
# Hooks registered by metaplugin must not exist for other plugins UT.
63
 
# So hooks must be unregistered (overwrite to None in fact).
64
 
def unregister_meta_hooks():
65
 
    db_base_plugin_v2.NeutronDbPluginV2.register_model_query_hook(
66
 
        models_v2.Network, 'metaplugin_net', None, None, None)
67
 
    db_base_plugin_v2.NeutronDbPluginV2.register_model_query_hook(
68
 
        models_v2.Port, 'metaplugin_port', None, None, None)
69
 
 
70
 
 
71
 
class MetaNeutronPluginV2Test(testlib_api.SqlTestCase,
72
 
                              testlib_plugin.PluginSetupHelper):
73
 
    """Class conisting of MetaNeutronPluginV2 unit tests."""
74
 
 
75
 
    has_l3 = True
76
 
 
77
 
    def setUp(self):
78
 
        super(MetaNeutronPluginV2Test, self).setUp()
79
 
        self.fake_tenant_id = uuidutils.generate_uuid()
80
 
        self.context = context.get_admin_context()
81
 
 
82
 
        self.addCleanup(unregister_meta_hooks)
83
 
 
84
 
        setup_metaplugin_conf(self.has_l3)
85
 
 
86
 
        self.client_cls_p = mock.patch('neutronclient.v2_0.client.Client')
87
 
        client_cls = self.client_cls_p.start()
88
 
        self.client_inst = mock.Mock()
89
 
        client_cls.return_value = self.client_inst
90
 
        self.client_inst.create_network.return_value = \
91
 
            {'id': 'fake_id'}
92
 
        self.client_inst.create_port.return_value = \
93
 
            {'id': 'fake_id'}
94
 
        self.client_inst.create_subnet.return_value = \
95
 
            {'id': 'fake_id'}
96
 
        self.client_inst.update_network.return_value = \
97
 
            {'id': 'fake_id'}
98
 
        self.client_inst.update_port.return_value = \
99
 
            {'id': 'fake_id'}
100
 
        self.client_inst.update_subnet.return_value = \
101
 
            {'id': 'fake_id'}
102
 
        self.client_inst.delete_network.return_value = True
103
 
        self.client_inst.delete_port.return_value = True
104
 
        self.client_inst.delete_subnet.return_value = True
105
 
        plugin = (meta_neutron_plugin.MetaPluginV2.__module__ + '.'
106
 
                  + meta_neutron_plugin.MetaPluginV2.__name__)
107
 
        self.setup_coreplugin(plugin)
108
 
        self.plugin = meta_neutron_plugin.MetaPluginV2(configfile=None)
109
 
 
110
 
    def _fake_network(self, flavor):
111
 
        data = {'network': {'name': flavor,
112
 
                            'admin_state_up': True,
113
 
                            'shared': False,
114
 
                            'router:external': [],
115
 
                            'tenant_id': self.fake_tenant_id,
116
 
                            ext_flavor.FLAVOR_NETWORK: flavor}}
117
 
        return data
118
 
 
119
 
    def _fake_port(self, net_id):
120
 
        return {'port': {'name': net_id,
121
 
                         'network_id': net_id,
122
 
                         'admin_state_up': True,
123
 
                         'device_id': 'bad_device_id',
124
 
                         'device_owner': 'bad_device_owner',
125
 
                         'admin_state_up': True,
126
 
                         'host_routes': [],
127
 
                         'fixed_ips': [],
128
 
                         'mac_address': self.plugin._generate_mac(),
129
 
                         'tenant_id': self.fake_tenant_id}}
130
 
 
131
 
    def _fake_subnet(self, net_id):
132
 
        allocation_pools = [{'start': '10.0.0.2',
133
 
                             'end': '10.0.0.254'}]
134
 
        return {'subnet': {'name': net_id,
135
 
                           'network_id': net_id,
136
 
                           'gateway_ip': '10.0.0.1',
137
 
                           'dns_nameservers': ['10.0.0.2'],
138
 
                           'host_routes': [],
139
 
                           'cidr': '10.0.0.0/24',
140
 
                           'allocation_pools': allocation_pools,
141
 
                           'enable_dhcp': True,
142
 
                           'ip_version': 4}}
143
 
 
144
 
    def _fake_router(self, flavor):
145
 
        data = {'router': {'name': flavor, 'admin_state_up': True,
146
 
                           'tenant_id': self.fake_tenant_id,
147
 
                           ext_flavor.FLAVOR_ROUTER: flavor,
148
 
                           'external_gateway_info': None}}
149
 
        return data
150
 
 
151
 
    def test_create_delete_network(self):
152
 
        network1 = self._fake_network('fake1')
153
 
        ret1 = self.plugin.create_network(self.context, network1)
154
 
        self.assertEqual('fake1', ret1[ext_flavor.FLAVOR_NETWORK])
155
 
 
156
 
        network2 = self._fake_network('fake2')
157
 
        ret2 = self.plugin.create_network(self.context, network2)
158
 
        self.assertEqual('fake2', ret2[ext_flavor.FLAVOR_NETWORK])
159
 
 
160
 
        network3 = self._fake_network('proxy')
161
 
        ret3 = self.plugin.create_network(self.context, network3)
162
 
        self.assertEqual('proxy', ret3[ext_flavor.FLAVOR_NETWORK])
163
 
 
164
 
        db_ret1 = self.plugin.get_network(self.context, ret1['id'])
165
 
        self.assertEqual('fake1', db_ret1['name'])
166
 
 
167
 
        db_ret2 = self.plugin.get_network(self.context, ret2['id'])
168
 
        self.assertEqual('fake2', db_ret2['name'])
169
 
 
170
 
        db_ret3 = self.plugin.get_network(self.context, ret3['id'])
171
 
        self.assertEqual('proxy', db_ret3['name'])
172
 
 
173
 
        db_ret4 = self.plugin.get_networks(self.context)
174
 
        self.assertEqual(3, len(db_ret4))
175
 
 
176
 
        db_ret5 = self.plugin.get_networks(
177
 
            self.context,
178
 
            {ext_flavor.FLAVOR_NETWORK: ['fake1']})
179
 
        self.assertEqual(1, len(db_ret5))
180
 
        self.assertEqual('fake1', db_ret5[0]['name'])
181
 
        self.plugin.delete_network(self.context, ret1['id'])
182
 
        self.plugin.delete_network(self.context, ret2['id'])
183
 
        self.plugin.delete_network(self.context, ret3['id'])
184
 
 
185
 
    def test_create_delete_port(self):
186
 
        network1 = self._fake_network('fake1')
187
 
        network_ret1 = self.plugin.create_network(self.context, network1)
188
 
        network2 = self._fake_network('fake2')
189
 
        network_ret2 = self.plugin.create_network(self.context, network2)
190
 
        network3 = self._fake_network('proxy')
191
 
        network_ret3 = self.plugin.create_network(self.context, network3)
192
 
 
193
 
        port1 = self._fake_port(network_ret1['id'])
194
 
        port2 = self._fake_port(network_ret2['id'])
195
 
        port3 = self._fake_port(network_ret3['id'])
196
 
 
197
 
        port1_ret = self.plugin.create_port(self.context, port1)
198
 
        port2_ret = self.plugin.create_port(self.context, port2)
199
 
        port3_ret = self.plugin.create_port(self.context, port3)
200
 
        ports_all = self.plugin.get_ports(self.context)
201
 
 
202
 
        self.assertEqual(network_ret1['id'], port1_ret['network_id'])
203
 
        self.assertEqual(network_ret2['id'], port2_ret['network_id'])
204
 
        self.assertEqual(network_ret3['id'], port3_ret['network_id'])
205
 
        self.assertEqual(3, len(ports_all))
206
 
 
207
 
        port1_dict = self.plugin._make_port_dict(port1_ret)
208
 
        port2_dict = self.plugin._make_port_dict(port2_ret)
209
 
        port3_dict = self.plugin._make_port_dict(port3_ret)
210
 
 
211
 
        self.assertEqual(port1_dict, port1_ret)
212
 
        self.assertEqual(port2_dict, port2_ret)
213
 
        self.assertEqual(port3_dict, port3_ret)
214
 
 
215
 
        port1['port']['admin_state_up'] = False
216
 
        port2['port']['admin_state_up'] = False
217
 
        port3['port']['admin_state_up'] = False
218
 
        self.plugin.update_port(self.context, port1_ret['id'], port1)
219
 
        self.plugin.update_port(self.context, port2_ret['id'], port2)
220
 
        self.plugin.update_port(self.context, port3_ret['id'], port3)
221
 
        port_in_db1 = self.plugin.get_port(self.context, port1_ret['id'])
222
 
        port_in_db2 = self.plugin.get_port(self.context, port2_ret['id'])
223
 
        port_in_db3 = self.plugin.get_port(self.context, port3_ret['id'])
224
 
        self.assertEqual(False, port_in_db1['admin_state_up'])
225
 
        self.assertEqual(False, port_in_db2['admin_state_up'])
226
 
        self.assertEqual(False, port_in_db3['admin_state_up'])
227
 
 
228
 
        self.plugin.delete_port(self.context, port1_ret['id'])
229
 
        self.plugin.delete_port(self.context, port2_ret['id'])
230
 
        self.plugin.delete_port(self.context, port3_ret['id'])
231
 
 
232
 
        self.plugin.delete_network(self.context, network_ret1['id'])
233
 
        self.plugin.delete_network(self.context, network_ret2['id'])
234
 
        self.plugin.delete_network(self.context, network_ret3['id'])
235
 
 
236
 
    def test_create_delete_subnet(self):
237
 
        # for this test we need to enable overlapping ips
238
 
        cfg.CONF.set_default('allow_overlapping_ips', True)
239
 
        network1 = self._fake_network('fake1')
240
 
        network_ret1 = self.plugin.create_network(self.context, network1)
241
 
        network2 = self._fake_network('fake2')
242
 
        network_ret2 = self.plugin.create_network(self.context, network2)
243
 
        network3 = self._fake_network('proxy')
244
 
        network_ret3 = self.plugin.create_network(self.context, network3)
245
 
 
246
 
        subnet1 = self._fake_subnet(network_ret1['id'])
247
 
        subnet2 = self._fake_subnet(network_ret2['id'])
248
 
        subnet3 = self._fake_subnet(network_ret3['id'])
249
 
 
250
 
        subnet1_ret = self.plugin.create_subnet(self.context, subnet1)
251
 
        subnet2_ret = self.plugin.create_subnet(self.context, subnet2)
252
 
        subnet3_ret = self.plugin.create_subnet(self.context, subnet3)
253
 
        self.assertEqual(network_ret1['id'], subnet1_ret['network_id'])
254
 
        self.assertEqual(network_ret2['id'], subnet2_ret['network_id'])
255
 
        self.assertEqual(network_ret3['id'], subnet3_ret['network_id'])
256
 
 
257
 
        subnet_in_db1 = self.plugin.get_subnet(self.context, subnet1_ret['id'])
258
 
        subnet_in_db2 = self.plugin.get_subnet(self.context, subnet2_ret['id'])
259
 
        subnet_in_db3 = self.plugin.get_subnet(self.context, subnet3_ret['id'])
260
 
 
261
 
        subnet1['subnet']['allocation_pools'].pop()
262
 
        subnet2['subnet']['allocation_pools'].pop()
263
 
        subnet3['subnet']['allocation_pools'].pop()
264
 
 
265
 
        self.plugin.update_subnet(self.context,
266
 
                                  subnet1_ret['id'], subnet1)
267
 
        self.plugin.update_subnet(self.context,
268
 
                                  subnet2_ret['id'], subnet2)
269
 
        self.plugin.update_subnet(self.context,
270
 
                                  subnet3_ret['id'], subnet3)
271
 
        subnet_in_db1 = self.plugin.get_subnet(self.context, subnet1_ret['id'])
272
 
        subnet_in_db2 = self.plugin.get_subnet(self.context, subnet2_ret['id'])
273
 
        subnet_in_db3 = self.plugin.get_subnet(self.context, subnet3_ret['id'])
274
 
 
275
 
        self.assertEqual(4, subnet_in_db1['ip_version'])
276
 
        self.assertEqual(4, subnet_in_db2['ip_version'])
277
 
        self.assertEqual(4, subnet_in_db3['ip_version'])
278
 
 
279
 
        self.plugin.delete_subnet(self.context, subnet1_ret['id'])
280
 
        self.plugin.delete_subnet(self.context, subnet2_ret['id'])
281
 
        self.plugin.delete_subnet(self.context, subnet3_ret['id'])
282
 
 
283
 
        self.plugin.delete_network(self.context, network_ret1['id'])
284
 
        self.plugin.delete_network(self.context, network_ret2['id'])
285
 
        self.plugin.delete_network(self.context, network_ret3['id'])
286
 
 
287
 
    def test_create_delete_router(self):
288
 
        router1 = self._fake_router('fake1')
289
 
        router_ret1 = self.plugin.create_router(self.context, router1)
290
 
        router2 = self._fake_router('fake2')
291
 
        router_ret2 = self.plugin.create_router(self.context, router2)
292
 
 
293
 
        self.assertEqual('fake1', router_ret1[ext_flavor.FLAVOR_ROUTER])
294
 
        self.assertEqual('fake2', router_ret2[ext_flavor.FLAVOR_ROUTER])
295
 
 
296
 
        router_in_db1 = self.plugin.get_router(self.context, router_ret1['id'])
297
 
        router_in_db2 = self.plugin.get_router(self.context, router_ret2['id'])
298
 
 
299
 
        self.assertEqual('fake1', router_in_db1[ext_flavor.FLAVOR_ROUTER])
300
 
        self.assertEqual('fake2', router_in_db2[ext_flavor.FLAVOR_ROUTER])
301
 
 
302
 
        self.plugin.delete_router(self.context, router_ret1['id'])
303
 
        self.plugin.delete_router(self.context, router_ret2['id'])
304
 
        with testtools.ExpectedException(meta_neutron_plugin.FlavorNotFound):
305
 
            self.plugin.get_router(self.context, router_ret1['id'])
306
 
 
307
 
    def test_extension_method(self):
308
 
        """Test if plugin methods are accessible from self.plugin
309
 
 
310
 
        This test compensates for the nondeterministic ordering of
311
 
        self.plugin's plugins dictionary. Fake Plugin 1 and Fake Plugin 2
312
 
        both have a function called fake_func and the order of
313
 
        self.plugin.plugins will determine which fake_func is called.
314
 
        """
315
 
        fake1 = self.plugin.plugins.keys().index('fake1')
316
 
        fake2 = self.plugin.plugins.keys().index('fake2')
317
 
        fake1_before_fake2 = fake1 < fake2
318
 
 
319
 
        fake_func_return = 'fake1' if fake1_before_fake2 else 'fake2'
320
 
 
321
 
        self.assertEqual(fake_func_return, self.plugin.fake_func())
322
 
        self.assertEqual('fake2', self.plugin.fake_func2())
323
 
 
324
 
    def test_extension_not_implemented_method(self):
325
 
        try:
326
 
            self.plugin.not_implemented()
327
 
        except AttributeError:
328
 
            return
329
 
        except Exception:
330
 
            self.fail("AttributeError Error is not raised")
331
 
 
332
 
        self.fail("No Error is not raised")
333
 
 
334
 
    def test_create_network_flavor_fail(self):
335
 
        with mock.patch('neutron.plugins.metaplugin.meta_db_v2.'
336
 
                        'add_network_flavor_binding',
337
 
                        side_effect=Exception):
338
 
            network = self._fake_network('fake1')
339
 
            self.assertRaises(meta_neutron_plugin.FaildToAddFlavorBinding,
340
 
                              self.plugin.create_network,
341
 
                              self.context,
342
 
                              network)
343
 
            count = self.plugin.get_networks_count(self.context)
344
 
            self.assertEqual(count, 0)
345
 
 
346
 
    def test_create_router_flavor_fail(self):
347
 
        with mock.patch('neutron.plugins.metaplugin.meta_db_v2.'
348
 
                        'add_router_flavor_binding',
349
 
                        side_effect=Exception):
350
 
            router = self._fake_router('fake1')
351
 
            self.assertRaises(meta_neutron_plugin.FaildToAddFlavorBinding,
352
 
                              self.plugin.create_router,
353
 
                              self.context,
354
 
                              router)
355
 
            count = self.plugin.get_routers_count(self.context)
356
 
            self.assertEqual(count, 0)
357
 
 
358
 
 
359
 
class MetaNeutronPluginV2TestWithoutL3(MetaNeutronPluginV2Test):
360
 
    """Tests without l3_plugin_list configration."""
361
 
 
362
 
    has_l3 = False
363
 
 
364
 
    def test_supported_extension_aliases(self):
365
 
        self.assertEqual(self.plugin.supported_extension_aliases,
366
 
                         ['flavor', 'external-net'])
367
 
 
368
 
    def test_create_delete_router(self):
369
 
        self.skipTest("Test case without router")
370
 
 
371
 
    def test_create_router_flavor_fail(self):
372
 
        self.skipTest("Test case without router")
373
 
 
374
 
 
375
 
class MetaNeutronPluginV2TestRpcFlavor(testlib_api.SqlTestCase):
376
 
    """Tests for rpc_flavor."""
377
 
 
378
 
    def setUp(self):
379
 
        super(MetaNeutronPluginV2TestRpcFlavor, self).setUp()
380
 
        self.addCleanup(unregister_meta_hooks)
381
 
 
382
 
    def test_rpc_flavor(self):
383
 
        setup_metaplugin_conf()
384
 
        cfg.CONF.set_override('rpc_flavor', 'fake1', 'META')
385
 
        self.plugin = meta_neutron_plugin.MetaPluginV2()
386
 
        self.assertEqual(topics.PLUGIN, 'q-plugin')
387
 
        ret = self.plugin.rpc_workers_supported()
388
 
        self.assertFalse(ret)
389
 
 
390
 
    def test_invalid_rpc_flavor(self):
391
 
        setup_metaplugin_conf()
392
 
        cfg.CONF.set_override('rpc_flavor', 'fake-fake', 'META')
393
 
        self.assertRaises(exc.Invalid,
394
 
                          meta_neutron_plugin.MetaPluginV2)
395
 
        self.assertEqual(topics.PLUGIN, 'q-plugin')
396
 
 
397
 
    def test_rpc_flavor_multiple_rpc_workers(self):
398
 
        setup_metaplugin_conf()
399
 
        cfg.CONF.set_override('rpc_flavor', 'fake2', 'META')
400
 
        self.plugin = meta_neutron_plugin.MetaPluginV2()
401
 
        self.assertEqual(topics.PLUGIN, 'q-plugin')
402
 
        ret = self.plugin.rpc_workers_supported()
403
 
        self.assertTrue(ret)
404
 
        ret = self.plugin.start_rpc_listeners()
405
 
        self.assertEqual('OK', ret)