x/vmware-nsx
Code Issues Proposed changes

Cleaning (removing) unused code..hooray ! fixes for extension tests

Browse Source
This commit is contained in:
rohitagarwalla 2011-08-25 16:29:30 -07:00
parent 7d7ac483cf
commit b454982d0e
4 changed files with 10 additions and 800 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
quantum/plugins/cisco/db/l2network_models.py
View File

@ -157,7 +157,7 @@ class QoS(BASE, L2NetworkBase):
qos_desc = Column(String(255))
def __init__(self, tenant_id, qos_name, qos_desc):
self.qos_id = uuid.uuid4()
self.qos_id = str(uuid.uuid4())
self.tenant_id = tenant_id
self.qos_name = qos_name
self.qos_desc = qos_desc
@ -178,7 +178,7 @@ class Credential(BASE, L2NetworkBase):
password = Column(String(255))
def __init__(self, tenant_id, credential_name, user_name, password):
self.credential_id = uuid.uuid4()
self.credential_id = str(uuid.uuid4())
self.tenant_id = tenant_id
self.credential_name = credential_name
self.user_name = user_name

256
quantum/plugins/cisco/db/ucs_db.py
View File

@ -25,262 +25,6 @@ from quantum.plugins.cisco.common import cisco_exceptions as c_exc
from quantum.plugins.cisco.db import ucs_models
def get_all_ucsmbinding():
"""Lists all the ucsm bindings"""
LOG.debug("get_all_ucsmbinding() called")
session = db.get_session()
try:
bindings = session.query(ucs_models.UcsmBinding).\
all()
return bindings
except exc.NoResultFound:
return []
def get_ucsmbinding(ucsm_ip):
"""Lists a ucsm binding"""
LOG.debug("get_ucsmbinding() called")
session = db.get_session()
try:
binding = session.query(ucs_models.UcsmBinding).\
filter_by(ucsm_ip=ucsm_ip).\
one()
return binding
except exc.NoResultFound:
raise c_exc.UcsmBindingNotFound(ucsm_ip=ucsm_ip)
def add_ucsmbinding(ucsm_ip, network_id):
"""Adds a ucsm binding"""
LOG.debug("add_ucsmbinding() called")
session = db.get_session()
try:
binding = session.query(ucs_models.UcsmBinding).\
filter_by(ucsm_ip=ucsm_ip).\
one()
raise c_exc.UcsmBindingAlreadyExists(ucsm_ip=ucsm_ip)
except exc.NoResultFound:
binding = ucs_models.UcsmBinding(ucsm_ip, network_id)
session.add(binding)
session.flush()
return binding
def remove_ucsmbinding(ucsm_ip):
"""Removes a ucsm binding"""
LOG.debug("remove_ucsmbinding() called")
session = db.get_session()
try:
binding = session.query(ucs_models.UcsmBinding).\
filter_by(ucsm_ip=ucsm_ip).\
one()
session.delete(binding)
session.flush()
return binding
except exc.NoResultFound:
pass
def update_ucsmbinding(ucsm_ip, new_network_id):
"""Updates ucsm binding"""
LOG.debug("update_ucsmbinding() called")
session = db.get_session()
try:
binding = session.query(ucs_models.UcsmBinding).\
filter_by(ucsm_ip=ucsm_ip).\
one()
if new_network_id:
binding.network_id = new_network_id
session.merge(binding)
session.flush()
return binding
except exc.NoResultFound:
raise c_exc.UcsmBindingNotFound(ucsm_ip=ucsm_ip)
def get_all_dynamicvnics():
"""Lists all the dynamic vnics"""
LOG.debug("get_all_dynamicvnics() called")
session = db.get_session()
try:
vnics = session.query(ucs_models.DynamicVnic).\
all()
return vnics
except exc.NoResultFound:
return []
def get_dynamicvnic(vnic_id):
"""Lists a dynamic vnic"""
LOG.debug("get_dynamicvnic() called")
session = db.get_session()
try:
vnic = session.query(ucs_models.DynamicVnic).\
filter_by(uuid=vnic_id).\
one()
return vnic
except exc.NoResultFound:
raise c_exc.DynamicVnicNotFound(vnic_id=vnic_id)
def add_dynamicvnic(device_name, blade_id, vnic_state):
"""Adds a dynamic vnic"""
LOG.debug("add_dynamicvnic() called")
session = db.get_session()
try:
vnic = session.query(ucs_models.DynamicVnic).\
filter_by(device_name=device_name).\
one()
raise c_exc.DynamicVnicAlreadyExists(device_name=device_name)
except exc.NoResultFound:
vnic = ucs_models.DynamicVnic(device_name, blade_id, vnic_state)
session.add(vnic)
session.flush()
return vnic
def remove_dynamicvnic(vnic_id):
"""Removes a dynamic vnic"""
LOG.debug("remove_dynamicvnic() called")
session = db.get_session()
try:
vnic = session.query(ucs_models.DynamicVnic).\
filter_by(uuid=vnic_id).\
one()
session.delete(vnic)
session.flush()
return vnic
except exc.NoResultFound:
pass
def update_dynamicvnic(vnic_id, new_device_name=None, new_blade_id=None,
vnic_state=None, blade_intf_dn=None,
blade_intf_order=None, blade_int_link_state=None,
blade_intf_oper_state=None, blade_intf_inst_type=None,
blade_intf_reservation=None):
"""Updates dynamic vnic"""
LOG.debug("update_dynamicvnic() called")
session = db.get_session()
try:
vnic = session.query(ucs_models.DynamicVnic).\
filter_by(uuid=vnic_id).\
one()
if new_device_name:
vnic.device_name = new_device_name
if new_blade_id:
vnic.blade_id = new_blade_id
if vnic_state:
vnic.vnic_state = vnic_state
if blade_intf_dn:
vnic.blade_intf_dn = blade_intf_dn
if blade_intf_order:
vnic.blade_intf_order = blade_intf_order
if blade_int_link_state:
vnic.blade_int_link_state = blade_int_link_state
if blade_intf_oper_state:
vnic.blade_intf_oper_state = blade_intf_oper_state
if blade_intf_inst_type:
vnic.blade_intf_inst_type = blade_intf_inst_type
if blade_intf_reservation:
vnic.blade_intf_reservation = blade_intf_reservation
session.merge(vnic)
session.flush()
return vnic
except exc.NoResultFound:
raise c_exc.DynamicVnicNotFound(vnic_id=vnic_id)
def get_all_blades():
"""Lists all the blades details"""
LOG.debug("get_all_blades() called")
session = db.get_session()
try:
blades = session.query(ucs_models.UcsBlade).\
all()
return blades
except exc.NoResultFound:
return []
def get_blade(blade_id):
"""Lists a blade details"""
LOG.debug("get_blade() called")
session = db.get_session()
try:
blade = session.query(ucs_models.UcsBlade).\
filter_by(uuid=blade_id).\
one()
return blade
except exc.NoResultFound:
raise c_exc.BladeNotFound(blade_id=blade_id)
def add_blade(mgmt_ip, mac_addr, chassis_id, ucsm_ip, blade_state,
vnics_used, hostname):
"""Adds a blade"""
LOG.debug("add_blade() called")
session = db.get_session()
try:
blade = session.query(ucs_models.UcsBlade).\
filter_by(mgmt_ip=mgmt_ip).\
one()
raise c_exc.BladeAlreadyExists(mgmt_ip=mgmt_ip)
except exc.NoResultFound:
blade = ucs_models.UcsBlade(mgmt_ip, mac_addr, chassis_id, ucsm_ip,
blade_state, vnics_used, hostname)
session.add(blade)
session.flush()
return blade
def remove_blade(blade_id):
"""Removes a blade"""
LOG.debug("remove_blade() called")
session = db.get_session()
try:
blade = session.query(ucs_models.UcsBlade).\
filter_by(uuid=blade_id).\
one()
session.delete(blade)
session.flush()
return blade
except exc.NoResultFound:
pass
def update_blade(blade_id, new_mgmt_ip=None, new_mac_addr=None,
new_chassis_id=None, new_ucsm_ip=None,
new_blade_state=None, new_vnics_used=None,
new_hostname=None):
"""Updates details of a blade"""
LOG.debug("update_blade() called")
session = db.get_session()
try:
blade = session.query(ucs_models.UcsBlade).\
filter_by(uuid=blade_id).\
one()
if new_mgmt_ip:
blade.mgmt_ip = new_mgmt_ip
if new_mac_addr:
blade.mac_addr = new_mac_addr
if new_chassis_id:
blade.chassis_id = new_chassis_id
if new_ucsm_ip:
blade.ucsm_ip = new_ucsm_ip
if new_blade_state:
blade.blade_state = new_blade_state
if new_vnics_used:
blade.vnics_used = new_vnics_used
if new_hostname:
blade.hostname = new_hostname
session.merge(blade)
session.flush()
return blade
except exc.NoResultFound:
raise c_exc.BladeNotFound(blade_id=blade_id)
def get_all_portbindings():
"""Lists all the port bindings"""
LOG.debug("db get_all_portbindings() called")

