← Back to branch summary

~snaiksat/neutron/sumit-fix-profileclient

« back to all changes in this revision

Viewing changes to quantum/plugins/cisco/db/db_test_plugin.py

persistence of l2network & ucs plugins using mysql
- db_conn.ini - configuration details of making a connection to the database
- db_test_plugin.py - contains abstraction methods for storing database values in a dict and unit test cases for DB testing
- l2network_db.py - db methods for l2network models
- l2network_models.py - class definitions for the l2 network tables
- ucs_db.py - db methods for ucs models
- ucs_models.py - class definition for the ucs tables 
dynamic loading of the 2nd layer plugin db's based on passed arguments 
Create, Delete, Get, Getall, Update database methods at - Quantum, L2Network and Ucs
Unit test cases for create, delete, getall and update operations for L2Network and Ucs plugins
pep8 checks done
branch based off revision 34 plugin-framework

changes after review comments - 
merged the latest changes from plugin-framework branch - revision 39
confirming to the new directory structure and moving all db related modules into cisco/db folder
updated db_test_plugin.py 
 - added import of cisco constants module 
 - added getLogger to LOG for logging component name   
 - updated import module paths for l2network_models/db and ucs_models/db to use the new directory structure
 - updated (rearranged) imports section to obey openstack alphabetical placement convention
updated db_conn.ini 
 - updated database name from cisco_naas to quantum_l2network
unit test cases ran successfully and pep8 checks done again


-------------s This line and the following will be ignored --------------

added:
  quantum/plugins/cisco/db/db_conn.ini
  quantum/plugins/cisco/db/db_test_plugin.py
  quantum/plugins/cisco/db/l2network_db.py
  quantum/plugins/cisco/db/l2network_models.py
  quantum/plugins/cisco/db/ucs_db.py
  quantum/plugins/cisco/db/ucs_models.py
modified:
  quantum/plugins/cisco/README
pending merges:
  rohitagarwalla 2011-07-31 [merge] merged the latest changes from plugin-framework branch - revision 39
    rohitagarwalla 2011-07-29 persistence of l2network & ucs plugins using mysql

Show diffs side-by-side

added added

removed removed

Lines of Context:
quantum/plugins/cisco/db/db_test_plugin.py
 
1
# vim: tabstop=4 shiftwidth=4 softtabstop=4
 
2
 
 
3
# Copyright 2011, Cisco Systems, Inc.
 
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
# @author: Rohit Agarwalla, Cisco Systems, Inc.
 
17
 
 
18
import ConfigParser
 
19
import os
 
20
import logging as LOG
 
21
import unittest
 
22
 
 
23
from optparse import OptionParser
 
24
from quantum.plugins.cisco.common import cisco_constants as const
 
25
 
 
26
import quantum.db.api as db
 
27
import quantum.db.models
 
28
import quantum.plugins.cisco.db.l2network_db as l2network_db
 
29
import quantum.plugins.cisco.db.l2network_models
 
30
 
 
31
CONF_FILE = "db_conn.ini"
 
32
LOG.getLogger(const.LOGGER_COMPONENT_NAME)
 
33
 
 
34
 
 
35
def find_config(basepath):
 
36
    for root, dirs, files in os.walk(basepath):
 
37
        if CONF_FILE in files:
 
38
            return os.path.join(root, CONF_FILE)
 
39
    return None
 
40
 
 
41
 
 
42
def db_conf(configfile=None):
 
43
    config = ConfigParser.ConfigParser()
 
44
    if configfile == None:
 
45
        if os.path.exists(CONF_FILE):
 
46
            configfile = CONF_FILE
 
47
        else:
 
48
            configfile = \
 
49
           find_config(os.path.abspath(os.path.dirname(__file__)))
 
50
    if configfile == None:
 
51
        raise Exception("Configuration file \"%s\" doesn't exist" %
 
52
              (configfile))
 
53
    LOG.debug("Using configuration file: %s" % configfile)
 
54
    config.read(configfile)
 
55
 
 
56
    DB_NAME = config.get("DATABASE", "name")
 
57
    DB_USER = config.get("DATABASE", "user")
 
58
    DB_PASS = config.get("DATABASE", "pass")
 
59
    DB_HOST = config.get("DATABASE", "host")
 
60
    options = {"sql_connection": "mysql://%s:%s@%s/%s" % (DB_USER,
 
61
    DB_PASS, DB_HOST, DB_NAME)}
 
62
    db.configure_db(options)
 
63
 
 
64
 
 
65
class UcsDB(object):
 
66
    def get_all_ucsmbindings(self):
 
67
        bindings = []
 
68
        try:
 
69
            for x in ucs_db.get_all_ucsmbinding():
 
70
                LOG.debug("Getting ucsm binding : %s" % x.ucsm_ip)
 
71
                bind_dict = {}
 
72
                bind_dict["ucsm-ip"] = str(x.ucsm_ip)
 
73
                bind_dict["network-id"] = str(x.network_id)
 
74
                bindings.append(bind_dict)
 
75
        except Exception, e:
 
76
            LOG.error("Failed to get all bindings: %s" % str(e))
 
77
        return bindings
 
78
 
 
79
    def get_ucsmbinding(self, ucsm_ip):
 
80
        binding = []
 
81
        try:
 
82
            for x in ucs_db.get_ucsmbinding(ucsm_ip):
 
83
                LOG.debug("Getting ucsm binding : %s" % x.ucsm_ip)
 
84
                bind_dict = {}
 
85
                bind_dict["ucsm-ip"] = str(res.ucsm_ip)
 
86
                bind_dict["network-id"] = str(res.network_id)
 
87
                binding.append(bind_dict)
 
88
        except Exception, e:
 
89
            LOG.error("Failed to get binding: %s" % str(e))
 
90
        return binding
 
91
 
 
92
    def create_ucsmbinding(self, ucsm_ip, networ_id):
 
93
        bind_dict = {}
 
94
        try:
 
95
            res = ucs_db.add_ucsmbinding(ucsm_ip, networ_id)
 
96
            LOG.debug("Created ucsm binding: %s" % res.ucsm_ip)
 
97
            bind_dict["ucsm-ip"] = str(res.ucsm_ip)
 
98
            bind_dict["network-id"] = str(res.network_id)
 
99
            return bind_dict
 
100
        except Exception, e:
 
101
            LOG.error("Failed to create ucsm binding: %s" % str(e))
 
102
 
 
103
    def delete_ucsmbinding(self, ucsm_ip):
 
104
        try:
 
105
            res = ucs_db.remove_ucsmbinding(ucsm_ip)
 
106
            LOG.debug("Deleted ucsm binding : %s" % res.ucsm_ip)
 
107
            bind_dict = {}
 
108
            bind_dict["ucsm-ip"] = str(res.ucsm_ip)
 
