← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/ml2/test_ml2_plugin.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/ml2/test_ml2_plugin.py
16
16
import contextlib
17
17
import functools
18
18
import mock
 
19
import six
19
20
import testtools
20
21
import uuid
21
22
import webob
22
23
 
 
24
from oslo_db import exception as db_exc
 
25
 
23
26
from neutron.common import constants
24
27
from neutron.common import exceptions as exc
25
28
from neutron.common import utils
26
29
from neutron import context
 
30
from neutron.db import api as db_api
27
31
from neutron.db import db_base_plugin_v2 as base_plugin
28
32
from neutron.db import l3_db
29
33
from neutron.extensions import external_net as external_net
30
 
from neutron.extensions import l3agentscheduler
31
34
from neutron.extensions import multiprovidernet as mpnet
32
35
from neutron.extensions import portbindings
33
36
from neutron.extensions import providernet as pnet
137
140
 
138
141
class TestMl2NetworksV2(test_plugin.TestNetworksV2,
139
142
                        Ml2PluginV2TestCase):
 
143
    def setUp(self, plugin=None):
 
144
        super(TestMl2NetworksV2, self).setUp()
 
145
        # provider networks
 
146
        self.pnets = [{'name': 'net1',
 
147
                       pnet.NETWORK_TYPE: 'vlan',
 
148
                       pnet.PHYSICAL_NETWORK: 'physnet1',
 
149
                       pnet.SEGMENTATION_ID: 1,
 
150
                       'tenant_id': 'tenant_one'},
 
151
                      {'name': 'net2',
 
152
                       pnet.NETWORK_TYPE: 'vlan',
 
153
                       pnet.PHYSICAL_NETWORK: 'physnet2',
 
154
                       pnet.SEGMENTATION_ID: 210,
 
155
                       'tenant_id': 'tenant_one'},
 
156
                      {'name': 'net3',
 
157
                       pnet.NETWORK_TYPE: 'vlan',
 
158
                       pnet.PHYSICAL_NETWORK: 'physnet2',
 
159
                       pnet.SEGMENTATION_ID: 220,
 
160
                       'tenant_id': 'tenant_one'}
 
161
                      ]
 
162
        # multiprovider networks
 
163
        self.mp_nets = [{'name': 'net4',
 
164
                         mpnet.SEGMENTS:
 
165
                             [{pnet.NETWORK_TYPE: 'vlan',
 
166
                               pnet.PHYSICAL_NETWORK: 'physnet2',
 
167
                               pnet.SEGMENTATION_ID: 1},
 
168
                              {pnet.NETWORK_TYPE: 'vlan',
 
169
                               pnet.PHYSICAL_NETWORK: 'physnet2',
 
170
                               pnet.SEGMENTATION_ID: 202}],
 
171
                         'tenant_id': 'tenant_one'}
 
172
                        ]
 
173
        self.nets = self.mp_nets + self.pnets
 
174
 
140
175
    def test_port_delete_helper_tolerates_failure(self):
141
176
        plugin = manager.NeutronManager.get_plugin()
142
177
        with mock.patch.object(plugin, "delete_port",
149
184
                               side_effect=exc.SubnetNotFound(subnet_id="1")):
150
185
            plugin._delete_subnets(None, [mock.MagicMock()])
151
186
 
 
187
    def _create_and_verify_networks(self, networks):
 
188
        for net_idx, net in enumerate(networks):
 
189
            # create
 
190
            req = self.new_create_request('networks',
 
191
                                          {'network': net})
 
192
            # verify
 
193
            network = self.deserialize(self.fmt,
 
194
                                       req.get_response(self.api))['network']
 
195
            if mpnet.SEGMENTS not in net:
 
196
                for k, v in six.iteritems(net):
 
197
                    self.assertEqual(net[k], network[k])
 
198
                    self.assertNotIn(mpnet.SEGMENTS, network)
 
199
            else:
 
200
                segments = network[mpnet.SEGMENTS]
 
201
                expected_segments = net[mpnet.SEGMENTS]
 
202
                self.assertEqual(len(expected_segments), len(segments))
 
203
                for expected, actual in zip(expected_segments, segments):
 
204
                    self.assertEqual(expected, actual)
 
205
 
 
206
    def _lookup_network_by_segmentation_id(self, seg_id, num_expected_nets):
 
