LOG.error("Failed to get all bindings: %s" % str(exc))
return bindings
- def get_nexusportbinding(self, port_id):
+ def get_nexusportbinding(self, vlan_id):
"""get nexus port binding"""
binding = []
try:
- for bind in nexus_db.get_nexusport_binding(port_id):
+ for bind in nexus_db.get_nexusport_binding(vlan_id):
LOG.debug("Getting nexus port binding : %s" % bind.port_id)
bind_dict = {}
bind_dict["port-id"] = str(bind.port_id)
bind_dict = {}
try:
res = nexus_db.add_nexusport_binding(port_id, vlan_id)
- LOG.debug("Created nexus port binding: %s" % res.port_id)
- bind_dict["port-id"] = str(bind.port_id)
- bind_dict["vlan-id"] = str(bind.vlan_id)
+ LOG.debug("Created nexus port binding : %s" % res.port_id)
+ bind_dict["port-id"] = str(res.port_id)
+ bind_dict["vlan-id"] = str(res.vlan_id)
return bind_dict
except Exception, exc:
- LOG.error("Failed to create ucsm binding: %s" % str(exc))
+ LOG.error("Failed to create nexus binding: %s" % str(exc))
- def delete_nexusportbinding(self, port_id):
+ def delete_nexusportbinding(self, vlan_id):
"""delete nexus port binding"""
+ bindings = []
try:
- res = nexus_db.remove_nexusport_binding(port_id)
- LOG.debug("Deleted nexus port binding : %s" % res.port_id)
- bind_dict = {}
- bind_dict["port-id"] = str(res.port_id)
- return bind_dict
+ bind = nexus_db.remove_nexusport_binding(vlan_id)
+ for res in bind:
+ LOG.debug("Deleted nexus port binding: %s" % res.vlan_id)
+ bind_dict = {}
+ bind_dict["port-id"] = res.port_id
+ bindings.append(bind_dict)
+ return bindings
except Exception, exc:
- raise Exception("Failed to delete dynamic vnic: %s" % str(exc))
+ raise Exception("Failed to delete nexus port binding: %s"
+ % str(exc))
- def update_nexusportbinding(self, port_id, new_vlan_id):
+ def update_nexusport_binding(self, port_id, new_vlan_id):
"""update nexus port binding"""
try:
res = nexus_db.update_nexusport_binding(port_id, new_vlan_id)
LOG.debug("Updating nexus port binding : %s" % res.port_id)
bind_dict = {}
- bind_dict["port-id"] = str(bind.port_id)
- bind_dict["vlan-id"] = str(bind.vlan_id)
+ bind_dict["port-id"] = str(res.port_id)
+ bind_dict["vlan-id"] = str(res.vlan_id)
return bind_dict
except Exception, exc:
- raise Exception("Failed to update dynamic vnic: %s" % str(exc))
+ raise Exception("Failed to update nexus port binding vnic: %s"
+ % str(exc))
class L2networkDB(object):
pp_list.append(pp_dict)
except Exception, exc:
LOG.error("Failed to get port profile: %s" % str(exc))
- return pp
+ return pp_list
def create_portprofile(self, tenant_id, name, vlan_id, qos):
"""Create a portprofile"""
class NexusDBTest(unittest.TestCase):
"""Class conisting of nexus DB unit tests"""
def setUp(self):
- """Setup for ucs db tests"""
+ """Setup for nexus db tests"""
l2network_db.initialize()
self.dbtest = NexusDB()
LOG.debug("Setup")
def testc_delete_nexusportbinding(self):
"""delete nexus port binding"""
binding1 = self.dbtest.create_nexusportbinding("port1", 10)
- self.dbtest.delete_nexusportbinding(binding1["port-id"])
+ self.dbtest.delete_nexusportbinding(10)
bindings = self.dbtest.get_all_nexusportbindings()
count = 0
for bind in bindings:
def testd_update_nexusportbinding(self):
"""update nexus port binding"""
binding1 = self.dbtest.create_nexusportbinding("port1", 10)
- binding1 = self.dbtest.update_nexusportbinding(binding1["port-id"], \
+ binding1 = self.dbtest.update_nexusport_binding(binding1["port-id"], \
20)
bindings = self.dbtest.get_all_nexusportbindings()
count = 0
self.tearDown_nexusportbinding()
def tearDown_nexusportbinding(self):
- """tear down ucsm binding table"""
+ """tear down nexusport binding table"""
LOG.debug("Tearing Down Nexus port Bindings")
binds = self.dbtest.get_all_nexusportbindings()
for bind in binds:
- port_id = bind["port-id"]
- self.dbtest.delete_nexusportbinding(port_id)
+ vlan_id = bind["vlan-id"]
+ self.dbtest.delete_nexusportbinding(vlan_id)
class L2networkDBTest(unittest.TestCase):