109
            return bind_dict
 
110
        except Exception, e:
 
111
            raise Exception("Failed to delete dynamic vnic: %s" % str(e))
 
112
 
 
113
    def update_ucsmbinding(self, ucsm_ip, network_id):
 
114
        try:
 
115
            res = ucs_db.update_ucsmbinding(ucsm_ip, network_id)
 
116
            LOG.debug("Updating ucsm binding : %s" % res.ucsm_ip)
 
117
            bind_dict = {}
 
118
            bind_dict["ucsm-ip"] = str(res.ucsm_ip)
 
119
            bind_dict["network-id"] = str(res.network_id)
 
120
            return bind_dict
 
121
        except Exception, e:
 
122
            raise Exception("Failed to update dynamic vnic: %s" % str(e))
 
123
 
 
124
    def get_all_dynamicvnics(self):
 
125
        vnics = []
 
126
        try:
 
127
            for x in ucs_db.get_all_dynamicvnics():
 
128
                LOG.debug("Getting dynamic vnic : %s" % x.uuid)
 
129
                vnic_dict = {}
 
130
                vnic_dict["vnic-id"] = str(x.uuid)
 
131
                vnic_dict["device-name"] = x.device_name
 
132
                vnic_dict["blade-id"] = str(x.blade_id)
 
133
                vnics.append(vnic_dict)
 
134
        except Exception, e:
 
135
            LOG.error("Failed to get all dynamic vnics: %s" % str(e))
 
136
        return vnics
 
137
 
 
138
    def get_dynamicvnic(self, vnic_id):
 
139
        vnic = []
 
140
        try:
 
141
            for x in ucs_db.get_dynamicvnic(vnic_id):
 
142
                LOG.debug("Getting dynamic vnic : %s" % x.uuid)
 
143
                vnic_dict = {}
 
144
                vnic_dict["vnic-id"] = str(x.uuid)
 
145
                vnic_dict["device-name"] = x.device_name
 
146
                vnic_dict["blade-id"] = str(x.blade_id)
 
147
                vnic.append(vnic_dict)
 
148
        except Exception, e:
 
149
            LOG.error("Failed to get dynamic vnic: %s" % str(e))
 
150
        return vnic
 
151
 
 
152
    def create_dynamicvnic(self, device_name, blade_id):
 
153
        vnic_dict = {}
 
154
        try:
 
155
            res = ucs_db.add_dynamicvnic(device_name, blade_id)
 
156
            LOG.debug("Created dynamic vnic: %s" % res.uuid)
 
157
            vnic_dict["vnic-id"] = str(res.uuid)
 
158
            vnic_dict["device-name"] = res.device_name
 
159
            vnic_dict["blade-id"] = str(res.blade_id)
 
160
            return vnic_dict
 
161
        except Exception, e:
 
162
            LOG.error("Failed to create dynamic vnic: %s" % str(e))
 
163
 
 
164
    def delete_dynamicvnic(self, vnic_id):
 
165
        try:
 
166
            res = ucs_db.remove_dynamicvnic(vnic_id)
 
167
            LOG.debug("Deleted dynamic vnic : %s" % res.uuid)
 
168
            vnic_dict = {}
 
169
            vnic_dict["vnic-id"] = str(res.uuid)
 
170
            return vnic_dict
 
171
        except Exception, e:
 
172
            raise Exception("Failed to delete dynamic vnic: %s" % str(e))
 
173
 
 
174
    def update_dynamicvnic(self, vnic_id, device_name=None, blade_id=None):
 
175
        try:
 
176
            res = ucs_db.update_dynamicvnic(vnic_id, device_name, blade_id)
 
177
            LOG.debug("Updating dynamic vnic : %s" % res.uuid)
 
178
            vnic_dict = {}
 
179
            vnic_dict["vnic-id"] = str(res.uuid)
 
180
            vnic_dict["device-name"] = res.device_name
 
181
            vnic_dict["blade-id"] = str(res.blade_id)
 
182
            return vnic_dict
 
183
        except Exception, e:
 
184
            raise Exception("Failed to update dynamic vnic: %s" % str(e))
 
185
 
 
186
    def get_all_blades(self):
 
187
        blades = []
 
188
        try:
 
189
            for x in ucs_db.get_all_blades():
 
190
                LOG.debug("Getting blade : %s" % x.uuid)
 
191
                blade_dict = {}
 
192
                blade_dict["blade-id"] = str(x.uuid)
 
193
                blade_dict["mgmt-ip"] = str(x.mgmt_ip)
 
194
                blade_dict["mac-addr"] = str(x.mac_addr)
 
195
                blade_dict["chassis-id"] = str(x.chassis_id)
 
196
                blade_dict["ucsm-ip"] = str(x.ucsm_ip)
 
197
                blades.append(blade_dict)
 
198
        except Exception, e:
 
199
            LOG.error("Failed to get all blades: %s" % str(e))
 
200
        return blades
 
201
 
 
202
    def get_blade(self, blade_id):
 
203
        blade = []
 
204
        try:
 
205
            for x in ucs_db.get_blade(blade_id):
 
206
                LOG.debug("Getting blade : %s" % x.uuid)
 
207
                blade_dict = {}
 
208
                blade_dict["blade-id"] = str(x.uuid)
 
209
                blade_dict["mgmt-ip"] = str(x.mgmt_ip)
 
210
                blade_dict["mac-addr"] = str(x.mac_addr)
 
211
                blade_dict["chassis-id"] = str(x.chassis_id)
 
212
                blade_dict["ucsm-ip"] = str(x.ucsm_ip)
 
213
                blade.append(blade_dict)
 
214
        except Exception, e:
 
215
            LOG.error("Failed to get all blades: %s" % str(e))
 
216
        return blade
 
217
 
 
218
    def create_blade(self, mgmt_ip, mac_addr, chassis_id, ucsm_ip):
 
219
        blade_dict = {}
 
220
        try:
 
221
            res = ucs_db.add_blade(mgmt_ip, mac_addr, chassis_id, ucsm_ip)
 
222
            LOG.debug("Created blade: %s" % res.uuid)
 
223
            blade_dict["blade-id"] = str(res.uuid)
 
224
            blade_dict["mgmt-ip"] = str(res.mgmt_ip)
 
225
            blade_dict["mac-addr"] = str(res.mac_addr)
 
226
            blade_dict["chassis-id"] = str(res.chassis_id)
 
227
            blade_dict["ucsm-ip"] = str(res.ucsm_ip)
 
228
            return blade_dict
 
229
        except Exception, e:
 
230
            LOG.error("Failed to create blade: %s" % str(e))
 
231
 
 
232
    def delete_blade(self, blade_id):
 
233
        try:
 
234
            res = ucs_db.remove_blade(blade_id)
 