83
quantum/plugins/cisco/db/ucs_models.py
View File

@ -15,95 +15,14 @@
# under the License.
# @author: Rohit Agarwalla, Cisco Systems, Inc.
import uuid
from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relation
from quantum.plugins.cisco.db.l2network_models import L2NetworkBase
from quantum.plugins.cisco.db import models
from quantum.plugins.cisco.db.l2network_models import L2NetworkBase
from quantum.plugins.cisco.db.models import BASE
class UcsmBinding(BASE, L2NetworkBase):
"""Represents a binding of ucsm to network_id"""
__tablename__ = 'ucsm_bindings'
id = Column(Integer, primary_key=True, autoincrement=True)
ucsm_ip = Column(String(255))
network_id = Column(String(255), ForeignKey("networks.uuid"),
nullable=False)
network = relation(models.Network)
def __init__(self, ucsm_ip, network_id):
self.ucsm_ip = ucsm_ip
self.network_id = network_id
def __repr__(self):
return "<UcsmBinding(%s,%s)>" % \
(self.ucsm_ip, self.network_id)
class DynamicVnic(BASE, L2NetworkBase):
"""Represents Cisco UCS Dynamic Vnics"""
__tablename__ = 'dynamic_vnics'
uuid = Column(String(255), primary_key=True)
device_name = Column(String(255))
blade_id = Column(String(255), ForeignKey("ucs_blades.uuid"),
nullable=False)
vnic_state = Column(String(255))
blade_intf_dn = Column(String(255))
blade_intf_order = Column(String(255))
blade_int_link_state = Column(String(255))
blade_intf_oper_state = Column(String(255))
blade_intf_inst_type = Column(String(255))
blade_intf_reservation = Column(String(255))
def __init__(self, device_name, blade_id, vnic_state):
self.uuid = uuid.uuid4()
self.device_name = device_name
self.blade_id = blade_id
self.vnic_state = vnic_state
def __repr__(self):
return "<Dyanmic Vnic(%s,%s,%s,%s)>" % \
(self.uuid, self.device_name, self.blade_id,
self.vnic_state)
class UcsBlade(BASE, L2NetworkBase):
"""Represents details of ucs blades"""
__tablename__ = 'ucs_blades'
uuid = Column(String(255), primary_key=True)
mgmt_ip = Column(String(255))
mac_addr = Column(String(255))
chassis_id = Column(String(255))
ucsm_ip = Column(String(255))
blade_state = Column(String(255))
vnics_used = Column(Integer)
hostname = Column(String(255))
dynamic_vnics = relation(DynamicVnic, order_by=DynamicVnic.uuid,
backref="blade")
def __init__(self, mgmt_ip, mac_addr, chassis_id, ucsm_ip,
blade_state, vnics_used, hostname):
self.uuid = uuid.uuid4()
self.mgmt_ip = mgmt_ip
self.mac_addr = mac_addr
self.chassis_id = chassis_id
self.ucsm_ip = ucsm_ip
self.blade_state = blade_state
self.vnics_used = vnics_used
self.hostname = hostname
def __repr__(self):
return "<UcsBlades (%s,%s,%s,%s,%s,%s,%s,%s)>" % \
(self.uuid, self.mgmt_ip, self.mac_addr, self.chassis_id,
self.ucsm_ip, self.blade_state, self.vnics_used, self.hostname)
class PortBinding(BASE, L2NetworkBase):
"""Represents Port binding to device interface"""
__tablename__ = 'port_bindings'

