← Back to branch summary

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

« back to all changes in this revision

Viewing changes to neutron/tests/unit/nec/test_db.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/nec/test_db.py
1
 
# Copyright 2012 NEC Corporation.  All rights reserved.
2
 
#
3
 
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4
 
#    not use this file except in compliance with the License. You may obtain
5
 
#    a copy of the License at
6
 
#
7
 
#         http://www.apache.org/licenses/LICENSE-2.0
8
 
#
9
 
#    Unless required by applicable law or agreed to in writing, software
10
 
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11
 
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12
 
#    License for the specific language governing permissions and limitations
13
 
#    under the License.
14
 
 
15
 
import contextlib
16
 
import random
17
 
 
18
 
from neutron.common import constants as q_const
19
 
from neutron.openstack.common import uuidutils
20
 
from neutron.plugins.nec.common import exceptions as nexc
21
 
from neutron.plugins.nec.db import api as ndb
22
 
from neutron.tests.unit.nec import test_nec_plugin
23
 
 
24
 
 
25
 
class NECPluginV2DBTestBase(test_nec_plugin.NecPluginV2TestCase):
26
 
    """Class conisting of NECPluginV2 DB unit tests."""
27
 
 
28
 
    def setUp(self):
29
 
        """Setup for tests."""
30
 
        super(NECPluginV2DBTestBase, self).setUp()
31
 
        self.session = self.context.session
32
 
 
33
 
    def get_ofc_item_random_params(self):
34
 
        """create random parameters for ofc_item test."""
35
 
        ofc_id = uuidutils.generate_uuid()
36
 
        neutron_id = uuidutils.generate_uuid()
37
 
        none = uuidutils.generate_uuid()
38
 
        return ofc_id, neutron_id, none
39
 
 
40
 
    @contextlib.contextmanager
41
 
    def portinfo_random_params(self):
42
 
        with self.port() as port:
43
 
            params = {'port_id': port['port']['id'],
44
 
                      'datapath_id': hex(random.randint(0, 0xffffffff)),
45
 
                      'port_no': random.randint(1, 100),
46
 
                      'vlan_id': random.randint(q_const.MIN_VLAN_TAG,
47
 
                                                q_const.MAX_VLAN_TAG),
48
 
                      'mac': ':'.join(["%02x" % random.randint(0, 0xff)
49
 
                                       for x in range(6)])
50
 
                      }
51
 
            yield params
52
 
 
53
 
 
54
 
class NECPluginV2DBOfcMappingTest(NECPluginV2DBTestBase):
55
 
 
56
 
    def test_add_ofc_item(self):
57
 
        """test add OFC item."""
58
 
        o, q, n = self.get_ofc_item_random_params()
59
 
        tenant = ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
60
 
        self.assertEqual(tenant.ofc_id, o)
61
 
        self.assertEqual(tenant.neutron_id, q)
62
 
 
63
 
    def test_add_ofc_item_duplicate_entry(self):
64
 
        o, q, n = self.get_ofc_item_random_params()
65
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
66
 
        self.assertRaises(nexc.NECDBException,
67
 
                          ndb.add_ofc_item,
68
 
                          self.session, 'ofc_tenant', q, o)
69
 
 
70
 
    def test_get_ofc_item(self):
71
 
        o, q, n = self.get_ofc_item_random_params()
72
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
73
 
        tenant = ndb.get_ofc_item(self.session, 'ofc_tenant', q)
74
 
        self.assertEqual(tenant.ofc_id, o)
75
 
        self.assertEqual(tenant.neutron_id, q)
76
 
 
77
 
    def test_get_ofc_item_for_nonexisting_entry(self):
78
 
        self.assertIsNone(
79
 
            ndb.get_ofc_item(self.session, 'ofc_tenant', 'non-exist-id'))
80
 
 
81
 
    def test_get_ofc_id(self):
82
 
        o, q, n = self.get_ofc_item_random_params()
83
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
84
 
        tenant_id = ndb.get_ofc_id(self.session, 'ofc_tenant', q)
85
 
        self.assertEqual(tenant_id, o)
86
 
 
87
 
    def test_get_ofc_id_for_nonexisting_entry(self):
88
 
        self.assertRaises(nexc.OFCMappingNotFound,
89
 
                          ndb.get_ofc_id,
90
 
                          self.session, 'ofc_tenant', 'non-exist-id')
