← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/nuage/test_nuage_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/nuage/test_nuage_plugin.py
11
11
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
12
#    License for the specific language governing permissions and limitations
13
13
#    under the License.
14
 
#
15
 
# @author: Ronak Shah, Aniket Dandekar, Nuage Networks, Alcatel-Lucent USA Inc.
16
14
 
17
15
 
18
16
import contextlib
59
57
_plugin_name = ('%s.NuagePlugin' % NUAGE_PLUGIN_PATH)
60
58
 
61
59
 
 
60
def getNuageClient():
 
61
    server = FAKE_SERVER
 
62
    serverauth = FAKE_SERVER_AUTH
 
63
    serverssl = FAKE_SERVER_SSL
 
64
    base_uri = FAKE_BASE_URI
 
65
    auth_resource = FAKE_AUTH_RESOURCE
 
66
    organization = FAKE_ORGANIZATION
 
67
    nuageclient = fake_nuageclient.FakeNuageClient(server,
 
68
                                                   base_uri,
 
69
                                                   serverssl,
 
70
                                                   serverauth,
 
71
                                                   auth_resource,
 
72
                                                   organization)
 
73
    return nuageclient
 
74
 
 
75
 
62
76
class NuagePluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
63
77
    def setUp(self, plugin=_plugin_name,
64
78
              ext_mgr=None, service_plugins=None):
67
81
            self.skipTest("Nuage Plugin does not support IPV6.")
68
82
 
69
83
        def mock_nuageClient_init(self):
70
 
            server = FAKE_SERVER
71
 
            serverauth = FAKE_SERVER_AUTH
72
 
            serverssl = FAKE_SERVER_SSL
73
 
            base_uri = FAKE_BASE_URI
74
 
            auth_resource = FAKE_AUTH_RESOURCE
75
 
            organization = FAKE_ORGANIZATION
76
 
            self.nuageclient = None
77
 
            self.nuageclient = fake_nuageclient.FakeNuageClient(server,
78
 
                                                                base_uri,
79
 
                                                                serverssl,
80
 
                                                                serverauth,
81
 
                                                                auth_resource,
82
 
                                                                organization)
 
84
            self.nuageclient = getNuageClient()
83
85
 
84
86
        with mock.patch.object(nuage_plugin.NuagePlugin,
85
87
                               'nuageclient_init', new=mock_nuageClient_init):
244
246
                        self.assertEqual(
245
247
                            fip2['floatingip']['port_id'],
246
248
                            body['floatingip']['port_id'])
 
249
                        self._delete('ports', p['port']['id'])
247
250
 
248
251
                    # Test that port has been successfully deleted.
