← Back to branch summary

~niedbalski/ubuntu/vivid/neutron/fixes-1447803

« back to all changes in this revision

Viewing changes to neutron/tests/unit/test_db_plugin.py

  • Committer: Package Import Robot
  • Author(s): James Page
  • Date: 2014-10-03 18:45:23 UTC
  • mfrom: (1.1.15)
  • Revision ID: package-import@ubuntu.com-20141003184523-4mt6dy1q3j8n30c9
Tags: 1:2014.2~rc1-0ubuntu1
https://launchpad.net/bugs/1323729
* New upstream release candidate:
  - d/p/*: Refreshed.
  - d/control: Add python-requests-mock to BD's.
  - d/control: Align versioned requirements with upstream.
* Transition linuxbridge and openvswitch plugin users to modular
  layer 2 plugin (LP: #1323729):
  - d/control: Mark removed plugin packages as transitional, depend
    on neutron-plugin-ml2, mark oldlibs/extra.
  - d/neutron-plugin-{linuxbridge,openvswitch}.install: Drop.
  - d/control: Depend on neutron-plugin-ml2 for linuxbridge
    agent package.
  - d/neutron-plugin-linuxbridge-agent.upstart: Use ml2 plugin
    configuration files.

Show diffs side-by-side

added added

removed removed

Lines of Context:
neutron/tests/unit/test_db_plugin.py
368
368
 
369
369
    def _list_ports(self, fmt, expected_res_status=None,
370
370
                    net_id=None, **kwargs):
371
 
        query_params = None
 
371
        query_params = []
372
372
        if net_id:
373
 
            query_params = "network_id=%s" % net_id
374
 
        port_req = self.new_list_request('ports', fmt, query_params)
 
373
            query_params.append("network_id=%s" % net_id)
 
374
        if kwargs.get('device_owner'):
 
375
            query_params.append("device_owner=%s" % kwargs.get('device_owner'))
 
376
        port_req = self.new_list_request('ports', fmt, '&'.join(query_params))
375
377
        if ('set_context' in kwargs and
376
378
                kwargs['set_context'] is True and
377
379
                'tenant_id' in kwargs):
518
520
    def network(self, name='net1',
519
521
                admin_state_up=True,
520
522
                fmt=None,
521
 
                do_delete=True,
522
523
                **kwargs):
523
524
        network = self._make_network(fmt or self.fmt, name,
524
525
                                     admin_state_up, **kwargs)
525
526
        yield network
526
 
        if do_delete:
527
 
            self._delete('networks', network['network']['id'])
528
527
 
529
528
    @contextlib.contextmanager
530
529
    def subnet(self, network=None,
537
536
               dns_nameservers=None,
538
537
               host_routes=None,
539
538
               shared=None,
540
 
               do_delete=True,
541
539
               ipv6_ra_mode=None,
542
540
               ipv6_address_mode=None):
543
541
        with optional_ctx(network, self.network) as network_to_use:
554
552
                                       ipv6_ra_mode=ipv6_ra_mode,
555
553
                                       ipv6_address_mode=ipv6_address_mode)
556
554
            yield subnet
557
 
            if do_delete:
558
 
                self._delete('subnets', subnet['subnet']['id'])
559
555
 
560
556
    @contextlib.contextmanager
561
 
    def port(self, subnet=None, fmt=None, do_delete=True,
562
 
             **kwargs):
 
557
    def port(self, subnet=None, fmt=None, **kwargs):
563
558
        with optional_ctx(subnet, self.subnet) as subnet_to_use:
564
559
            net_id = subnet_to_use['subnet']['network_id']
565
560
            port = self._make_port(fmt or self.fmt, net_id, **kwargs)
566
561
            yield port
567
 
            if do_delete:
568
 
                self._delete('ports', port['port']['id'])
569
562
 
570
563
    def _test_list_with_sort(self, resource,
571
564
                             items, sorts, resources=None, query_params=''):
783
776
            self.assertEqual('myname', port['port']['name'])
784
777
 
785
778
    def test_create_port_as_admin(self):
786
 
        with self.network(do_delete=False) as network:
 
779
        with self.network() as network:
787
780
            self._create_port(self.fmt,
788
781
                              network['network']['id'],
789
782
                              webob.exc.HTTPCreated.code,
1059
1052
            self.assertEqual(port['port']['id'], sport['port']['id'])
1060
1053
 
1061
1054
    def test_delete_port(self):
1062
 
        with self.port(do_delete=False) as port:
 
1055
        with self.port() as port:
1063
1056
            self._delete('ports', port['port']['id'])
1064
1057
            self._show('ports', port['port']['id'],
1065
1058
                       expected_code=webob.exc.HTTPNotFound.code)
1316
1309
                res = self._create_port(self.fmt, net_id=net_id, **kwargs)
1317
1310
                self.assertEqual(res.status_int, webob.exc.HTTPConflict.code)
1318
1311
 
1319
 
    def test_requested_subnet_delete(self):
1320
 
        with self.subnet() as subnet:
1321
 
            with self.port(subnet=subnet) as port:
1322
 
                ips = port['port']['fixed_ips']
1323
 
                self.assertEqual(len(ips), 1)
1324
 
                self.assertEqual(ips[0]['ip_address'], '10.0.0.2')
1325
 
                self.assertEqual(ips[0]['subnet_id'], subnet['subnet']['id'])
1326
 
                req = self.new_delete_request('subnet',
1327
 
                                              subnet['subnet']['id'])
1328
 
                res = req.get_response(self.api)
1329
 
                self.assertEqual(res.status_int, webob.exc.HTTPNotFound.code)
1330
 
 
1331
1312
    def test_requested_subnet_id(self):
1332
1313
        with self.subnet() as subnet:
1333
1314
            with self.port(subnet=subnet) as port:
1685
1666
        ctx = context.get_admin_context()
1686
1667
        with self.subnet() as subnet:
1687
1668
            with contextlib.nested(
1688
 
                self.port(subnet=subnet, device_id='owner1', do_delete=False),
1689
 
                self.port(subnet=subnet, device_id='owner1', do_delete=False),
 
1669
                self.port(subnet=subnet, device_id='owner1'),
 
1670
                self.port(subnet=subnet, device_id='owner1'),
1690
1671
                self.port(subnet=subnet, device_id='owner2'),
1691
1672
            ) as (p1, p2, p3):
1692
1673
                network_id = subnet['subnet']['network_id']
1703
1684
        ctx = context.get_admin_context()
1704
1685
        with self.subnet() as subnet:
1705
1686
            with contextlib.nested(
1706
 
                self.port(subnet=subnet, device_id='owner1', do_delete=False),
 
1687
                self.port(subnet=subnet, device_id='owner1'),
1707
1688
                self.port(subnet=subnet, device_id='owner1'),
1708
1689
                self.port(subnet=subnet, device_id='owner2'),
1709
1690
            ) as (p1, p2, p3):
2298
2279
                                         sorted(expected[k]))
2299
2280
                    else:
2300
2281
                        self.assertEqual(subnet['subnet'][k], expected[k])
2301
 
            return subnet
 
2282
        self._delete('subnets', subnet['subnet']['id'])
 
2283
        return subnet
2302
2284
 
2303
2285
    def test_create_subnet(self):
2304
2286
        gateway_ip = '10.0.0.1'
2359
2341
            res = subnet_req.get_response(self.api)
2360
2342
            self.assertEqual(res.status_int, webob.exc.HTTPClientError.code)
2361
2343
 
 
2344
    def test_create_subnet_bad_V4_cidr_prefix_len(self):
 
2345
        with self.network() as network:
 
2346
            data = {'subnet': {'network_id': network['network']['id'],
 
2347
                    'cidr': '0.0.0.0/0',
 
2348
                    'ip_version': '4',
 
2349
                    'tenant_id': network['network']['tenant_id'],
 
2350
                    'gateway_ip': '0.0.0.1'}}
 
2351
            subnet_req = self.new_create_request('subnets', data)
 
2352
            res = subnet_req.get_response(self.api)
 
2353
            self.assertEqual(res.status_int, webob.exc.HTTPClientError.code)
 
2354
 
2362
2355
    def test_create_subnet_bad_V6_cidr(self):
2363
2356
        with self.network() as network:
2364
2357
            data = {'subnet': {'network_id': network['network']['id'],
2440
2433
                    res = self._create_subnet_bulk(self.fmt, 2,
2441
2434
                                                   net['network']['id'],
2442
2435
                                                   'test')
 
2436
                self._delete('networks', net['network']['id'])
2443
2437
                # We expect a 500 as we injected a fault in the plugin
2444
2438
                self._validate_behavior_on_bulk_failure(
2445
2439
                    res, 'subnets', webob.exc.HTTPServerError.code
2543
2537
        with self.network() as network:
2544
2538
            with contextlib.nested(
2545
2539
                self.subnet(network=network),
2546
 
                self.subnet(network=network, cidr='10.0.1.0/24',
2547
 
                            do_delete=False)) as (subnet1, subnet2):
 
2540
                self.subnet(network=network, cidr='10.0.1.0/24'),
 
2541
            ) as (subnet1, subnet2):
2548
2542
                subnet1_id = subnet1['subnet']['id']
2549
2543
                subnet2_id = subnet2['subnet']['id']
2550
2544
                with self.port(
2580
2574
                                set_context=True)
2581
2575
 
2582
2576
    def test_create_subnet_as_admin(self):
2583
 
        with self.network(do_delete=False) as network:
 
2577
        with self.network() as network:
2584
2578
            self._create_subnet(self.fmt,
2585
2579
                                network['network']['id'],
2586
2580
                                '10.0.2.0/24',
3851
3845
            'max_subnet_host_routes',
3852
3846
            n_exc.HostRoutesExhausted)
3853
3847
 
 
3848
    def test_port_prevents_network_deletion(self):
 
3849
        with self.port() as p:
 
3850
            self._delete('networks', p['port']['network_id'],
 
3851
                         expected_code=webob.exc.HTTPConflict.code)
 
3852
 
 
3853
    def test_port_prevents_subnet_deletion(self):
 
3854
        with self.port() as p:
 
3855
            self._delete('subnets', p['port']['fixed_ips'][0]['subnet_id'],
 
3856
                         expected_code=webob.exc.HTTPConflict.code)
 
3857
 
3854
3858
 
3855
3859
class DbModelTestCase(base.BaseTestCase):
3856
3860
    """DB model tests."""
3889
3893
                                   '_rebuild_availability_ranges') as rebuild:
3890
3894
 
3891
3895
                exception = n_exc.IpAddressGenerationFailure(net_id='n')
3892
 
                generate.side_effect = exception
3893
 
 
3894
 
                # I want the side_effect to throw an exception once but I
3895
 
                # didn't see a way to do this.  So, let it throw twice and
3896
 
                # catch the second one.  Check below to ensure that
3897
 
                # _try_generate_ip was called twice.
3898
 
                try:
3899
 
                    db_base_plugin_v2.NeutronDbPluginV2._generate_ip('c', 's')
3900
 
                except n_exc.IpAddressGenerationFailure:
3901
 
                    pass
 
3896
                # fail first call but not second
 
3897
                generate.side_effect = [exception, None]
 
3898
                db_base_plugin_v2.NeutronDbPluginV2._generate_ip('c', 's')
3902
3899
 
3903
3900
        self.assertEqual(2, generate.call_count)
3904
3901
        rebuild.assert_called_once_with('c', 's')