235
            LOG.debug("Deleted blade : %s" % res.uuid)
 
236
            blade_dict = {}
 
237
            blade_dict["blade-id"] = str(res.uuid)
 
238
            return blade_dict
 
239
        except Exception, e:
 
240
            raise Exception("Failed to delete blade: %s" % str(e))
 
241
 
 
242
    def update_blade(self, blade_id, mgmt_ip=None, mac_addr=None,\
 
243
                     chassis_id=None, ucsm_ip=None):
 
244
        try:
 
245
            res = ucs_db.update_blade(blade_id, mgmt_ip, mac_addr, \
 
246
                                      chassis_id, ucsm_ip)
 
247
            LOG.debug("Updating blade : %s" % res.uuid)
 
248
            blade_dict = {}
 
249
            blade_dict["blade-id"] = str(res.uuid)
 
250
            blade_dict["mgmt-ip"] = str(res.mgmt_ip)
 
251
            blade_dict["mac-addr"] = str(res.mac_addr)
 
252
            blade_dict["chassis-id"] = str(res.chassis_id)
 
253
            blade_dict["ucsm-ip"] = str(res.ucsm_ip)
 
254
            return blade_dict
 
255
        except Exception, e:
 
256
            raise Exception("Failed to update blade: %s" % str(e))
 
257
 
 
258
    def get_all_port_bindings(self):
 
259
        port_bindings = []
 
260
        try:
 
261
            for x in ucs_db.get_all_portbindings():
 
262
                LOG.debug("Getting port binding for port: %s" % x.port_id)
 
263
                port_bind_dict = {}
 
264
                port_bind_dict["port-id"] = x.port_id
 
265
                port_bind_dict["dynamic-vnic-id"] = str(x.dynamic_vnic_id)
 
266
                port_bind_dict["portprofile-name"] = x.portprofile_name
 
267
                port_bind_dict["vlan-name"] = x.vlan_name
 
268
                port_bind_dict["vlan-id"] = str(x.vlan_id)
 
269
                port_bind_dict["qos"] = x.qos
 
270
                port_bindings.append(port_bind_dict)
 
271
        except Exception, e:
 
272
            LOG.error("Failed to get all port bindings: %s" % str(e))
 
273
        return port_bindings
 
274
 
 
275
    def get_port_binding(self):
 
276
        port_binding = []
 
277
        try:
 
278
            for x in ucs_db.get_portbinding(port_id):
 
279
                LOG.debug("Getting port binding for port: %s" % x.port_id)
 
280
                port_bind_dict = {}
 
281
                port_bind_dict["port-id"] = x.port_id
 
282
                port_bind_dict["dynamic-vnic-id"] = str(x.dynamic_vnic_id)
 
283
                port_bind_dict["portprofile-name"] = x.portprofile_name
 
284
                port_bind_dict["vlan-name"] = x.vlan_name
 
285
                port_bind_dict["vlan-id"] = str(x.vlan_id)
 
286
                port_bind_dict["qos"] = x.qos
 
287
                port_bindings.append(port_bind_dict)
 
288
        except Exception, e:
 
289
            LOG.error("Failed to get port binding: %s" % str(e))
 
290
        return port_binding
 
291
 
 
292
    def create_port_binding(self, port_id, dynamic_vnic_id, portprofile_name, \
 
293
                            vlan_name, vlan_id, qos):
 
294
        port_bind_dict = {}
 
295
        try:
 
296
            res = ucs_db.add_portbinding(port_id, dynamic_vnic_id, \
 
297
                                  portprofile_name, vlan_name, vlan_id, qos)
 
298
            LOG.debug("Created port binding: %s" % res.port_id)
 
299
            port_bind_dict["port-id"] = res.port_id
 
300
            port_bind_dict["dynamic-vnic-id"] = str(res.dynamic_vnic_id)
 
301
            port_bind_dict["portprofile-name"] = res.portprofile_name
 
302
            port_bind_dict["vlan-name"] = res.vlan_name
 
303
            port_bind_dict["vlan-id"] = str(res.vlan_id)
 
304
            port_bind_dict["qos"] = res.qos
 
305
            return port_bind_dict
 
306
        except Exception, e:
 
307
            LOG.error("Failed to create port binding: %s" % str(e))
 
308
 
 
309
    def delete_port_binding(self, port_id):
 
310
        try:
 
311
            res = ucs_db.remove_portbinding(port_id)
 
312
            LOG.debug("Deleted port binding : %s" % res.port_id)
 
313
            port_bind_dict = {}
 
314
            port_bind_dict["port-id"] = res.port_id
 
315
            return port_bind_dict
 
316
        except Exception, e:
 
317
            raise Exception("Failed to delete port profile: %s" % str(e))
 
318
 
 
319
    def update_port_binding(self, port_id, dynamic_vnic_id, \
 
320
                         portprofile_name, vlan_name, vlan_id, qos):
 
321
        try:
 
322
            res = ucs_db.update_portbinding(port_id, dynamic_vnic_id, \
 
323
                               portprofile_name, vlan_name, vlan_id, qos)
 
324
            LOG.debug("Updating port binding: %s" % res.port_id)
 
325
            port_bind_dict = {}
 
326
            port_bind_dict["port-id"] = res.port_id
 
327
            port_bind_dict["dynamic-vnic-id"] = str(res.dynamic_vnic_id)
 
328
            port_bind_dict["portprofile-name"] = res.portprofile_name
 
329
            port_bind_dict["vlan-name"] = res.vlan_name
 
330
            port_bind_dict["vlan-id"] = str(res.vlan_id)
 
331
            port_bind_dict["qos"] = res.qos
 
332
            return port_bind_dict
 
333
        except Exception, e:
 
334
            raise Exception("Failed to update portprofile binding:%s" % str(e))
 
335
 
 
336
 
 
337
class QuantumDB(object):
 
338
    def get_all_networks(self, tenant_id):
 
339
        nets = []
 
340
        try:
 
341
            for x in db.network_list(tenant_id):
 
342
                LOG.debug("Getting network: %s" % x.uuid)
 
343
                net_dict = {}
 
344
                net_dict["tenant-id"] = x.tenant_id
 
345
                net_dict["net-id"] = str(x.uuid)
 
346
                net_dict["net-name"] = x.name
 
347
                nets.append(net_dict)
 
348
        except Exception, e:
 
349
            LOG.error("Failed to get all networks: %s" % str(e))
 
350
        return nets
 
351
 
 
352
    def get_network(self, network_id):
 
353
        net = []
 
354
        try:
 
355
            for x in db.network_get(network_id):
 
356
                LOG.debug("Getting network: %s" % x.uuid)
 
357
                net_dict = {}
 
358
                net_dict["tenant-id"] = x.tenant_id
 