91
 
 
92
 
    def test_exists_ofc_item(self):
93
 
        o, q, n = self.get_ofc_item_random_params()
94
 
        self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
95
 
 
96
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
97
 
        self.assertTrue(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
98
 
 
99
 
        ndb.del_ofc_item(self.session, 'ofc_tenant', q)
100
 
        self.assertFalse(ndb.exists_ofc_item(self.session, 'ofc_tenant', q))
101
 
 
102
 
    def test_find_ofc_item(self):
103
 
        o, q, n = self.get_ofc_item_random_params()
104
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
105
 
        tenant = ndb.find_ofc_item(self.session, 'ofc_tenant', o)
106
 
        self.assertEqual(tenant.ofc_id, o)
107
 
        self.assertEqual(tenant.neutron_id, q)
108
 
 
109
 
    def test_find_ofc_item_for_nonexisting_entry(self):
110
 
        self.assertIsNone(
111
 
            ndb.find_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
112
 
 
113
 
    def test_del_ofc_item(self):
114
 
        o, q, n = self.get_ofc_item_random_params()
115
 
        ndb.add_ofc_item(self.session, 'ofc_tenant', q, o)
116
 
        self.assertTrue(ndb.del_ofc_item(self.session, 'ofc_tenant', q))
117
 
 
118
 
        self.assertIsNone(ndb.get_ofc_item(self.session, 'ofc_tenant', q))
119
 
        self.assertIsNone(ndb.find_ofc_item(self.session, 'ofc_tenant', o))
120
 
 
121
 
    def test_del_ofc_item_for_nonexisting_entry(self):
122
 
        self.assertFalse(
123
 
            ndb.del_ofc_item(self.session, 'ofc_tenant', 'non-existi-id'))
124
 
 
125
 
 
126
 
class NECPluginV2DBPortInfoTest(NECPluginV2DBTestBase):
127
 
 
128
 
    def _compare_portinfo(self, portinfo, expected):
129
 
        self.assertEqual(portinfo.id, expected['port_id'])
130
 
        self.assertEqual(portinfo.datapath_id, expected['datapath_id'])
131
 
        self.assertEqual(portinfo.port_no, expected['port_no'])
132
 
        self.assertEqual(portinfo.vlan_id, expected['vlan_id'])
133
 
        self.assertEqual(portinfo.mac, expected['mac'])
134
 
 
135
 
    def _add_portinfo(self, session, params):
136
 
        return ndb.add_portinfo(session, params['port_id'],
137
 
                                params['datapath_id'], params['port_no'],
138
 
                                params['vlan_id'], params['mac'])
139
 
 
140
 
    def testd_add_portinfo(self):
141
 
        """test add portinfo."""
142
 
        with self.portinfo_random_params() as params:
143
 
            portinfo = self._add_portinfo(self.session, params)
144
 
            self._compare_portinfo(portinfo, params)
145
 
 
146
 
            exception_raised = False
147
 
            try:
148
 
                self._add_portinfo(self.session, params)
149
 
            except nexc.NECDBException:
150
 
                exception_raised = True
151
 
            self.assertTrue(exception_raised)
152
 
 
153
 
    def teste_get_portinfo(self):
154
 
        """test get portinfo."""
155
 
        with self.portinfo_random_params() as params:
156
 
            self._add_portinfo(self.session, params)
157
 
            portinfo = ndb.get_portinfo(self.session, params['port_id'])
158
 
            self._compare_portinfo(portinfo, params)
159
 
 
160
 
            nonexist_id = uuidutils.generate_uuid()
161
 
            portinfo_none = ndb.get_portinfo(self.session, nonexist_id)
162
 
            self.assertIsNone(portinfo_none)
163
 
 
164
 
    def testf_del_portinfo(self):
165
 
        """test delete portinfo."""
166
 
        with self.portinfo_random_params() as params:
167
 
            self._add_portinfo(self.session, params)
168
 
            portinfo = ndb.get_portinfo(self.session, params['port_id'])
169
 
            self.assertEqual(portinfo.id, params['port_id'])
170
 
            ndb.del_portinfo(self.session, params['port_id'])
171
 
            portinfo_none = ndb.get_portinfo(self.session, params['port_id'])
172
 
            self.assertIsNone(portinfo_none)