207
        params_str = "%s=%s" % (pnet.SEGMENTATION_ID, seg_id)
 
208
        net_req = self.new_list_request('networks', None,
 
209
                                        params=params_str)
 
210
        networks = self.deserialize(self.fmt, net_req.get_response(self.api))
 
211
        if num_expected_nets:
 
212
            self.assertIsNotNone(networks)
 
213
            self.assertEqual(num_expected_nets, len(networks['networks']))
 
214
        else:
 
215
            self.assertIsNone(networks)
 
216
        return networks
 
217
 
 
218
    def test_list_networks_with_segmentation_id(self):
 
219
        self._create_and_verify_networks(self.pnets)
 
220
        # verify we can find the network that we expect
 
221
        lookup_vlan_id = 1
 
222
        expected_net = [n for n in self.pnets
 
223
                        if n[pnet.SEGMENTATION_ID] == lookup_vlan_id].pop()
 
224
        networks = self._lookup_network_by_segmentation_id(lookup_vlan_id, 1)
 
225
        # verify all provider attributes
 
226
        network = networks['networks'][0]
 
227
        for attr in pnet.ATTRIBUTES:
 
228
            self.assertEqual(expected_net[attr], network[attr])
 
229
 
 
230
    def test_list_mpnetworks_with_segmentation_id(self):
 
231
        self._create_and_verify_networks(self.nets)
 
232
 
 
233
        # get all networks with seg_id=1 (including multisegment networks)
 
234
        lookup_vlan_id = 1
 
235
        networks = self._lookup_network_by_segmentation_id(lookup_vlan_id, 2)
 
236
 
 
237
        # get the mpnet
 
238
        networks = [n for n in networks['networks'] if mpnet.SEGMENTS in n]
 
239
        network = networks.pop()
 
240
        # verify attributes of the looked up item
 
241
        segments = network[mpnet.SEGMENTS]
 
242
        expected_segments = self.mp_nets[0][mpnet.SEGMENTS]
 
243
        self.assertEqual(len(expected_segments), len(segments))
 
244
        for expected, actual in zip(expected_segments, segments):
 
245
            self.assertEqual(expected, actual)
 
246
 
 
247
    def test_create_network_segment_allocation_fails(self):
 
248
        plugin = manager.NeutronManager.get_plugin()
 
249
        with mock.patch.object(plugin.type_manager, 'create_network_segments',
 
250
            side_effect=db_exc.RetryRequest(ValueError())) as f:
 
251
            self.assertRaises(ValueError,
 
252
                              plugin.create_network,
 
253
                              context.get_admin_context(),
 
254
                              {'network': {'tenant_id': 'sometenant',
 
255
                                           'name': 'dummy',
 
256
                                           'admin_state_up': True,
 
257
                                           'shared': False}})
 
258
            self.assertEqual(db_api.MAX_RETRIES + 1, f.call_count)
 
259
 
152
260
 
153
261
class TestMl2SubnetsV2(test_plugin.TestSubnetsV2,
154
262
                       Ml2PluginV2TestCase):
217
325
                self._validate_behavior_on_bulk_failure(
218
326
                    res, 'ports', webob.exc.HTTPServerError.code)
219
327
 
 
328
    def test_create_ports_bulk_with_sec_grp(self):
 
329
        ctx = context.get_admin_context()
 
330
        plugin = manager.NeutronManager.get_plugin()
 
331
        with contextlib.nested(
 
332
            self.network(),
 
333
            mock.patch.object(plugin.notifier,
 
334
                              'security_groups_member_updated'),
 
335
            mock.patch.object(plugin.notifier,
 
336
                              'security_groups_provider_updated')
 
337
        ) as (net, m_upd, p_upd):
 
338
 
 
339
            res = self._create_port_bulk(self.fmt, 3, net['network']['id'],
 
340
                                         'test', True, context=ctx)
 
341
            ports = self.deserialize(self.fmt, res)
 
342
            used_sg = ports['ports'][0]['security_groups']
 
343
            m_upd.assert_called_once_with(ctx, used_sg)
 
344
            self.assertFalse(p_upd.called)
 
345
 
 
346
    def test_create_ports_bulk_with_sec_grp_member_provider_update(self):
 