359
                net_dict["net-id"] = str(x.uuid)
 
360
                net_dict["net-name"] = x.name
 
361
                nets.append(net_dict)
 
362
        except Exception, e:
 
363
            LOG.error("Failed to get network: %s" % str(e))
 
364
        return net
 
365
 
 
366
    def create_network(self, tenant_id, net_name):
 
367
        net_dict = {}
 
368
        try:
 
369
            res = db.network_create(tenant_id, net_name)
 
370
            LOG.debug("Created network: %s" % res.uuid)
 
371
            net_dict["tenant-id"] = res.tenant_id
 
372
            net_dict["net-id"] = str(res.uuid)
 
373
            net_dict["net-name"] = res.name
 
374
            return net_dict
 
375
        except Exception, e:
 
376
            LOG.error("Failed to create network: %s" % str(e))
 
377
 
 
378
    def delete_network(self, net_id):
 
379
        try:
 
380
            net = db.network_destroy(net_id)
 
381
            LOG.debug("Deleted network: %s" % net.uuid)
 
382
            net_dict = {}
 
383
            net_dict["net-id"] = str(net.uuid)
 
384
            return net_dict
 
385
        except Exception, e:
 
386
            raise Exception("Failed to delete port: %s" % str(e))
 
387
 
 
388
    def rename_network(self, tenant_id, net_id, new_name):
 
389
        try:
 
390
            net = db.network_rename(net_id, tenant_id, new_name)
 
391
            LOG.debug("Renamed network: %s" % net.uuid)
 
392
            net_dict = {}
 
393
            net_dict["net-id"] = str(net.uuid)
 
394
            net_dict["net-name"] = net.name
 
395
            return net_dict
 
396
        except Exception, e:
 
397
            raise Exception("Failed to rename network: %s" % str(e))
 
398
 
 
399
    def get_all_ports(self, net_id):
 
400
        ports = []
 
401
        try:
 
402
            for x in db.port_list(net_id):
 
403
                LOG.debug("Getting port: %s" % x.uuid)
 
404
                port_dict = {}
 
405
                port_dict["port-id"] = str(x.uuid)
 
406
                port_dict["net-id"] = str(x.network_id)
 
407
                port_dict["int-id"] = x.interface_id
 
408
                port_dict["state"] = x.state
 
409
                ports.append(port_dict)
 
410
            return ports
 
411
        except Exception, e:
 
412
            LOG.error("Failed to get all ports: %s" % str(e))
 
413
 
 
414
    def get_port(self, port_id):
 
415
        port = []
 
416
        try:
 
417
            for x in db.port_get(port_id):
 
418
                LOG.debug("Getting port: %s" % x.uuid)
 
419
                port_dict = {}
 
420
                port_dict["port-id"] = str(x.uuid)
 
421
                port_dict["net-id"] = str(x.network_id)
 
422
                port_dict["int-id"] = x.interface_id
 
423
                port_dict["state"] = x.state
 
424
                port.append(port_dict)
 
425
            return port
 
426
        except Exception, e:
 
427
            LOG.error("Failed to get port: %s" % str(e))
 
428
 
 
429
    def create_port(self, net_id):
 
430
        port_dict = {}
 
431
        try:
 
432
            port = db.port_create(net_id)
 
433
            LOG.debug("Creating port %s" % port.uuid)
 
434
            port_dict["port-id"] = str(port.uuid)
 
435
            port_dict["net-id"] = str(port.network_id)
 
436
            port_dict["int-id"] = port.interface_id
 
437
            port_dict["state"] = port.state
 
438
            return port_dict
 
439
        except Exception, e:
 
440
            LOG.error("Failed to create port: %s" % str(e))
 
441
 
 
442
    def delete_port(self, port_id):
 
443
        try:
 
444
            port = db.port_destroy(port_id)
 
445
            LOG.debug("Deleted port %s" % port.uuid)
 
446
            port_dict = {}
 
447
            port_dict["port-id"] = str(port.uuid)
 
448
            return port_dict
 
449
        except Exception, e:
 
450
            raise Exception("Failed to delete port: %s" % str(e))
 
451
 
 
452
    def update_port(self, port_id, port_state):
 
453
        try:
 
454
            port = db.port_set_state(port_id, port_state)
 
455
            LOG.debug("Updated port %s" % port.uuid)
 
456
            port_dict = {}
 
457
            port_dict["port-id"] = str(port.uuid)
 
458
            port_dict["net-id"] = str(port.network_id)
 
459
            port_dict["int-id"] = port.interface_id
 
460
            port_dict["state"] = port.state
 
461
            return port_dict
 
462
        except Exception, e:
 
463
            raise Exception("Failed to update port state: %s" % str(e))
 
464
 
 
465
 
 
466
class L2networkDB(object):
 
467
    def get_all_vlan_bindings(self):
 
468
        vlans = []
 
469
        try:
 
470
            for x in l2network_db.get_all_vlan_bindings():
 
471
                LOG.debug("Getting vlan bindings for vlan: %s" % x.vlan_id)
 
472
                vlan_dict = {}
 
473
                vlan_dict["vlan-id"] = str(x.vlan_id)
 
474
                vlan_dict["vlan-name"] = x.vlan_name
 
475
                vlan_dict["net-id"] = str(x.network_id)
 
476
                vlans.append(vlan_dict)
 
477
        except Exception, e:
 
478
            LOG.error("Failed to get all vlan bindings: %s" % str(e))
 
479
        return vlans
 
480
 
 
481
    def get_vlan_binding(self, network_id):
 
482
        vlan = []
 
483
        try:
 
484
            for x in l2network_db.get_vlan_binding(network_id):
 
485
                LOG.debug("Getting vlan binding for vlan: %s" % x.vlan_id)
 
486
                vlan_dict = {}
 
487
                vlan_dict["vlan-id"] = str(x.vlan_id)
 
488
                vlan_dict["vlan-name"] = x.vlan_name
 
489
                vlan_dict["net-id"] = str(x.network_id)
 
490
                vlan.append(vlan_dict)
 
491
        except Exception, e:
 
492
            LOG.error("Failed to get vlan binding: %s" % str(e))
 
493
        return vlan
 
494
 
 
495
    def create_vlan_binding(self, vlan_id, vlan_name, network_id):
 
496
        vlan_dict = {}
 
497
        try:
 
498
            res = l2network_db.add_vlan_binding(vlan_id, vlan_name, network_id)
 
499
            LOG.debug("Created vlan binding for vlan: %s" % res.vlan_id)
 
500
            vlan_dict["vlan-id"] = str(res.vlan_id)
 
501
            vlan_dict["vlan-name"] = res.vlan_name
 
502
            vlan_dict["net-id"] = str(res.network_id)
 
503
            return vlan_dict
 
