← Back to branch summary

~zulcss/ubuntu/precise/quantum/trunk

« back to all changes in this revision

Viewing changes to quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py

  • Committer: Chuck Short
  • Date: 2012-11-26 19:51:11 UTC
  • mfrom: (26.1.1 raring-proposed)
  • Revision ID: zulcss@ubuntu.com-20121126195111-jnz2cr4xi6whemw2
* New upstream release for the Ubuntu Cloud Archive.
* debian/patches/*: Refreshed for opening of Grizzly.
* New upstream release.
* debian/rules: FTFBS if there is missing binaries.
* debian/quantum-server.install: Add quantum-debug.

Show diffs side-by-side

added added

removed removed

Lines of Context:
quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py
1
 
# vim: tabstop=4 shiftwidth=4 softtabstop=4
2
 
#
3
 
# Copyright 2011 Cisco Systems, Inc.  All rights reserved.
4
 
#
5
 
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
6
 
#    not use this file except in compliance with the License. You may obtain
7
 
#    a copy of the License at
8
 
#
9
 
#         http://www.apache.org/licenses/LICENSE-2.0
10
 
#
11
 
#    Unless required by applicable law or agreed to in writing, software
12
 
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
13
 
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
14
 
#    License for the specific language governing permissions and limitations
15
 
#    under the License.
16
 
#
17
 
# @author: Shubhangi Satras, Cisco Systems, Inc.
18
 
# @author: Peter Strunk, Cisco Systems, Inc.
19
 
# @author: Atul Gaikad, Cisco Systems, Inc.
20
 
# @author: Tyler Smith, Cisco Systems, Inc.
21
 
 
22
 
import logging
23
 
import unittest
24
 
 
25
 
from quantum.common import exceptions as exc
26
 
from quantum.openstack.common import importutils
27
 
from quantum.plugins.cisco.common import cisco_constants as const
28
 
from quantum.plugins.cisco.common import cisco_credentials as creds
29
 
from quantum.plugins.cisco.db import api as db
30
 
from quantum.plugins.cisco.db import l2network_db as cdb
31
 
from quantum.plugins.cisco import l2network_plugin_configuration as conf
32
 
from quantum.plugins.cisco.models import l2network_multi_blade
33
 
 
34
 
 
35
 
logging.basicConfig(level=logging.WARN)
36
 
LOG = logging.getLogger(__name__)
37
 
 
38
 
 
39
 
# Set some data to use in tests
40
 
tenant_id = "network_admin"
41
 
net_name = "TestNetwork1"
42
 
new_net_name = "NewTestNetwork1"
43
 
net_id = "44"
44
 
port_id = "p0005"
45
 
port_state = const.PORT_UP
46
 
interface_id = "vif-01"
47
 
vlan_id = "102"
48
 
 
49
 
 
50
 
def vlan_name(id):
51
 
    return "q-%svlan" % id[0:10]
52
 
 
53
 
 
54
 
class TestMultiBlade(unittest.TestCase):
55
 
    """
56
 
    Tests for the multi-blade model for the L2Network plugin
57
 
    """
58
 
    _plugins = {}
59
 
    _inventory = {}
60
 
 
61
 
    def setUp(self):
62
 
        """Setup our tests"""
63
 
        # Initialize cdb and credentials
64
 
        db.configure_db({'sql_connection': 'sqlite:///:memory:'})
65
 
        cdb.initialize()
66
 
        creds.Store.initialize()
67
 
 
68
 
        # Create a place a store net and port ids for the druation of the test
69
 
        self.net_id = 0
70
 
        self.port_id = 0
71
 
 
72
 
        # Create the multiblade object
73
 
        self._l2network_multiblade = (
74
 
            l2network_multi_blade.L2NetworkMultiBlade())
75
 
        self.plugin_key = (
76
 
            "quantum.plugins.cisco.ucs.cisco_ucs_plugin.UCSVICPlugin")
77
 
 
78
 
        # Get UCS inventory to make sure all UCSs are affected by tests
79
 
        for key in conf.PLUGINS[const.PLUGINS].keys():
80
 
            if key in conf.PLUGINS[const.INVENTORY].keys():
81
 
                plugin_obj = conf.PLUGINS[const.INVENTORY][key]
82
 
                self._inventory[key] = importutils.import_object(plugin_obj)
83
 
 
84
 
        self.ucs_count = self._inventory['ucs_plugin']._inventory.__len__()
85
 
 
86
 
    def tearDown(self):
87
 
        """Tear down our tests"""
88
 
        try:
89
 
            port = db.port_get(self.net_id, self.port_id)
90
 
            self._l2network_multiblade.delete_port([tenant_id, self.net_id,
91
 
                                                    self.port_id])
92
 
        except exc.NetworkNotFound:
93
 
            # We won't always have a port to remove
94
 
            pass
95
 
        except exc.PortNotFound:
96
 
            # We won't always have a port to remove
97
 
            pass
98
 
 
99
 
        try:
100
 
            net = db.network_get(self.net_id)
101
 
            self._l2network_multiblade.delete_network([tenant_id, self.net_id])
102
 
        except exc.NetworkNotFound:
103
 
            # We won't always have a network to remove
104
 
            pass
105
 
        db.clear_db()
106
 
 
107
 
    def test_create_network(self):
108
 
        """Support for the Quantum core API call"""
109
 
        LOG.debug("test_create_network - START")
110
 
 
111
 
        # Create the network in the test DB, then with the model
112
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
113
 
        networks = self._l2network_multiblade.create_network([
114
 
            tenant_id,
115
 
            net_name,
116
 
            self.net_id,
117
 
            vlan_name(self.net_id),
118
 
            vlan_id,
119
 
        ])
120
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
121
 
 
122
 
        for network in networks:
123
 
            self.assertEqual(network[const.NET_ID], self.net_id)
124
 
            self.assertEqual(network[const.NET_NAME], net_name)
125
 
 
126
 
        LOG.debug("test_create_network - END")
127
 
 
128
 
    def test_delete_network(self):
129
 
        """Support for the Quantum core API call"""
130
 
        LOG.debug("test_delete_network - START")
131
 
 
132
 
        # Create the network in the test DB, then with the model
133
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
134
 
        self._l2network_multiblade.create_network([tenant_id,
135
 
                                                   net_name,
136
 
                                                   self.net_id,
137
 
                                                   vlan_name(self.net_id),
138
 
                                                   vlan_id])
139
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
140
 
 
141
 
        networks = self._l2network_multiblade.delete_network([tenant_id,
142
 
                                                              self.net_id])
143
 
        cdb.remove_vlan_binding(self.net_id)
144
 
        db.network_destroy(self.net_id)
145
 
        for network in networks:
146
 
            self.assertEqual(network[const.NET_ID], self.net_id)
147
 
            self.assertEqual(network[const.NET_NAME], net_name)
148
 
 
149
 
        LOG.debug("test_delete_network - END")
150
 
 
151
 
    def test_delete_networkDNE(self):
152
 
        """Support for the Quantum core API call"""
153
 
        LOG.debug("test_delete_networkDNE - START")
154
 
 
155
 
        self.assertRaises(exc.NetworkNotFound,
156
 
                          self._l2network_multiblade.delete_network,
157
 
                          [tenant_id, net_id])
158
 
 
159
 
        LOG.debug("test_delete_networkDNE - END")
160
 
 
161
 
    def test_update_network(self):
162
 
        """Support for the Quantum core API call"""
163
 
        LOG.debug("test_update_network - START")
164
 
 
165
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
166
 
 
167
 
        self._l2network_multiblade.create_network([tenant_id,
168
 
                                                   net_name,
169
 
                                                   self.net_id,
170
 
                                                   vlan_name(self.net_id),
171
 
                                                   vlan_id])
172
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
173
 
 
174
 
        net_details = db.network_update(self.net_id, tenant_id,
175
 
                                        name=new_net_name)
176
 
        networks = self._l2network_multiblade.update_network([
177
 
            tenant_id,
178
 
            self.net_id,
179
 
            {'name': new_net_name},
180
 
        ])
181
 
 
182
 
        for network in networks:
183
 
            self.assertEqual(network[const.NET_ID], self.net_id)
184
 
            self.assertEqual(network[const.NET_NAME], new_net_name)
185
 
        LOG.debug("test_update_network - END")
186
 
 
187
 
    def test_update_networkDNE(self):
188
 
        """Support for the Quantum core API call"""
189
 
        LOG.debug("test_update_networkDNE - START")
190
 
        self.assertRaises(exc.NetworkNotFound,
191
 
                          self._l2network_multiblade.update_network,
192
 
                          [tenant_id, net_id, {'name': new_net_name}])
193
 
        LOG.debug("test_update_networkDNE - END")
194
 
 
195
 
    def test_get_all_networks(self):
196
 
        """Not implemented for this model"""
197
 
        pass
198
 
 
199
 
    def test_get_network_details(self):
200
 
        """Not implemented for this model"""
201
 
        pass
202
 
 
203
 
    def test_create_port(self):
204
 
        """Support for the Quantum core API call"""
205
 
        LOG.debug("test_create_port - START")
206
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
207
 
        self._l2network_multiblade.create_network([tenant_id,
208
 
                                                   net_name,
209
 
                                                   self.net_id,
210
 
                                                   vlan_name(self.net_id),
211
 
                                                   vlan_id])
212
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
213
 
 
214
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
215
 
        port = self._l2network_multiblade.create_port([tenant_id,
216
 
                                                       self.net_id,
217
 
                                                       port_state,
218
 
                                                       self.port_id])
219
 
 
220
 
        self.assertEqual(self.port_id, port[0][const.PORTID])
221
 
        LOG.debug("test_create_port - END")
222
 
 
223
 
    def test_delete_port(self):
224
 
        """Support for the Quantum core API call"""
225
 
        LOG.debug("test_delete_port - START")
226
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
227
 
        self._l2network_multiblade.create_network([tenant_id,
228
 
                                                   net_name,
229
 
                                                   self.net_id,
230
 
                                                   vlan_name(self.net_id),
231
 
                                                   vlan_id])
232
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
233
 
 
234
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
235
 
        self._l2network_multiblade.create_port([tenant_id,
236
 
                                                self.net_id,
237
 
                                                port_state, self.port_id])
238
 
 
239
 
        port = self._l2network_multiblade.delete_port([tenant_id,
240
 
                                                       self.net_id,
241
 
                                                       self.port_id])
242
 
 
243
 
        self.assertEqual(self.port_id, port[0][const.PORTID])
244
 
 
245
 
        # Recreating port so tear down doesn't cause an error
246
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
247
 
        self._l2network_multiblade.create_port([tenant_id,
248
 
                                                self.net_id,
249
 
                                                port_state, self.port_id])
250
 
 
251
 
        LOG.debug("test_delete_port - END")
252
 
 
253
 
    def test_get_all_ports(self):
254
 
        """Not implemented for this model"""
255
 
        pass
256
 
 
257
 
    def test_update_port(self):
258
 
        """Not implemented for this model"""
259
 
        pass
260
 
 
261
 
    def test_update_portDNE(self):
262
 
        """Not implemented for this model"""
263
 
        pass
264
 
 
265
 
    def test_update_port_networkDNE(self):
266
 
        """Not implemented for this model"""
267
 
        pass
268
 
 
269
 
    def test_port_details(self):
270
 
        """Not implemented for this model"""
271
 
        pass
272
 
 
273
 
    def test_plug_interface(self):
274
 
        """Support for the Quantum core API call"""
275
 
        LOG.debug("test_plug_interface - START")
276
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
277
 
        self._l2network_multiblade.create_network([tenant_id,
278
 
                                                   net_name,
279
 
                                                   self.net_id,
280
 
                                                   vlan_name(self.net_id),
281
 
                                                   vlan_id])
282
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
283
 
 
284
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
285
 
        self._l2network_multiblade.create_port([tenant_id,
286
 
                                                self.net_id,
287
 
                                                port_state, self.port_id])
288
 
 
289
 
        interface = self._l2network_multiblade.plug_interface(
290
 
            [tenant_id, self.net_id, self.port_id, interface_id])
291
 
        port = db.port_set_attachment(self.net_id, self.port_id, interface_id)
292
 
 
293
 
        self.assertEqual(self.port_id, interface[0][const.PORTID])
294
 
        self.assertEqual(port[const.INTERFACEID], interface_id)
295
 
        LOG.debug("test_plug_interface - END")
296
 
 
297
 
    def test_plug_interface_networkDNE(self):
298
 
        """Support for the Quantum core API call"""
299
 
        LOG.debug("test_plug_interface_networkDNE - START")
300
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
301
 
        self._l2network_multiblade.create_network([tenant_id,
302
 
                                                   net_name,
303
 
                                                   self.net_id,
304
 
                                                   vlan_name(self.net_id),
305
 
                                                   vlan_id])
306
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
307
 
 
308
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
309
 
        self._l2network_multiblade.create_port([tenant_id,
310
 
                                                self.net_id,
311
 
                                                port_state, self.port_id])
312
 
 
313
 
        self.assertRaises(exc.NetworkNotFound,
314
 
                          self._l2network_multiblade.plug_interface,
315
 
                          [tenant_id, net_id, self.port_id, interface_id])
316
 
 
317
 
        LOG.debug("test_plug_interface_networkDNE - END")
318
 
 
319
 
    def test_plug_interface_portDNE(self):
320
 
        """Support for the Quantum core API call"""
321
 
        LOG.debug("test_plug_interface_portDNE - START")
322
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
323
 
        self._l2network_multiblade.create_network([tenant_id,
324
 
                                                   net_name,
325
 
                                                   self.net_id,
326
 
                                                   vlan_name(self.net_id),
327
 
                                                   vlan_id])
328
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
329
 
 
330
 
        self.assertRaises(exc.PortNotFound,
331
 
                          self._l2network_multiblade.plug_interface,
332
 
                          [tenant_id, self.net_id, port_id, interface_id])
333
 
 
334
 
        LOG.debug("test_plug_interface_portDNE - START")
335
 
 
336
 
    def test_unplug_interface(self):
337
 
        """Support for the Quantum core API call"""
338
 
        LOG.debug("test_unplug_interface - START")
339
 
        self.net_id = db.network_create(tenant_id, net_name)[const.UUID]
340
 
        self._l2network_multiblade.create_network([tenant_id,
341
 
                                                   net_name,
342
 
                                                   self.net_id,
343
 
                                                   vlan_name(self.net_id),
344
 
                                                   vlan_id])
345
 
        cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id)
346
 
 
347
 
        self.port_id = db.port_create(self.net_id, port_state)[const.UUID]
348
 
        self._l2network_multiblade.create_port([tenant_id,
349
 
                                                self.net_id,
350
 
                                                port_state, self.port_id])
351
 
 
352
 
        self._l2network_multiblade.plug_interface([tenant_id, self.net_id,
353
 
                                                   self.port_id, interface_id])
354
 
        db.port_set_attachment(self.net_id, self.port_id, interface_id)
355
 
        interface = self._l2network_multiblade.unplug_interface([tenant_id,
356
 
                                                                 self.net_id,
357
 
                                                                 self.port_id])
358
 
 
359
 
        self.assertEqual(self.port_id, interface[0][const.PORTID])
360
 
        LOG.debug("test_unplug_interface - END")