347
        ctx = context.get_admin_context()
 
348
        plugin = manager.NeutronManager.get_plugin()
 
349
        with contextlib.nested(
 
350
            self.network(),
 
351
            mock.patch.object(plugin.notifier,
 
352
                              'security_groups_member_updated'),
 
353
            mock.patch.object(plugin.notifier,
 
354
                              'security_groups_provider_updated')
 
355
        ) as (net, m_upd, p_upd):
 
356
 
 
357
            net_id = net['network']['id']
 
358
            data = [{
 
359
                    'network_id': net_id,
 
360
                    'tenant_id': self._tenant_id
 
361
                    },
 
362
                    {
 
363
                    'network_id': net_id,
 
364
                    'tenant_id': self._tenant_id,
 
365
                    'device_owner': constants.DEVICE_OWNER_DHCP
 
366
                    }
 
367
                    ]
 
368
 
 
369
            res = self._create_bulk_from_list(self.fmt, 'port',
 
370
                                              data, context=ctx)
 
371
            ports = self.deserialize(self.fmt, res)
 
372
            used_sg = ports['ports'][0]['security_groups']
 
373
            m_upd.assert_called_once_with(ctx, used_sg)
 
374
            p_upd.assert_called_once_with(ctx)
 
375
 
 
376
            m_upd.reset_mock()
 
377
            p_upd.reset_mock()
 
378
            data[0]['device_owner'] = constants.DEVICE_OWNER_DHCP
 
379
            self._create_bulk_from_list(self.fmt, 'port',
 
380
                                        data, context=ctx)
 
381
            self.assertFalse(m_upd.called)
 
382
            p_upd.assert_called_once_with(ctx)
 
383
 
 
384
    def test_create_ports_bulk_with_sec_grp_provider_update_ipv6(self):
 
385
        ctx = context.get_admin_context()
 
386
        plugin = manager.NeutronManager.get_plugin()
 
387
        fake_prefix = '2001:db8::/64'
 
388
        fake_gateway = 'fe80::1'
 
389
        with self.network() as net:
 
390
            with contextlib.nested(
 
391
                self.subnet(net, gateway_ip=fake_gateway,
 
392
                            cidr=fake_prefix, ip_version=6),
 
393
                mock.patch.object(
 
394
                    plugin.notifier, 'security_groups_member_updated'),
 
395
                mock.patch.object(
 
396
                    plugin.notifier, 'security_groups_provider_updated')
 
397
            ) as (snet_v6, m_upd, p_upd):
 
398
 
 
399
                net_id = net['network']['id']
 
400
                data = [{
 
401
                        'network_id': net_id,
 
402
                        'tenant_id': self._tenant_id,
 
403
                        'fixed_ips': [{'subnet_id': snet_v6['subnet']['id']}],
 
404
                        'device_owner': constants.DEVICE_OWNER_ROUTER_INTF
 
405
                        }
 
406
                        ]
 
407
                self._create_bulk_from_list(self.fmt, 'port',
 
408
                                            data, context=ctx)
 
409
                self.assertFalse(m_upd.called)
 
410
                p_upd.assert_called_once_with(ctx)
 
411
 
220
412
    def test_delete_port_no_notify_in_disassociate_floatingips(self):
221
413
        ctx = context.get_admin_context()
222
414
        plugin = manager.NeutronManager.get_plugin()
223
415
        l3plugin = manager.NeutronManager.get_service_plugins().get(
224
416
            service_constants.L3_ROUTER_NAT)
225
417
        with contextlib.nested(
226
 
            self.port(do_delete=False),
 
418
            self.port(),
227
419
            mock.patch.object(l3plugin, 'disassociate_floatingips'),
228
420
            mock.patch.object(l3plugin, 'notify_routers_updated')
229
421
        ) as (port, disassociate_floatingips, notify):
268
460
            self.assertIsNone(l3plugin.disassociate_floatingips(ctx, port_id))
269
461
 
270
462
 
 
463
class TestMl2PluginOnly(Ml2PluginV2TestCase):
 
464
    """For testing methods that don't call drivers"""
 
465
 
 
466
    def _test_check_mac_update_allowed(self, vif_type, expect_change=True):
 
467
        plugin = manager.NeutronManager.get_plugin()
 