504
        except Exception, e:
 
505
            LOG.error("Failed to create vlan binding: %s" % str(e))
 
506
 
 
507
    def delete_vlan_binding(self, network_id):
 
508
        try:
 
509
            res = l2network_db.remove_vlan_binding(network_id)
 
510
            LOG.debug("Deleted vlan binding for vlan: %s" % res.vlan_id)
 
511
            vlan_dict = {}
 
512
            vlan_dict["vlan-id"] = str(res.vlan_id)
 
513
            return vlan_dict
 
514
        except Exception, e:
 
515
            raise Exception("Failed to delete vlan binding: %s" % str(e))
 
516
 
 
517
    def update_vlan_binding(self, network_id, vlan_id, vlan_name):
 
518
        try:
 
519
            res = l2network_db.update_vlan_binding(network_id, vlan_id, \
 
520
                                                            vlan_name)
 
521
            LOG.debug("Updating vlan binding for vlan: %s" % res.vlan_id)
 
522
            vlan_dict = {}
 
523
            vlan_dict["vlan-id"] = str(res.vlan_id)
 
524
            vlan_dict["vlan-name"] = res.vlan_name
 
525
            vlan_dict["net-id"] = str(res.network_id)
 
526
            return vlan_dict
 
527
        except Exception, e:
 
528
            raise Exception("Failed to update vlan binding: %s" % str(e))
 
529
 
 
530
    def get_all_portprofiles(self):
 
531
        pps = []
 
532
        try:
 
533
            for x in l2network_db.get_all_portprofiles():
 
534
                LOG.debug("Getting port profile : %s" % x.uuid)
 
535
                pp_dict = {}
 
536
                pp_dict["portprofile-id"] = str(x.uuid)
 
537
                pp_dict["portprofile-name"] = x.name
 
538
                pp_dict["vlan-id"] = str(x.vlan_id)
 
539
                pp_dict["qos"] = x.qos
 
540
                pps.append(pp_dict)
 
541
        except Exception, e:
 
542
            LOG.error("Failed to get all port profiles: %s" % str(e))
 
543
        return pps
 
544
 
 
545
    def get_portprofile(self, port_id):
 
546
        pp = []
 
547
        try:
 
548
            for x in l2network_db.get_portprofile(port_id):
 
549
                LOG.debug("Getting port profile : %s" % x.uuid)
 
550
                pp_dict = {}
 
551
                pp_dict["portprofile-id"] = str(x.uuid)
 
552
                pp_dict["portprofile-name"] = x.name
 
553
                pp_dict["vlan-id"] = str(x.vlan_id)
 
554
                pp_dict["qos"] = x.qos
 
555
                pp.append(pp_dict)
 
556
        except Exception, e:
 
557
            LOG.error("Failed to get port profile: %s" % str(e))
 
558
        return pp
 
559
 
 
560
    def create_portprofile(self, name, vlan_id, qos):
 
561
        pp_dict = {}
 
562
        try:
 
563
            res = l2network_db.add_portprofile(name, vlan_id, qos)
 
564
            LOG.debug("Created port profile: %s" % res.uuid)
 
565
            pp_dict["portprofile-id"] = str(res.uuid)
 
566
            pp_dict["portprofile-name"] = res.name
 
567
            pp_dict["vlan-id"] = str(res.vlan_id)
 
568
            pp_dict["qos"] = res.qos
 
569
            return pp_dict
 
570
        except Exception, e:
 
571
            LOG.error("Failed to create port profile: %s" % str(e))
 
572
 
 
573
    def delete_portprofile(self, pp_id):
 
574
        try:
 
575
            res = l2network_db.remove_portprofile(pp_id)
 
576
            LOG.debug("Deleted port profile : %s" % res.uuid)
 
577
            pp_dict = {}
 
578
            pp_dict["pp-id"] = str(res.uuid)
 
579
            return pp_dict
 
580
        except Exception, e:
 
581
            raise Exception("Failed to delete port profile: %s" % str(e))
 
582
 
 
583
    def update_portprofile(self, pp_id, name, vlan_id, qos):
 
584
        try:
 
585
            res = l2network_db.update_portprofile(pp_id, name, vlan_id, qos)
 
586
            LOG.debug("Updating port profile : %s" % res.uuid)
 
587
            pp_dict = {}
 
588
            pp_dict["portprofile-id"] = str(res.uuid)
 
589
            pp_dict["portprofile-name"] = res.name
 
590
            pp_dict["vlan-id"] = str(res.vlan_id)
 
591
            pp_dict["qos"] = res.qos
 
592
            return pp_dict
 
593
        except Exception, e:
 
594
            raise Exception("Failed to update port profile: %s" % str(e))
 
595
 
 
596
    def get_all_pp_bindings(self):
 
597
        pp_bindings = []
 
598
        try:
 
599
            for x in l2network_db.get_all_pp_bindings():
 
600
                LOG.debug("Getting port profile binding: %s" % \
 
601
                                               x.portprofile_id)
 
602
                ppbinding_dict = {}
 
603
                ppbinding_dict["portprofile-id"] = str(x.portprofile_id)
 
604
                ppbinding_dict["net-id"] = str(x.network_id)
 
605
                ppbinding_dict["tenant-id"] = x.tenant_id
 
606
                ppbinding_dict["default"] = x.default
 
607
                pp_bindings.append(ppbinding_dict)
 
608
        except Exception, e:
 
609
            LOG.error("Failed to get all port profiles: %s" % str(e))
 
610
        return pp_bindings
 
611
 
 
612
    def get_pp_binding(self, pp_id):
 
613
        pp_binding = []
 
614
        try:
 
615
            for x in l2network_db.get_pp_binding(pp_id):
 
616
                LOG.debug("Getting port profile binding: %s" % \
 
617
                                                 x.portprofile_id)
 
618
                ppbinding_dict = {}
 
619
                ppbinding_dict["portprofile-id"] = str(x.portprofile_id)
 
620
                ppbinding_dict["net-id"] = str(x.network_id)
 
621
                ppbinding_dict["tenant-id"] = x.tenant_id
 
622
                ppbinding_dict["default"] = x.default
 
623
                pp_bindings.append(ppbinding_dict)
 
624
        except Exception, e:
 
625
            LOG.error("Failed to get port profile binding: %s" % str(e))
 
626
        return pp_binding
 
627
 
 
628
    def create_pp_binding(self, tenant_id, net_id, pp_id, default):
 
629
        ppbinding_dict = {}
 
630
        try:
 
631
            res = l2network_db.add_pp_binding(tenant_id, net_id, pp_id, \
 
632
                                                                default)
 
633
            LOG.debug("Created port profile binding: %s" % res.portprofile_id)
 
634
            ppbinding_dict["portprofile-id"] = str(res.portprofile_id)
 