467
quantum/plugins/cisco/tests/unit/test_database.py
View File

@ -35,235 +35,6 @@ LOG.getLogger(const.LOGGER_COMPONENT_NAME)
class UcsDB(object):
"""Class consisting of methods to call ucs db methods"""
def get_all_ucsmbindings(self):
"""get all ucsm bindings"""
bindings = []
try:
for res in ucs_db.get_all_ucsmbinding():
LOG.debug("Getting ucsm binding : %s" % res.ucsm_ip)
bind_dict = {}
bind_dict["ucsm-ip"] = str(res.ucsm_ip)
bind_dict["network-id"] = str(res.network_id)
bindings.append(bind_dict)
except Exception, exc:
LOG.error("Failed to get all bindings: %s" % str(exc))
return bindings
def get_ucsmbinding(self, ucsm_ip):
"""get ucsm binding"""
binding = []
try:
for res in ucs_db.get_ucsmbinding(ucsm_ip):
LOG.debug("Getting ucsm binding : %s" % res.ucsm_ip)
bind_dict = {}
bind_dict["ucsm-ip"] = str(res.ucsm_ip)
bind_dict["network-id"] = str(res.network_id)
binding.append(bind_dict)
except Exception, exc:
LOG.error("Failed to get binding: %s" % str(exc))
return binding
def create_ucsmbinding(self, ucsm_ip, network_id):
"""create ucsm binding"""
bind_dict = {}
try:
res = ucs_db.add_ucsmbinding(ucsm_ip, network_id)
LOG.debug("Created ucsm binding: %s" % res.ucsm_ip)
bind_dict["ucsm-ip"] = str(res.ucsm_ip)
bind_dict["network-id"] = str(res.network_id)
return bind_dict
except Exception, exc:
LOG.error("Failed to create ucsm binding: %s" % str(exc))
def delete_ucsmbinding(self, ucsm_ip):
"""delete ucsm binding"""
try:
res = ucs_db.remove_ucsmbinding(ucsm_ip)
LOG.debug("Deleted ucsm binding : %s" % res.ucsm_ip)
bind_dict = {}
bind_dict["ucsm-ip"] = str(res.ucsm_ip)
return bind_dict
except Exception, exc:
raise Exception("Failed to delete dynamic vnic: %s" % str(exc))
def update_ucsmbinding(self, ucsm_ip, network_id):
"""update ucsm binding"""
try:
res = ucs_db.update_ucsmbinding(ucsm_ip, network_id)
LOG.debug("Updating ucsm binding : %s" % res.ucsm_ip)
bind_dict = {}
bind_dict["ucsm-ip"] = str(res.ucsm_ip)
bind_dict["network-id"] = str(res.network_id)
return bind_dict
except Exception, exc:
raise Exception("Failed to update dynamic vnic: %s" % str(exc))
def get_all_dynamicvnics(self):
"""get all dynamic vnics"""
vnics = []
try:
for res in ucs_db.get_all_dynamicvnics():
LOG.debug("Getting dynamic vnic : %s" % res.uuid)
vnic_dict = {}
vnic_dict["vnic-id"] = str(res.uuid)
vnic_dict["device-name"] = res.device_name
vnic_dict["blade-id"] = str(res.blade_id)
vnic_dict["vnic_state"] = res.vnic_state
vnics.append(vnic_dict)
except Exception, exc:
LOG.error("Failed to get all dynamic vnics: %s" % str(exc))
return vnics
def get_dynamicvnic(self, vnic_id):
"""get dynamic vnic"""
vnic = []
try:
for res in ucs_db.get_dynamicvnic(vnic_id):
LOG.debug("Getting dynamic vnic : %s" % res.uuid)
vnic_dict = {}
vnic_dict["vnic-id"] = str(res.uuid)
vnic_dict["device-name"] = res.device_name
vnic_dict["blade-id"] = str(res.blade_id)
vnic_dict["vnic_state"] = res.vnic_state
vnic.append(vnic_dict)
except Exception, exc:
LOG.error("Failed to get dynamic vnic: %s" % str(exc))
return vnic
def create_dynamicvnic(self, device_name, blade_id, vnic_state):
"""create dynamic vnic"""
vnic_dict = {}
try:
res = ucs_db.add_dynamicvnic(device_name, blade_id, vnic_state)
LOG.debug("Created dynamic vnic: %s" % res.uuid)
vnic_dict["vnic-id"] = str(res.uuid)
vnic_dict["device-name"] = res.device_name
vnic_dict["blade-id"] = str(res.blade_id)
vnic_dict["vnic_state"] = res.vnic_state
return vnic_dict
except Exception, exc:
LOG.error("Failed to create dynamic vnic: %s" % str(exc))
def delete_dynamicvnic(self, vnic_id):
"""delete dynamic vnic"""
try:
res = ucs_db.remove_dynamicvnic(vnic_id)
LOG.debug("Deleted dynamic vnic : %s" % res.uuid)
vnic_dict = {}
vnic_dict["vnic-id"] = str(res.uuid)
return vnic_dict
except Exception, exc:
raise Exception("Failed to delete dynamic vnic: %s" % str(exc))
def update_dynamicvnic(self, vnic_id, device_name=None, blade_id=None,
vnic_state=None):
"""update dynamic vnic"""
try:
res = ucs_db.update_dynamicvnic(vnic_id, device_name, blade_id,
vnic_state)
LOG.debug("Updating dynamic vnic : %s" % res.uuid)
vnic_dict = {}
vnic_dict["vnic-id"] = str(res.uuid)
vnic_dict["device-name"] = res.device_name
vnic_dict["blade-id"] = str(res.blade_id)
vnic_dict["vnic_state"] = res.vnic_state
return vnic_dict
except Exception, exc:
raise Exception("Failed to update dynamic vnic: %s" % str(exc))
def get_all_blades(self):
"""get all blades"""
blades = []
try:
for res in ucs_db.get_all_blades():
LOG.debug("Getting blade : %s" % res.uuid)
blade_dict = {}
blade_dict["blade-id"] = str(res.uuid)
blade_dict["mgmt-ip"] = str(res.mgmt_ip)
blade_dict["mac-addr"] = str(res.mac_addr)
blade_dict["chassis-id"] = str(res.chassis_id)
blade_dict["ucsm-ip"] = str(res.ucsm_ip)
blade_dict["blade_state"] = str(res.blade_state)
blade_dict["vnics_used"] = str(res.vnics_used)
blade_dict["hostname"] = str(res.hostname)
blades.append(blade_dict)
except Exception, exc:
LOG.error("Failed to get all blades: %s" % str(exc))
return blades
def get_blade(self, blade_id):
"""get blade"""
blade = []
try:
for res in ucs_db.get_blade(blade_id):
LOG.debug("Getting blade : %s" % res.uuid)
blade_dict = {}
blade_dict["blade-id"] = str(res.uuid)
blade_dict["mgmt-ip"] = str(res.mgmt_ip)
blade_dict["mac-addr"] = str(res.mac_addr)
blade_dict["chassis-id"] = str(res.chassis_id)
blade_dict["ucsm-ip"] = str(res.ucsm_ip)
blade_dict["blade_state"] = str(res.blade_state)
blade_dict["vnics_used"] = str(res.vnics_used)
blade_dict["hostname"] = str(res.hostname)
blade.append(blade_dict)
except Exception, exc:
LOG.error("Failed to get all blades: %s" % str(exc))
return blade
def create_blade(self, mgmt_ip, mac_addr, chassis_id, ucsm_ip,
blade_state, vnics_used, hostname):
"""create blade"""
blade_dict = {}
try:
res = ucs_db.add_blade(mgmt_ip, mac_addr, chassis_id, ucsm_ip,
blade_state, vnics_used, hostname)
LOG.debug("Created blade: %s" % res.uuid)
blade_dict["blade-id"] = str(res.uuid)
blade_dict["mgmt-ip"] = str(res.mgmt_ip)
blade_dict["mac-addr"] = str(res.mac_addr)
blade_dict["chassis-id"] = str(res.chassis_id)
blade_dict["ucsm-ip"] = str(res.ucsm_ip)
blade_dict["blade_state"] = str(res.blade_state)
blade_dict["vnics_used"] = str(res.vnics_used)
blade_dict["hostname"] = str(res.hostname)
return blade_dict
except Exception, exc:
LOG.error("Failed to create blade: %s" % str(exc))
def delete_blade(self, blade_id):
"""delete blade"""
try:
res = ucs_db.remove_blade(blade_id)
LOG.debug("Deleted blade : %s" % res.uuid)
blade_dict = {}
blade_dict["blade-id"] = str(res.uuid)
return blade_dict
except Exception, exc:
raise Exception("Failed to delete blade: %s" % str(exc))
def update_blade(self, blade_id, mgmt_ip=None, mac_addr=None,
chassis_id=None, ucsm_ip=None, blade_state=None,
vnics_used=None, hostname=None):
"""update blade"""
try:
res = ucs_db.update_blade(blade_id, mgmt_ip, mac_addr,
chassis_id, ucsm_ip, blade_state,
vnics_used, hostname)
LOG.debug("Updating blade : %s" % res.uuid)
blade_dict = {}
blade_dict["blade-id"] = str(res.uuid)
blade_dict["mgmt-ip"] = str(res.mgmt_ip)
blade_dict["mac-addr"] = str(res.mac_addr)
blade_dict["chassis-id"] = str(res.chassis_id)
blade_dict["ucsm-ip"] = str(res.ucsm_ip)
blade_dict["blade_state"] = str(res.blade_state)
blade_dict["vnics_used"] = str(res.vnics_used)
blade_dict["hostname"] = str(res.hostname)
return blade_dict
except Exception, exc:
raise Exception("Failed to update blade: %s" % str(exc))
def get_all_port_bindings(self):
"""get all port binding"""
port_bindings = []
@ -824,177 +595,14 @@ class UcsDBTest(unittest.TestCase):
"""Tear Down"""
db.clear_db()
def testa_create_ucsmbinding(self):
"""create ucsm binding"""
net1 = self.quantum.create_network("t1", "netid1")
binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", net1["net-id"])
self.assertTrue(binding1["ucsm-ip"] == "1.2.3.4")
self.teardown_ucsmbinding()
self.teardown_network()
def testb_getall_ucsmbindings(self):
"""get all ucsm bindings"""
net1 = self.quantum.create_network("t1", "netid1")
binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", net1["net-id"])
binding2 = self.dbtest.create_ucsmbinding("2.3.4.5", net1["net-id"])
bindings = self.dbtest.get_all_ucsmbindings()
count = 0
for bind in bindings:
if net1["net-id"] == bind["network-id"]:
count += 1
self.assertTrue(count == 2)
self.teardown_ucsmbinding()
self.teardown_network()
def testc_delete_ucsmbinding(self):
"""delete ucsm binding"""
net1 = self.quantum.create_network("t1", "netid1")
binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", net1["net-id"])
self.dbtest.delete_ucsmbinding(binding1["ucsm-ip"])
bindings = self.dbtest.get_all_ucsmbindings()
count = 0
for bind in bindings:
if "net " in bind["network-id"]:
count += 1
self.assertTrue(count == 0)
self.teardown_ucsmbinding()
self.teardown_network()
def testd_update_ucsmbinding(self):
"""update ucsm binding"""
net1 = self.quantum.create_network("t1", "netid1")
net2 = self.quantum.create_network("t1", "netid2")
binding1 = self.dbtest.create_ucsmbinding("1.2.3.4", net1["net-id"])
binding1 = self.dbtest.update_ucsmbinding(binding1["ucsm-ip"],
net2["net-id"])
bindings = self.dbtest.get_all_ucsmbindings()
count = 0
for bind in bindings:
if net2["net-id"] == bind["network-id"]:
count += 1
self.assertTrue(count == 1)
self.teardown_ucsmbinding()
self.teardown_network()
def teste_create_dynamicvnic(self):
"""create dynamic vnic"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
self.assertTrue(vnic1["device-name"] == "eth1")
self.teardown_dyanmicvnic()
def testf_getall_dyanmicvnics(self):
"""get all dynamic vnics"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
vnic2 = self.dbtest.create_dynamicvnic("eth2", blade1["blade-id"],
"UP")
vnics = self.dbtest.get_all_dynamicvnics()
count = 0
for vnic in vnics:
if "eth" in vnic["device-name"]:
count += 1
self.assertTrue(count == 2)
self.teardown_dyanmicvnic()
def testg_delete_dyanmicvnic(self):
"""delete dynamic vnic"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
self.dbtest.delete_dynamicvnic(vnic1["vnic-id"])
vnics = self.dbtest.get_all_dynamicvnics()
count = 0
for vnic in vnics:
if "eth " in vnic["device-name"]:
count += 1
self.assertTrue(count == 0)
self.teardown_dyanmicvnic()
def testh_updatedynamicvnic(self):
"""update dynamic vnic"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
vnic1 = self.dbtest.update_dynamicvnic(vnic1["vnic-id"], "neweth1",
blade1["blade-id"], "DOWN")
vnics = self.dbtest.get_all_dynamicvnics()
count = 0
for vnic in vnics:
if "new" in vnic["device-name"]:
count += 1
self.assertTrue(count == 1)
self.teardown_dyanmicvnic()
def testi_create_ucsblade(self):
"""create ucs blade"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
self.assertTrue(blade1["mgmt-ip"] == "1.2.3.4")
self.teardown_ucsblade()
def testj_getall_ucsblade(self):
"""get all ucs blades"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
blade2 = self.dbtest.create_blade("2.3.4.5", "efgh", "chassis1",
"9.8.7.6", "UP", 3, "blade2")
blades = self.dbtest.get_all_blades()
count = 0
for blade in blades:
if "chassis" in blade["chassis-id"]:
count += 1
self.assertTrue(count == 2)
self.teardown_ucsblade()
def testk_delete_ucsblade(self):
"""delete ucs blades"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
self.dbtest.delete_blade(blade1["blade-id"])
blades = self.dbtest.get_all_blades()
count = 0
for blade in blades:
if "chassis " in blade["chassis-id"]:
count += 1
self.assertTrue(count == 0)
self.teardown_ucsblade()
def testl_update_ucsblade(self):
"""update ucs blade"""
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
blade2 = self.dbtest.update_blade(blade1["blade-id"], "2.3.4.5",
"newabcd", "chassis1", "9.8.7.6",
"UP", 3, "blade1")
blades = self.dbtest.get_all_blades()
count = 0
for blade in blades:
if "new" in blade["mac-addr"]:
count += 1
self.assertTrue(count == 1)
self.teardown_ucsblade()
def testm_create_portbinding(self):
"""create port binding"""
net1 = self.quantum.create_network("t1", "netid1")
port1 = self.quantum.create_port(net1["net-id"])
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
port_bind1 = self.dbtest.create_port_binding(port1["port-id"],
vnic1["vnic-id"], "pp1", "vlan1", 10, "qos1")
"vnic1", "pp1", "vlan1", 10, "qos1")
self.assertTrue(port_bind1["port-id"] == port1["port-id"])
self.teardown_portbinding()
self.teardown_dyanmicvnic()
self.teardown_ucsblade()
self.teardown_network_port()
def testn_getall_portbindings(self):
@ -1002,16 +610,10 @@ class UcsDBTest(unittest.TestCase):
net1 = self.quantum.create_network("t1", "netid1")
port1 = self.quantum.create_port(net1["net-id"])
port2 = self.quantum.create_port(net1["net-id"])
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
vnic2 = self.dbtest.create_dynamicvnic("eth2", blade1["blade-id"],
"UP")
port_bind1 = self.dbtest.create_port_binding(port1["port-id"],
vnic1["vnic-id"], "pp1", "vlan1", 10, "qos1")
"vnic1", "pp1", "vlan1", 10, "qos1")
port_bind2 = self.dbtest.create_port_binding(port2["port-id"],
vnic2["vnic-id"], "pp2", "vlan2", 20, "qos2")
"vnic2", "pp2", "vlan2", 20, "qos2")
port_bindings = self.dbtest.get_all_port_bindings()
count = 0
for pbind in port_bindings:
@ -1019,20 +621,14 @@ class UcsDBTest(unittest.TestCase):
count += 1
self.assertTrue(count == 2)
self.teardown_portbinding()
self.teardown_dyanmicvnic()
self.teardown_ucsblade()
self.teardown_network_port()
def testo_delete_portbinding(self):
"""delete port binding"""
net1 = self.quantum.create_network("t1", "netid1")
port1 = self.quantum.create_port(net1["net-id"])
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
port_bind1 = self.dbtest.create_port_binding(port1["port-id"],
vnic1["vnic-id"], "pp1", "vlan1", 10, "qos1")
"vnic1", "pp1", "vlan1", 10, "qos1")
self.dbtest.delete_port_binding(port1["port-id"])
port_bindings = self.dbtest.get_all_port_bindings()
count = 0
@ -1041,22 +637,16 @@ class UcsDBTest(unittest.TestCase):
count += 1
self.assertTrue(count == 0)
self.teardown_portbinding()
self.teardown_dyanmicvnic()
self.teardown_ucsblade()
self.teardown_network_port()
def testp_update_portbinding(self):
"""update port binding"""
net1 = self.quantum.create_network("t1", "netid1")
port1 = self.quantum.create_port(net1["net-id"])
blade1 = self.dbtest.create_blade("1.2.3.4", "abcd", "chassis1",
"9.8.7.6", "UP", 2, "blade1")
vnic1 = self.dbtest.create_dynamicvnic("eth1", blade1["blade-id"],
"UP")
port_bind1 = self.dbtest.create_port_binding(port1["port-id"],
vnic1["vnic-id"], "pp1", "vlan1", 10, "qos1")
"vnic1", "pp1", "vlan1", 10, "qos1")
port_bind1 = self.dbtest.update_port_binding(port1["port-id"],
vnic1["vnic-id"], "newpp1", "newvlan1", 11, "newqos1")
"vnic1", "newpp1", "newvlan1", 11, "newqos1")
port_bindings = self.dbtest.get_all_port_bindings()
count = 0
for pbind in port_bindings:
@ -1064,35 +654,8 @@ class UcsDBTest(unittest.TestCase):
count += 1
self.assertTrue(count == 1)
self.teardown_portbinding()
self.teardown_dyanmicvnic()
self.teardown_ucsblade()
self.teardown_network_port()
def teardown_ucsmbinding(self):
"""tear down ucsm binding"""
LOG.debug("Tearing Down Ucsm Bindings")
binds = self.dbtest.get_all_ucsmbindings()
for bind in binds:
ucsmip = bind["ucsm-ip"]
self.dbtest.delete_ucsmbinding(ucsmip)
def teardown_dyanmicvnic(self):
"""tear down dynamic vnics"""
LOG.debug("Tearing Down Dynamic Vnics")
vnics = self.dbtest.get_all_dynamicvnics()
for vnic in vnics:
vnicid = vnic["vnic-id"]
self.dbtest.delete_dynamicvnic(vnicid)
self.teardown_ucsblade()
def teardown_ucsblade(self):
"""tear down ucs blades"""
LOG.debug("Tearing Down Blades")
blades = self.dbtest.get_all_blades()
for blade in blades:
bladeid = blade["blade-id"]
self.dbtest.delete_blade(bladeid)
def teardown_portbinding(self):
"""tear down port binding"""
LOG.debug("Tearing Down Port Binding")
@ -1101,14 +664,6 @@ class UcsDBTest(unittest.TestCase):
portid = port_binding["port-id"]
self.dbtest.delete_port_binding(portid)
def teardown_network(self):
"""tearDown Network table"""
LOG.debug("Tearing Down Network")
nets = self.quantum.get_all_networks("t1")
for net in nets:
netid = net["net-id"]
self.quantum.delete_network(netid)
def teardown_network_port(self):
"""tearDown for Network and Port table"""
networks = self.quantum.get_all_networks("t1")
@ -1391,7 +946,7 @@ class L2networkDBTest(unittest.TestCase):
self.assertTrue(used == True)
used = l2network_db.release_vlanid(vlanid)
self.assertTrue(used == False)
self.teardown_vlanid()
#counting on default teardown here to clear db
def teardown_network(self):
"""tearDown Network table"""
@ -1437,14 +992,6 @@ class L2networkDBTest(unittest.TestCase):
portid = pp_binding["port-id"]
self.dbtest.delete_pp_binding("t1", portid, ppid)
def teardown_vlanid(self):
"""tearDown VlanID table"""
LOG.debug("Tearing Down Vlan IDs")
vlanids = l2network_db.get_all_vlanids()
for vlanid in vlanids:
vlan_id = vlanid["vlan_id"]
l2network_db.delete_vlanid(vlan_id)
class QuantumDBTest(unittest.TestCase):
"""Class conisting of Quantum DB unit tests"""