249
252
                    body = self._show('ports', p['port']['id'],
340
343
            subnet_res = subnet_req.get_response(self.api)
341
344
            self.assertEqual(exc.HTTPCreated.code, subnet_res.status_int)
342
345
 
 
346
    def test_delete_subnet_port_exists_returns_409(self):
 
347
        gateway_ip = '10.0.0.1'
 
348
        cidr = '10.0.0.0/24'
 
349
        res = self._create_network(fmt=self.fmt, name='net',
 
350
                                   admin_state_up=True)
 
351
        network = self.deserialize(self.fmt, res)
 
352
        subnet = self._make_subnet(self.fmt, network, gateway_ip,
 
353
                                   cidr, ip_version=4)
 
354
        self._create_port(self.fmt,
 
355
                          network['network']['id'])
 
356
        req = self.new_delete_request('subnets', subnet['subnet']['id'])
 
357
        res = req.get_response(self.api)
 
358
        self.assertEqual(409, res.status_int)
 
359
 
343
360
 
344
361
class TestNuagePluginPortBinding(NuagePluginV2TestCase,
345
362
                                 test_bindings.PortBindingsTestCase):
369
386
class TestNuageL3NatTestCase(NuagePluginV2TestCase,
370
387
                             test_l3_plugin.L3NatDBIntTestCase):
371
388
 
 
389
    def test_update_port_with_assoc_floatingip(self):
 
390
        with self.subnet(cidr='200.0.0.0/24') as public_sub:
 
391
            self._set_net_external(public_sub['subnet']['network_id'])
 
392
            with self.port() as port:
 
393
                p_id = port['port']['id']
 
394
                with self.floatingip_with_assoc(port_id=p_id):
 
395
                    # Update the port with dummy vm info
 
396
                    port_dict = {
 
397
                        'device_id': uuidutils.generate_uuid(),
 
398
                        'device_owner': 'compute:Nova'
 
399
                    }
 
400
                    port = self._update('ports', port['port']['id'],
 
401
                                        {'port': port_dict})
 
402
                    self.assertEqual(port_dict['device_id'],
 
403
                                     port['port']['device_id'])
 
404
 
 
405
    def test_disassociated_floatingip_delete(self):
 
406
        with self.subnet(cidr='200.0.0.0/24') as public_sub:
 
407
            self._set_net_external(public_sub['subnet']['network_id'])
 
408
            with self.port() as port:
 
409
                p_id = port['port']['id']
 
410
                with self.floatingip_with_assoc(port_id=p_id) as fip:
 
411
 
 
412
                    # Disassociate fip from the port
 
413
                    fip = self._update('floatingips', fip['floatingip']['id'],
 
414
                                       {'floatingip': {'port_id': None}})
 
415
                    self.assertIsNone(fip['floatingip']['router_id'])
 
416
 
372
417
    def test_network_update_external_failure(self):
373
418
        self._test_network_update_external_failure()
374
419
 
426
471
    def test_router_update_with_dup_destination_address(self):
427
472
        with self.router() as r:
428
473
            with self.subnet(cidr='10.0.1.0/24') as s:
429
 
                with self.port(subnet=s, do_delete=False) as p:
 
474
                with self.port(subnet=s) as p:
430
475
                    self._router_interface_action('add',
431
476
                                                  r['router']['id'],
432
477
                                                  None,
502
547
class TestNuageProviderNetTestCase(NuagePluginV2TestCase):
503
548
 
504
549
    def test_create_provider_network(self):
505
 
        phy_net = uuidutils.generate_uuid()
 
550
        phys_net = uuidutils.generate_uuid()
506
551
        data = {'network': {'name': 'pnet1',
507
552
                            'tenant_id': 'admin',
508
553
                            pnet.NETWORK_TYPE: 'vlan',
509
 
                            pnet.PHYSICAL_NETWORK: phy_net,
 
554
                            pnet.PHYSICAL_NETWORK: phys_net,
510
555
                            pnet.SEGMENTATION_ID: 123}}
511
556
        network_req = self.new_create_request('networks', data, self.fmt)
512
557
        net = self.deserialize(self.fmt, network_req.get_response(self.api))
513
558
        self.assertEqual('vlan', net['network'][pnet.NETWORK_TYPE])
514
 
        self.assertEqual(phy_net, net['network'][pnet.PHYSICAL_NETWORK])
 
559
        self.assertEqual(phys_net, net['network'][pnet.PHYSICAL_NETWORK])
515
560
        self.assertEqual(123, net['network'][pnet.SEGMENTATION_ID])
516
561
 
517
562
    def test_create_provider_network_no_admin(self):
518
 
        phy_net = uuidutils.generate_uuid()
 
563
        phys_net = uuidutils.generate_uuid()
519
564
        data = {'network': {'name': 'pnet1',
520
565
                            'tenant_id': 'no_admin',
521
566
                            pnet.NETWORK_TYPE: 'vlan',
522
 
                            pnet.PHYSICAL_NETWORK: phy_net,
 
567
                            pnet.PHYSICAL_NETWORK: phys_net,
523
568
                            pnet.SEGMENTATION_ID: 123}}
524
569
        network_req = self.new_create_request('networks', data, self.fmt)
525
570
        network_req.environ['neutron.context'] = context.Context(
527
572
        res = network_req.get_response(self.api)
528
573
        self.assertEqual(exc.HTTPForbidden.code, res.status_int)
529
574
 
 
575
    def test_get_network_for_provider_network(self):
 
576
        phys_net = uuidutils.generate_uuid()
 
577
        data = {'network': {'name': 'pnet1',
 
578
                            'tenant_id': 'admin',
 
579
                            pnet.NETWORK_TYPE: 'vlan',
 
580
                            pnet.PHYSICAL_NETWORK: phys_net,
 
581
                            pnet.SEGMENTATION_ID: 123}}
 
582
        network_req = self.new_create_request('networks', data, self.fmt)
 
583
        res = self.deserialize(self.fmt, network_req.get_response(self.api))
 
584
 
 
585
        get_req = self.new_show_request('networks', res['network']['id'])
 
586
        net = self.deserialize(self.fmt, get_req.get_response(self.api))
 
587
        self.assertEqual('vlan', net['network'][pnet.NETWORK_TYPE])
 
588
        self.assertEqual(phys_net, net['network'][pnet.PHYSICAL_NETWORK])
 
589
        self.assertEqual(123, net['network'][pnet.SEGMENTATION_ID])
 
590
 
 
591
    def test_list_networks_for_provider_network(self):
 
592
        phys_net = uuidutils.generate_uuid()
 
593
        data1 = {'network': {'name': 'pnet1',
 
594
                            'tenant_id': 'admin',
 
595
                            pnet.NETWORK_TYPE: 'vlan',
 
596
                            pnet.PHYSICAL_NETWORK: phys_net,
 
597
                            pnet.SEGMENTATION_ID: 123}}
 
598
        network_req_1 = self.new_create_request('networks', data1, self.fmt)
 
599
        network_req_1.get_response(self.api)
 
600
        data2 = {'network': {'name': 'pnet2',
 
601
                            'tenant_id': 'admin',
 
602
                            pnet.NETWORK_TYPE: 'vlan',
 
603
                            pnet.PHYSICAL_NETWORK: phys_net,
 
604
                            pnet.SEGMENTATION_ID: 234}}
 
605
        network_req_2 = self.new_create_request('networks', data2, self.fmt)
 
606
        network_req_2.get_response(self.api)
 
607
 
 
608
        list_req = self.new_list_request('networks')
 
609
        pnets = self.deserialize(self.fmt, list_req.get_response(self.api))
 
610
        self.assertEqual(2, len(pnets['networks']))
 
611
        self.assertEqual('vlan', pnets['networks'][0][pnet.NETWORK_TYPE])
 
612
        self.assertEqual(phys_net, pnets['networks'][0][pnet.PHYSICAL_NETWORK])
 
613
        self.assertEqual(123, pnets['networks'][0][pnet.SEGMENTATION_ID])
 
614
        self.assertEqual('vlan', pnets['networks'][1][pnet.NETWORK_TYPE])
 
615
        self.assertEqual(phys_net, pnets['networks'][1][pnet.PHYSICAL_NETWORK])
 
616
        self.assertEqual(234, pnets['networks'][1][pnet.SEGMENTATION_ID])
 
617
 
530
618
 
531
619
class TestNuageSecurityGroupTestCase(NuagePluginV2TestCase,
532
620
                                     test_sg.TestSecurityGroups):
541
629
                # The Nuage plugin reserve the first port
542
630
                port = ports['ports'][1]
543
631
                self.assertEqual(1, len(port[ext_sg.SECURITYGROUPS]))
544
 
                self._delete('ports', port['id'])
 
 
b'\\ No newline at end of file'
 
632
                self._delete('ports', port['id'])