"for tenant %(tenant_id)s")
+class NexusPortBindingNotFound(exceptions.QuantumException):
+ """NexusPort Binding is not present"""
+ message = _("Nexus Port Binding %(port_id) is not present")
+
+
try:
_("test")
except NameError:
from quantum.plugins.cisco.common import cisco_exceptions as c_exc
import l2network_models
+import logging as LOG
import quantum.plugins.cisco.db.api as db
def create_vlanids():
"""Prepopulates the vlan_bindings table"""
+ LOG.debug("create_vlanids() called")
session = db.get_session()
try:
vlanid = session.query(l2network_models.VlanID).\
def get_all_vlanids():
"""Gets all the vlanids"""
+ LOG.debug("get_all_vlanids() called")
session = db.get_session()
try:
vlanids = session.query(l2network_models.VlanID).\
def is_vlanid_used(vlan_id):
"""Checks if a vlanid is in use"""
+ LOG.debug("is_vlanid_used() called")
session = db.get_session()
try:
vlanid = session.query(l2network_models.VlanID).\
def release_vlanid(vlan_id):
"""Sets the vlanid state to be unused"""
+ LOG.debug("release_vlanid() called")
session = db.get_session()
try:
vlanid = session.query(l2network_models.VlanID).\
def delete_vlanid(vlan_id):
"""Deletes a vlanid entry from db"""
+ LOG.debug("delete_vlanid() called")
session = db.get_session()
try:
vlanid = session.query(l2network_models.VlanID).\
def reserve_vlanid():
"""Reserves the first unused vlanid"""
+ LOG.debug("reserve_vlanid() called")
session = db.get_session()
try:
vlanids = session.query(l2network_models.VlanID).\
def get_all_vlan_bindings():
"""Lists all the vlan to network associations"""
+ LOG.debug("get_all_vlan_bindings() called")
session = db.get_session()
try:
bindings = session.query(l2network_models.VlanBinding).\
def get_vlan_binding(netid):
"""Lists the vlan given a network_id"""
+ LOG.debug("get_vlan_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.VlanBinding).\
def add_vlan_binding(vlanid, vlanname, netid):
"""Adds a vlan to network association"""
+ LOG.debug("add_vlan_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.VlanBinding).\
def remove_vlan_binding(netid):
"""Removes a vlan to network association"""
+ LOG.debug("remove_vlan_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.VlanBinding).\
def update_vlan_binding(netid, newvlanid=None, newvlanname=None):
"""Updates a vlan to network association"""
+ LOG.debug("update_vlan_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.VlanBinding).\
def get_all_portprofiles():
"""Lists all the port profiles"""
+ LOG.debug("get_all_portprofiles() called")
session = db.get_session()
try:
pps = session.query(l2network_models.PortProfile).\
def get_portprofile(tenantid, ppid):
"""Lists a port profile"""
+ LOG.debug("get_portprofile() called")
session = db.get_session()
try:
pp = session.query(l2network_models.PortProfile).\
def add_portprofile(tenantid, ppname, vlanid, qos):
"""Adds a port profile"""
+ LOG.debug("add_portprofile() called")
session = db.get_session()
try:
pp = session.query(l2network_models.PortProfile).\
def remove_portprofile(tenantid, ppid):
"""Removes a port profile"""
+ LOG.debug("remove_portprofile() called")
session = db.get_session()
try:
pp = session.query(l2network_models.PortProfile).\
pass
-def update_portprofile(tenantid, ppid, newppname=None, newvlanid=None,
+def update_portprofile(tenantid, ppid, newppname=None, newvlanid=None,
newqos=None):
"""Updates port profile"""
+ LOG.debug("update_portprofile() called")
session = db.get_session()
try:
pp = session.query(l2network_models.PortProfile).\
def get_all_pp_bindings():
"""Lists all the port profiles"""
+ LOG.debug("get_all_pp_bindings() called")
session = db.get_session()
try:
bindings = session.query(l2network_models.PortProfileBinding).\
def get_pp_binding(tenantid, ppid):
"""Lists a port profile binding"""
+ LOG.debug("get_pp_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.PortProfileBinding).\
def add_pp_binding(tenantid, portid, ppid, default):
"""Adds a port profile binding"""
+ LOG.debug("add_pp_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.PortProfileBinding).\
def remove_pp_binding(tenantid, portid, ppid):
"""Removes a port profile binding"""
+ LOG.debug("remove_pp_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.PortProfileBinding).\
pass
-def update_pp_binding(tenantid, ppid, newtenantid=None, newportid=None,
+def update_pp_binding(tenantid, ppid, newtenantid=None, newportid=None,
newdefault=None):
"""Updates port profile binding"""
+ LOG.debug("update_pp_binding() called")
session = db.get_session()
try:
binding = session.query(l2network_models.PortProfileBinding).\
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011, Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+# @author: Rohit Agarwalla, Cisco Systems, Inc.
+
+import logging as LOG
+
+from sqlalchemy.orm import exc
+
+import quantum.plugins.cisco.db.api as db
+import nexus_models
+
+from quantum.plugins.cisco.common import cisco_exceptions as c_exc
+
+
+def get_all_nexusport_bindings():
+ """Lists all the nexusport bindings"""
+ LOG.debug("get_all_nexusport_bindings() called")
+ session = db.get_session()
+ try:
+ bindings = session.query(nexus_models.NexusPortBinding).\
+ all()
+ return bindings
+ except exc.NoResultFound:
+ return []
+
+
+def get_nexusport_binding(vlan_id):
+ """Lists a nexusport binding"""
+ LOG.debug("get_nexusport_binding() called")
+ session = db.get_session()
+ try:
+ binding = session.query(nexus_models.NexusPortBinding).\
+ filter_by(vlan_id=vlan_id).\
+ all()
+ return binding
+ except exc.NoResultFound:
+ raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
+
+
+def add_nexusport_binding(port_id, vlan_id):
+ """Adds a nexusport binding"""
+ LOG.debug("add_nexusport_binding() called")
+ session = db.get_session()
+ binding = nexus_models.NexusPortBinding(port_id, vlan_id)
+ session.add(binding)
+ session.flush()
+ return binding
+
+
+def remove_nexusport_binding(vlan_id):
+ """Removes a nexusport binding"""
+ LOG.debug("remove_nexusport_binding() called")
+ session = db.get_session()
+ try:
+ binding = session.query(nexus_models.NexusPortBinding).\
+ filter_by(vlan_id=vlan_id).\
+ all()
+ for bind in binding:
+ session.delete(bind)
+ session.flush()
+ return binding
+ except exc.NoResultFound:
+ pass
+
+
+def update_nexusport_binding(port_id, new_vlan_id):
+ """Updates nexusport binding"""
+ LOG.debug("update_nexusport_binding called")
+ session = db.get_session()
+ try:
+ binding = session.query(nexus_models.NexusPortBinding).\
+ filter_by(port_id=port_id).\
+ one()
+ if new_vlan_id:
+ binding["vlan_id"] = new_vlan_id
+ session.merge(binding)
+ session.flush()
+ return binding
+ except exc.NoResultFound:
+ raise c_exc.NexusPortBindingNotFound()
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011, Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+# @author: Rohit Agarwalla, Cisco Systems, Inc.
+
+from sqlalchemy import Column, Integer, String
+
+from quantum.plugins.cisco.db.l2network_models import L2NetworkBase
+from quantum.plugins.cisco.db.models import BASE
+
+
+class NexusPortBinding(BASE, L2NetworkBase):
+ """Represents a binding of nexus port to vlan_id"""
+ __tablename__ = 'nexusport_bindings'
+
+ id = Column(Integer, primary_key=True, autoincrement=True)
+ port_id = Column(String(255))
+ #vlan_id = Column(Integer, ForeignKey("vlan_bindings.vlan_id"), \
+ # nullable=False)
+ vlan_id = Column(Integer, nullable=False)
+
+ def __init__(self, port_id, vlan_id):
+ self.port_id = port_id
+ self.vlan_id = vlan_id
+
+ def __repr__(self):
+ return "<NexusPortBinding (%s,%d)>" % \
+ (self.port_id, self.vlan_id)
import quantum.plugins.cisco.db.api as db
import quantum.plugins.cisco.db.l2network_db as l2network_db
+import quantum.plugins.cisco.db.nexus_db as nexus_db
LOG.getLogger(const.LOGGER_COMPONENT_NAME)
+class NexusDB(object):
+ """Class consisting of methods to call nexus db methods"""
+ def get_all_nexusportbindings(self):
+ """get all nexus port bindings"""
+ bindings = []
+ try:
+ for bind in nexus_db.get_all_nexusport_bindings():
+ LOG.debug("Getting nexus port binding : %s" % bind.port_id)
+ bind_dict = {}
+ bind_dict["port-id"] = str(bind.port_id)
+ bind_dict["vlan-id"] = str(bind.vlan_id)
+ bindings.append(bind_dict)
+ except Exception, exc:
+ LOG.error("Failed to get all bindings: %s" % str(exc))
+ return bindings
+
+ def get_nexusportbinding(self, port_id):
+ """get nexus port binding"""
+ binding = []
+ try:
+ for bind in nexus_db.get_nexusport_binding(port_id):
+ LOG.debug("Getting nexus port binding : %s" % bind.port_id)
+ bind_dict = {}
+ bind_dict["port-id"] = str(bind.port_id)
+ bind_dict["vlan-id"] = str(bind.vlan_id)
+ binding.append(bind_dict)
+ except Exception, exc:
+ LOG.error("Failed to get all bindings: %s" % str(exc))
+ return binding
+
+ def create_nexusportbinding(self, port_id, vlan_id):
+ """create nexus port binding"""
+ 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)
+ return bind_dict
+ except Exception, exc:
+ LOG.error("Failed to create ucsm binding: %s" % str(exc))
+
+ def delete_nexusportbinding(self, port_id):
+ """delete nexus port binding"""
+ 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
+ except Exception, exc:
+ raise Exception("Failed to delete dynamic vnic: %s" % str(exc))
+
+ def update_nexusportbinding(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)
+ return bind_dict
+ except Exception, exc:
+ raise Exception("Failed to update dynamic vnic: %s" % str(exc))
+
+
class L2networkDB(object):
"""Class conisting of methods to call L2network db methods"""
def get_all_vlan_bindings(self):
raise Exception("Failed to unplug interface: %s" % str(exc))
+class NexusDBTest(unittest.TestCase):
+ """Class conisting of nexus DB unit tests"""
+ def setUp(self):
+ """Setup for ucs db tests"""
+ l2network_db.initialize()
+ self.dbtest = NexusDB()
+ LOG.debug("Setup")
+
+ def tearDown(self):
+ """Tear Down"""
+ db.clear_db()
+
+ def testa_create_nexusportbinding(self):
+ """create nexus port binding"""
+ binding1 = self.dbtest.create_nexusportbinding("port1", 10)
+ self.assertTrue(binding1["port-id"] == "port1")
+ self.tearDown_nexusportbinding()
+
+ def testb_getall_nexusportbindings(self):
+ """get all nexus port binding"""
+ binding1 = self.dbtest.create_nexusportbinding("port1", 10)
+ binding2 = self.dbtest.create_nexusportbinding("port2", 10)
+ bindings = self.dbtest.get_all_nexusportbindings()
+ count = 0
+ for bind in bindings:
+ if "port" in bind["port-id"]:
+ count += 1
+ self.assertTrue(count == 2)
+ self.tearDown_nexusportbinding()
+
+ def testc_delete_nexusportbinding(self):
+ """delete nexus port binding"""
+ binding1 = self.dbtest.create_nexusportbinding("port1", 10)
+ self.dbtest.delete_nexusportbinding(binding1["port-id"])
+ bindings = self.dbtest.get_all_nexusportbindings()
+ count = 0
+ for bind in bindings:
+ if "port " in bind["port-id"]:
+ count += 1
+ self.assertTrue(count == 0)
+ self.tearDown_nexusportbinding()
+
+ def testd_update_nexusportbinding(self):
+ """update nexus port binding"""
+ binding1 = self.dbtest.create_nexusportbinding("port1", 10)
+ binding1 = self.dbtest.update_nexusportbinding(binding1["port-id"], \
+ 20)
+ bindings = self.dbtest.get_all_nexusportbindings()
+ count = 0
+ for bind in bindings:
+ if "20" in str(bind["vlan-id"]):
+ count += 1
+ self.assertTrue(count == 1)
+ self.tearDown_nexusportbinding()
+
+ def tearDown_nexusportbinding(self):
+ """tear down ucsm 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)
+
+
class L2networkDBTest(unittest.TestCase):
"""Class conisting of L2network DB unit tests"""
def setUp(self):
Code Review / openstack-build / neutron-build.git / commitdiff