635
            ppbinding_dict["net-id"] = str(res.network_id)
 
636
            ppbinding_dict["tenant-id"] = res.tenant_id
 
637
            ppbinding_dict["default"] = res.default
 
638
            return ppbinding_dict
 
639
        except Exception, e:
 
640
            LOG.error("Failed to create port profile binding: %s" % str(e))
 
641
 
 
642
    def delete_pp_binding(self, pp_id):
 
643
        try:
 
644
            res = l2network_db.remove_pp_binding(pp_id)
 
645
            LOG.debug("Deleted port profile binding : %s" % res.portprofile_id)
 
646
            ppbinding_dict = {}
 
647
            ppbinding_dict["portprofile-id"] = str(res.portprofile_id)
 
648
            return ppbinding_dict
 
649
        except Exception, e:
 
650
            raise Exception("Failed to delete port profile: %s" % str(e))
 
651
 
 
652
    def update_pp_binding(self, pp_id, tenant_id, net_id, default):
 
653
        try:
 
654
            res = l2network_db.update_pp_binding(pp_id, tenant_id, net_id,\
 
655
                                                                   default)
 
656
            LOG.debug("Updating port profile binding: %s" % res.portprofile_id)
 
657
            ppbinding_dict = {}
 
658
            ppbinding_dict["portprofile-id"] = str(res.portprofile_id)
 
659
            ppbinding_dict["net-id"] = str(res.network_id)
 
660
            ppbinding_dict["tenant-id"] = res.tenant_id
 
661
            ppbinding_dict["default"] = res.default
 
662
            return ppbinding_dict
 
663
        except Exception, e:
 
664
            raise Exception("Failed to update portprofile binding:%s" % str(e))
 
665
 
 
666
 
 
667
class UcsDBTest(unittest.TestCase):
 
668
    def setUp(self):
 
669
        self.dbtest = UcsDB()
 
670
        LOG.debug("Setup")
 
671
 
 
672
    def testACreateUcsmBinding(self):
 
673
        binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", "net1")
 
674
        self.assertTrue(binding1["ucsm-ip"] == "1.2.3.4")
 
675
        self.tearDownUcsmBinding()
 
676
 
 
677
    def testBGetAllUcsmBindings(self):
 
678
        binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", "net1")
 
679
        binding2 = self.dbtest.create_ucsmbinding("2.3.4.5", "net1")
 
680
        bindings = self.dbtest.get_all_ucsmbindings()
 
681
        count = 0
 
682
        for x in bindings:
 
683
            if "net" in x["network-id"]:
 
684
                count += 1
 
685
        self.assertTrue(count == 2)
 
686
        self.tearDownUcsmBinding()
 
687
 
 
688
    def testCDeleteUcsmBinding(self):
 
689
        binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", "net1")
 
690
        self.dbtest.delete_ucsmbinding(binding1["ucsm-ip"])
 
691
        bindings = self.dbtest.get_all_ucsmbindings()
 
692
        count = 0
 
693
        for x in bindings:
 
694
            if "net " in x["network-id"]:
 
695
                count += 1
 
696
        self.assertTrue(count == 0)
 
697
        self.tearDownUcsmBinding()
 
698
 
 
699
    def testDUpdateUcsmBinding(self):
 
700
        binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", "net1")
 
701
        binding1 = self.dbtest.update_ucsmbinding(binding1["ucsm-ip"], \
 
702
                                                             "newnet1")
 
703
        bindings = self.dbtest.get_all_ucsmbindings()
 
704
        count = 0
 
705
        for x in bindings:
 
706
            if "new" in x["network-id"]:
 
707
                count += 1
 
708
        self.assertTrue(count == 1)
 
709
        self.tearDownUcsmBinding()
 
710
 
 
711
    def testECreateDynamicVnic(self):
 
712
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
713
                                                               "9.8.7.6")
 
714
        vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"])
 
715
        self.assertTrue(vnic1["device-name"] == "eth1")
 
716
        self.tearDownDyanmicVnic()
 
717
 
 
718
    def testFGetAllDyanmicVnics(self):
 
719
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
720
                                                                "9.8.7.6")
 
721
        vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"])
 
722
        vnic2 = self.dbtest.create_dynamicvnic("eth2", blade1["blade-id"])
 
723
        vnics = self.dbtest.get_all_dynamicvnics()
 
724
        count = 0
 
725
        for x in vnics:
 
726
            if "eth" in x["device-name"]:
 
727
                count += 1
 
728
        self.assertTrue(count == 2)
 
729
        self.tearDownDyanmicVnic()
 
730
 
 
731
    def testGDeleteDyanmicVnic(self):
 
732
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
733
                                                               "9.8.7.6")
 
734
        vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"])
 
735
        self.dbtest.delete_dynamicvnic(vnic1["vnic-id"])
 
736
        vnics = self.dbtest.get_all_dynamicvnics()
 
737
        count = 0
 
738
        for x in vnics:
 
739
            if "eth " in x["device-name"]:
 
740
                count += 1
 
741
        self.assertTrue(count == 0)
 
742
        self.tearDownDyanmicVnic()
 
743
 
 
744
    def testHUpdateDynamicVnic(self):
 
745
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
746
                                                                "9.8.7.6")
 
747
        vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"])
 
748
        vnic1 = self.dbtest.update_dynamicvnic(vnic1["vnic-id"], "neweth1", \
 
749
                                                              "newblade2")
 
750
        vnics = self.dbtest.get_all_dynamicvnics()
 
751
        count = 0
 
752
        for x in vnics:
 
753
            if "new" in x["device-name"]:
 
754
                count += 1
 
755
        self.assertTrue(count == 1)
 
756
        self.tearDownDyanmicVnic()
 
757
 
 
758
    def testICreateUcsBlade(self):
 
759
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
760
                                                              "9.8.7.6")
 
761
        self.assertTrue(blade1["mgmt-ip"] == "1.2.3.4")
 
762
        self.tearDownUcsBlade()
 
763
 
 
764
    def testJGetAllUcsBlade(self):
 
765
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
766
                                                                 "9.8.7.6")
 
767
        blade2 = self.dbtest.create_blade("2.3.4.5", "efgh", "chassis1", \
 
768
                                                                 "9.8.7.6")
 
769
        blades = self.dbtest.get_all_blades()
 
770
        count = 0
 
771
        for x in blades:
 
772
            if "chassis" in x["chassis-id"]:
 
773
                count += 1
 
774
        self.assertTrue(count == 2)
 
775
        self.tearDownUcsBlade()
 
776
 
 
777
    def testKDeleteUcsBlade(self):
 
778
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
779
                                                                 "9.8.7.6")
 
780
        self.dbtest.delete_blade(blade1["blade-id"])
 