468
        port = {'mac_address': "fake_mac", 'id': "fake_id"}
 
469
        if expect_change:
 
470
            new_attrs = {"mac_address": "dummy_mac"}
 
471
        else:
 
472
            new_attrs = {"mac_address": port['mac_address']}
 
473
        binding = mock.Mock()
 
474
        binding.vif_type = vif_type
 
475
        mac_changed = plugin._check_mac_update_allowed(port, new_attrs,
 
476
                                                       binding)
 
477
        self.assertEqual(expect_change, mac_changed)
 
478
 
 
479
    def test_check_mac_update_allowed_if_no_mac_change(self):
 
480
        self._test_check_mac_update_allowed(portbindings.VIF_TYPE_UNBOUND,
 
481
                                            expect_change=False)
 
482
 
 
483
    def test_check_mac_update_allowed_unless_bound(self):
 
484
        with testtools.ExpectedException(exc.PortBound):
 
485
            self._test_check_mac_update_allowed(portbindings.VIF_TYPE_OVS)
 
486
 
 
487
 
271
488
class TestMl2DvrPortsV2(TestMl2PortsV2):
272
489
    def setUp(self):
273
490
        super(TestMl2DvrPortsV2, self).setUp()
291
508
            mock.patch.object(manager.NeutronManager,
292
509
                              'get_service_plugins',
293
510
                              return_value=self.service_plugins),
294
 
            self.port(do_delete=False,
295
 
                      device_owner=device_owner),
 
511
            self.port(device_owner=device_owner),
296
512
            mock.patch.object(self.l3plugin, 'notify_routers_updated'),
297
513
            mock.patch.object(self.l3plugin, 'disassociate_floatingips',
298
514
                              return_value=fip_set),
329
545
            mock.patch.object(manager.NeutronManager,
330
546
                              'get_service_plugins',
331
547
                              return_value=self.service_plugins),
332
 
            self.port(do_delete=False,
333
 
                      device_owner='compute:None'),
 
548
            self.port(device_owner='compute:None'),
334
549
            mock.patch.object(self.l3plugin, 'dvr_deletens_if_no_port',
335
550
                              return_value=[ns_to_delete]),
336
 
            mock.patch.object(self.l3plugin, 'remove_router_from_l3_agent',
337
 
                side_effect=l3agentscheduler.RouterNotHostedByL3Agent(
338
 
                            router_id=ns_to_delete['router_id'],
339
 
                            agent_id=ns_to_delete['agent_id']))
 
551
            mock.patch.object(self.l3plugin, 'remove_router_from_l3_agent')
340
552
        ) as (get_service_plugin, port, dvr_delns_ifno_port,
341
553
              remove_router_from_l3_agent):
342
554
 
444
656
            mech_context = driver_context.PortContext(
445
657
                plugin, self.context, port['port'],
446
658
                plugin.get_network(self.context, port['port']['network_id']),
447
 
                binding)
 
659
                binding, None)
448
660
        with contextlib.nested(
449
661
            mock.patch('neutron.plugins.ml2.plugin.'
450
662
                       'db.get_locked_port_and_binding',
479
691
                            router_id='old_router_id',
480
692
                            vif_type=portbindings.VIF_TYPE_OVS,
481
693
                            vnic_type=portbindings.VNIC_NORMAL,
482
 
                            cap_port_filter=False,
483
694
                            status=constants.PORT_STATUS_DOWN)
484
695
        plugin = manager.NeutronManager.get_plugin()
485
696
        mock_network = {'id': 'net_id'}
 
697
        mock_port = {'id': 'port_id'}
486
698
        context = mock.Mock()
487
699
        new_router_id = 'new_router'
488
700
        attrs = {'device_id': new_router_id, portbindings.HOST_ID: host_id}
490
702
            with mock.patch.object(ml2_db, 'get_network_segments',
491
703
                                   return_value=[]):
492
704
                mech_context = driver_context.PortContext(
493
 
                    self, context, 'port', mock_network, binding)
 
705
                    self, context, mock_port, mock_network, binding, None)
494
706
                plugin._process_dvr_port_binding(mech_context, context, attrs)
495
707
                self.assertEqual(new_router_id,
496
708
                                 mech_context._binding.router_id)
814
1026
 
815
1027
    def test_extend_dictionary_no_segments(self):