781
        blades = self.dbtest.get_all_blades()
 
782
        count = 0
 
783
        for x in blades:
 
784
            if "chassis " in x["chassis-id"]:
 
785
                count += 1
 
786
        self.assertTrue(count == 0)
 
787
        self.tearDownUcsBlade()
 
788
 
 
789
    def testLUpdateUcsBlade(self):
 
790
        blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1", \
 
791
                                                                "9.8.7.6")
 
792
        blade2 = self.dbtest.update_blade(blade1["blade-id"], "2.3.4.5", \
 
793
                                          "newabcd", "chassis1", "9.8.7.6")
 
794
        blades = self.dbtest.get_all_blades()
 
795
        count = 0
 
796
        for x in blades:
 
797
            if "new" in x["mac-addr"]:
 
798
                count += 1
 
799
        self.assertTrue(count == 1)
 
800
        self.tearDownUcsBlade()
 
801
 
 
802
    def testMCreatePortBinding(self):
 
803
        port_bind1 = self.dbtest.create_port_binding("port1", "dv1", "pp1", \
 
804
                                                        "vlan1", 10, "qos1")
 
805
        self.assertTrue(port_bind1["port-id"] == "port1")
 
806
        self.tearDownPortBinding()
 
807
 
 
808
    def testNGetAllPortBinding(self):
 
809
        port_bind1 = self.dbtest.create_port_binding("port1", "dv1", "pp1", \
 
810
                                                         "vlan1", 10, "qos1")
 
811
        port_bind2 = self.dbtest.create_port_binding("port2", "dv2", "pp2", \
 
812
                                                         "vlan2", 20, "qos2")
 
813
        port_bindings = self.dbtest.get_all_port_bindings()
 
814
        count = 0
 
815
        for x in port_bindings:
 
816
            if "port" in x["port-id"]:
 
817
                count += 1
 
818
        self.assertTrue(count == 2)
 
819
        self.tearDownPortBinding()
 
820
 
 
821
    def testODeletePortBinding(self):
 
822
        port_bind1 = self.dbtest.create_port_binding("port1", "dv1", "pp1", \
 
823
                                                     "vlan1", 10, "qos1")
 
824
        self.dbtest.delete_port_binding("port1")
 
825
        port_bindings = self.dbtest.get_all_port_bindings()
 
826
        count = 0
 
827
        for x in port_bindings:
 
828
            if "port " in x["port-id"]:
 
829
                count += 1
 
830
        self.assertTrue(count == 0)
 
831
        self.tearDownPortBinding()
 
832
 
 
833
    def testPUpdatePortBinding(self):
 
834
        port_bind1 = self.dbtest.create_port_binding("port1", "dv1", "pp1", \
 
835
                                                     "vlan1", 10, "qos1")
 
836
        port_bind1 = self.dbtest.update_port_binding("port1", "newdv1", \
 
837
                                         "newpp1", "newvlan1", 11, "newqos1")
 
838
        port_bindings = self.dbtest.get_all_port_bindings()
 
839
        count = 0
 
840
        for x in port_bindings:
 
841
            if "new" in x["dynamic-vnic-id"]:
 
842
                count += 1
 
843
        self.assertTrue(count == 1)
 
844
        self.tearDownPortBinding()
 
845
 
 
846
    def tearDownUcsmBinding(self):
 
847
        print "Tearing Down Ucsm Bindings"
 
848
        binds = self.dbtest.get_all_ucsmbindings()
 
849
        for bind in binds:
 
850
            ip = bind["ucsm-ip"]
 
851
            self.dbtest.delete_ucsmbinding(ip)
 
852
 
 
853
    def tearDownDyanmicVnic(self):
 
854
        print "Tearing Down Dynamic Vnics"
 
855
        vnics = self.dbtest.get_all_dynamicvnics()
 
856
        for vnic in vnics:
 
857
            id = vnic["vnic-id"]
 
858
            self.dbtest.delete_dynamicvnic(id)
 
859
        self.tearDownUcsBlade()
 
860
 
 
861
    def tearDownUcsBlade(self):
 
862
        print "Tearing Down Blades"
 
863
        blades = self.dbtest.get_all_blades()
 
864
        for blade in blades:
 
865
            id = blade["blade-id"]
 
866
            self.dbtest.delete_blade(id)
 
867
 
 
868
    def tearDownPortBinding(self):
 
869
        print "Tearing Down Port Binding"
 
870
        port_bindings = self.dbtest.get_all_port_bindings()
 
871
        for port_binding in port_bindings:
 
872
            id = port_binding["port-id"]
 
873
            self.dbtest.delete_port_binding(id)
 
874
 
 
875
 
 
876
class L2networkDBTest(unittest.TestCase):
 
877
    def setUp(self):
 
878
        self.dbtest = L2networkDB()
 
879
        LOG.debug("Setup")
 
880
 
 
881
    def testACreateVlanBinding(self):
 
882
        vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", "netid1")
 
883
        self.assertTrue(vlan1["vlan-id"] == "10")
 
884
        self.tearDownVlanBinding()
 
885
 
 
886
    def testBGetAllVlanBindings(self):
 
887
        vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", "netid1")
 
888
        vlan2 = self.dbtest.create_vlan_binding(20, "vlan2", "netid2")
 
889
        vlans = self.dbtest.get_all_vlan_bindings()
 
890
        count = 0
 
891
        for x in vlans:
 
892
            if "netid" in x["net-id"]:
 
893
                count += 1
 
894
        self.assertTrue(count == 2)
 
895
        self.tearDownVlanBinding()
 
896
 
 
897
    def testCDeleteVlanBinding(self):
 
898
        vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", "netid1")
 
899
        self.dbtest.delete_vlan_binding("netid1")
 
900
        vlans = self.dbtest.get_all_vlan_bindings()
 
901
        count = 0
 
902
        for x in vlans:
 
903
            if "netid " in x["net-id"]:
 
904
                count += 1
 
905
        self.assertTrue(count == 0)
 
906
        self.tearDownVlanBinding()
 
907
 
 
908
    def testDUpdateVlanBinding(self):
 
909
        vlan1 = self.dbtest.create_vlan_binding(10, "vlan1", "netid1")
 
910
        vlan1 = self.dbtest.update_vlan_binding("netid1", 11, "newvlan1")
 
911
        vlans = self.dbtest.get_all_vlan_bindings()
 
912
        count = 0
 
913
        for x in vlans:
 
914
            if "new" in x["vlan-name"]:
 
915
                count += 1
 
916
        self.assertTrue(count == 1)
 
917
        self.tearDownVlanBinding()
 
918
 
 
919
    def testICreatePortProfile(self):
 
920
        pp1 = self.dbtest.create_portprofile("portprofile1", 10, "qos1")
 
921
        self.assertTrue(pp1["portprofile-name"] == "portprofile1")
 
922
        self.tearDownPortProfile()
 
923
 
 
924
    def testJGetAllPortProfile(self):
 
925
        pp1 = self.dbtest.create_portprofile("portprofile1", 10, "qos1")
 
926
        pp2 = self.dbtest.create_portprofile("portprofile2", 20, "qos2")
 
927
        pps = self.dbtest.get_all_portprofiles()
 
928
        count = 0
 
929
        for x in pps:
 
930
            if "portprofile" in x["portprofile-name"]:
 
931
                count += 1
 
932
        self.assertTrue(count == 2)
 
933
        self.tearDownPortProfile()
 
934
 
 
935
    def testKDeletePortProfile(self):
 
936
        pp1 = self.dbtest.create_portprofile("portprofile1", 10, "qos1")
 
937
        self.dbtest.delete_portprofile(pp1["portprofile-id"])
 
938
        pps = self.dbtest.get_all_portprofiles()
 
939
        count = 0
 
940
        for x in pps:
 
941
            if "portprofile " in x["portprofile-name"]:
 
942
                count += 1
 
943
        self.assertTrue(count == 0)
 
944
        self.tearDownPortProfile()
 
945
 
 
946
    def testLUpdatePortProfile(self):
 
947
        pp1 = self.dbtest.create_portprofile("portprofile1", 10, "qos1")
 
948
        pp1 = self.dbtest.update_portprofile(pp1["portprofile-id"], \
 
949
                                          "newportprofile1", 20, "qos2")
 
950
        pps = self.dbtest.get_all_portprofiles()
 
951
        count = 0
 
952
        for x in pps:
 
953
            if "new" in x["portprofile-name"]:
 
954
                count += 1
 
955
        self.assertTrue(count == 1)
 
956
        self.tearDownPortProfile()
 
957
 
 
958
    def testMCreatePortProfileBinding(self):
 
959
        pp_binding1 = self.dbtest.create_pp_binding("t1", "net1", \
 
960
                                                    "portprofile1", "0")
 
961
        self.assertTrue(pp_binding1["portprofile-id"] == "portprofile1")
 
962
        self.tearDownPortProfileBinding()
 
963
 
 
964
    def testNGetAllPortProfileBinding(self):
 
965
        pp_binding1 = self.dbtest.create_pp_binding("t1", "net1", \
 
966
                                                     "portprofile1", "0")
 
967
        pp_binding2 = self.dbtest.create_pp_binding("t2", "net2", \
 
968
                                                     "portprofile2", "0")
 
969
        pp_bindings = self.dbtest.get_all_pp_bindings()
 
970
        count = 0
 
971
        for x in pp_bindings:
 
972
            if "portprofile" in x["portprofile-id"]:
 
973
                count += 1
 
974
        self.assertTrue(count == 2)
 
975
        self.tearDownPortProfileBinding()
 
976
 
 
977
    def testODeletePortProfileBinding(self):
 
978
        pp_binding1 = self.dbtest.create_pp_binding("t1", "net1", \
 
979
                                                     "portprofile1", "0")
 
980
        self.dbtest.delete_pp_binding(pp_binding1["portprofile-id"])
 
981
        pp_bindings = self.dbtest.get_all_pp_bindings()
 
982
        count = 0
 
983
        for x in pp_bindings:
 
984
            if "portprofile " in x["portprofile-id"]:
 
985
                count += 1
 
986
        self.assertTrue(count == 0)
 
987
        self.tearDownPortProfileBinding()
 
988
 
 
989
    def testPUpdatePortProfileBinding(self):
 
990
        pp_binding1 = self.dbtest.create_pp_binding("t1", "net1", \
 
991
                                                      "portprofile1", "0")
 
992
        pp_binding1 = self.dbtest.update_pp_binding("portprofile1", \
 
993
                                                  "newt1", "newnet1", "1")
 
994
        pp_bindings = self.dbtest.get_all_pp_bindings()
 
995
        count = 0
 
996
        for x in pp_bindings:
 
997
            if "new" in x["net-id"]:
 
998
                count += 1
 
999
        self.assertTrue(count == 1)
 
1000
        self.tearDownPortProfileBinding()
 
1001
 
 
1002
    def tearDownVlanBinding(self):
 
1003
        print "Tearing Down Vlan Binding"
 
1004
        vlans = self.dbtest.get_all_vlan_bindings()
 
1005
        for vlan in vlans:
 
1006
            id = vlan["net-id"]
 
1007
            self.dbtest.delete_vlan_binding(id)
 
1008
 
 
1009
    def tearDownPortProfile(self):
 
1010
        print "Tearing Down Port Profile"
 
1011
        pps = self.dbtest.get_all_portprofiles()
 
1012
        for pp in pps:
 
1013
            id = pp["portprofile-id"]
 
1014
            self.dbtest.delete_portprofile(id)
 
1015
 
 
1016
    def tearDownPortProfileBinding(self):
 
1017
        print "Tearing Down Port Profile Binding"
 
1018
        pp_bindings = self.dbtest.get_all_pp_bindings()
 
1019
        for pp_binding in pp_bindings:
 
1020
            id = pp_binding["portprofile-id"]
 
1021
            self.dbtest.delete_pp_binding(id)
 
1022
 
 
1023
if __name__ == "__main__":
 
1024
    usagestr = "Usage: %prog [OPTIONS] <command> [args]"
 
1025
    parser = OptionParser(usage=usagestr)
 
1026
    parser.add_option("-v", "--verbose", dest="verbose",
 
1027
      action="store_true", default=False, help="turn on verbose logging")
 
1028
 
 
1029
    options, args = parser.parse_args()
 
1030
 
 
1031
    if options.verbose:
 
1032
        LOG.basicConfig(level=LOG.DEBUG)
 
1033
    else:
 
1034
        LOG.basicConfig(level=LOG.WARN)
 
1035
 
 
1036
    #load the models and db based on the 2nd level plugin argument
 
1037
    if args[0] == "ucs":
 
1038
        ucs_db = __import__("quantum.plugins.cisco.db.ucs_db", \
 
1039
                fromlist=["ucs_db"])
 
1040
        ucs_model = __import__("quantum.plugins.cisco.db.ucs_models", \
 
1041
                fromlist=["ucs_models"])
 
1042
 
 
1043
    db_conf()
 
1044
 
 
1045
    # Run the tests
 
1046
    suite = unittest.TestLoader().loadTestsFromTestCase(L2networkDBTest)
 
1047
    unittest.TextTestRunner(verbosity=2).run(suite)
 
1048
    suite = unittest.TestLoader().loadTestsFromTestCase(UcsDBTest)
 
1049
    unittest.TextTestRunner(verbosity=2).run(suite)