816
1028
        network = dict(name='net_no_segment', id='5', tenant_id='tenant_one')
817
 
        self.driver.type_manager._extend_network_dict_provider(self.context,
818
 
                                                               network)
 
1029
        self.driver.type_manager.extend_network_dict_provider(self.context,
 
1030
                                                              network)
819
1031
        self.assertIsNone(network[pnet.NETWORK_TYPE])
820
1032
        self.assertIsNone(network[pnet.PHYSICAL_NETWORK])
821
1033
        self.assertIsNone(network[pnet.SEGMENTATION_ID])
1080
1292
    def setUp(self):
1081
1293
        super(TestMl2PluginCreateUpdateDeletePort, self).setUp()
1082
1294
        self.context = mock.MagicMock()
 
1295
        self.notify_p = mock.patch('neutron.callbacks.registry.notify')
 
1296
        self.notify = self.notify_p.start()
1083
1297
 
1084
1298
    def _ensure_transaction_is_closed(self):
1085
1299
        transaction = self.context.session.begin(subtransactions=True)
1097
1311
            return_value=new_host_port)
1098
1312
        plugin._check_mac_update_allowed = mock.Mock(return_value=True)
1099
1313
 
1100
 
        plugin._notify_l3_agent_new_port = mock.Mock()
1101
 
        plugin._notify_l3_agent_new_port.side_effect = (
1102
 
            lambda c, p: self._ensure_transaction_is_closed())
 
1314
        self.notify.side_effect = (
 
1315
            lambda r, e, t, **kwargs: self._ensure_transaction_is_closed())
1103
1316
 
1104
1317
        return plugin
1105
1318
 
1115
1328
 
1116
1329
            plugin.create_port(self.context, mock.MagicMock())
1117
1330
 
1118
 
            plugin._notify_l3_agent_new_port.assert_called_once_with(
1119
 
                self.context, new_host_port)
 
1331
            kwargs = {'context': self.context, 'port': new_host_port}
 
1332
            self.notify.assert_called_once_with('port', 'after_create',
 
1333
                plugin, **kwargs)
1120
1334
 
1121
1335
    def test_update_port_rpc_outside_transaction(self):
1122
1336
        with contextlib.nested(
1123
1337
            mock.patch.object(ml2_plugin.Ml2Plugin, '__init__'),
1124
1338
            mock.patch.object(base_plugin.NeutronDbPluginV2, 'update_port'),
1125
 
            mock.patch.object(manager.NeutronManager, 'get_service_plugins'),
1126
 
        ) as (init, super_update_port, get_service_plugins):
 
1339
        ) as (init, super_update_port):
1127
1340
            init.return_value = None
1128
 
            l3plugin = mock.Mock()
1129
 
            l3plugin.supported_extension_aliases = [
1130
 
                constants.L3_DISTRIBUTED_EXT_ALIAS,
1131
 
            ]
1132
 
            get_service_plugins.return_value = {
1133
 
                service_constants.L3_ROUTER_NAT: l3plugin,
 
1341
            new_host_port = mock.Mock()
 
1342
            plugin = self._create_plugin_for_create_update_port(new_host_port)
 
1343
 
 
1344
            plugin.update_port(self.context, 'fake_id', mock.MagicMock())
 
1345
 
 
1346
            kwargs = {
 
1347
                'context': self.context,
 
1348
                'port': new_host_port,
 
1349
                'mac_address_updated': True,
1134
1350
            }
1135
 
 
1136
 
            new_host_port = mock.Mock()
1137
 
            plugin = self._create_plugin_for_create_update_port(new_host_port)
1138
 
 
1139
 
            plugin.update_port(self.context, 'fake_id', mock.MagicMock())
1140
 
 
1141
 
            plugin._notify_l3_agent_new_port.assert_called_once_with(
1142
 
                self.context, new_host_port)
1143
 
            l3plugin.dvr_vmarp_table_update.assert_called_once_with(
1144
 
                self.context, mock.ANY, "add")
 
1351
            self.notify.assert_called_once_with('port', 'after_update',
 
1352
                plugin, **kwargs)
1145
1353
 
1146
1354
    def test_vmarp_table_update_outside_of_delete_transaction(self):
1147
1355
        l3plugin = mock.Mock()