class NexusPortBindingNotFound(exceptions.QuantumException):
"""NexusPort Binding is not present."""
- message = _("Nexus Port Binding %(port_id)s is not present.")
+ message = _("Nexus Port Binding (%(filters)s) is not present")
+
+ def __init__(self, **kwargs):
+ filters = ','.join('%s=%s' % i for i in kwargs.items())
+ super(NexusPortBindingNotFound, self).__init__(filters=filters)
class PortVnicBindingAlreadyExists(exceptions.QuantumException):
LOG.debug(_("get_nexusport_binding() called"))
session = db.get_session()
- # FIXME(rpodolyaka): https://bugs.launchpad.net/quantum/+bug/1174323
- return (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(vlan_id=vlan_id).filter_by(switch_ip=switch_ip).
- filter_by(port_id=port_id).
- filter_by(instance_id=instance_id).all())
+ filters = dict(port_id=port_id, vlan_id=vlan_id, switch_ip=switch_ip,
+ instance_id=instance_id)
+ bindings = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(**filters).all())
+ if not bindings:
+ raise c_exc.NexusPortBindingNotFound(**filters)
+
+ return bindings
def get_nexusvlan_binding(vlan_id, switch_ip):
LOG.debug(_("get_nexusvlan_binding() called"))
session = db.get_session()
- # FIXME(rpodolyaka): https://bugs.launchpad.net/quantum/+bug/1174323
- return (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(vlan_id=vlan_id).filter_by(switch_ip=switch_ip).
- all())
+ filters = dict(vlan_id=vlan_id, switch_ip=switch_ip)
+ bindings = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(**filters).all())
+ if not bindings:
+ raise c_exc.NexusPortBindingNotFound(**filters)
+
+ return bindings
def add_nexusport_binding(port_id, vlan_id, switch_ip, instance_id):
session.flush()
return binding
except exc.NoResultFound:
- raise c_exc.NexusPortBindingNotFound()
+ raise c_exc.NexusPortBindingNotFound(port_id=port_id)
def get_nexusvm_binding(vlan_id, instance_id):
"""Lists nexusvm bindings."""
LOG.debug(_("get_nexusvm_binding() called"))
session = db.get_session()
- try:
- binding = (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(instance_id=instance_id).
- filter_by(vlan_id=vlan_id).first())
- return binding
- except exc.NoResultFound:
- raise c_exc.NexusPortBindingNotFound(vlan_id=vlan_id)
+
+ filters = dict(instance_id=instance_id, vlan_id=vlan_id)
+ binding = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(**filters).first())
+ if not binding:
+ raise c_exc.NexusPortBindingNotFound(**filters)
+
+ return binding
def get_port_vlan_switch_binding(port_id, vlan_id, switch_ip):
LOG.debug(_("get_port_vlan_switch_binding() called"))
session = db.get_session()
- # FIXME(rpodolyaka): https://bugs.launchpad.net/quantum/+bug/1174323
- return (session.query(nexus_models_v2.NexusPortBinding).
- filter_by(port_id=port_id).filter_by(switch_ip=switch_ip).
- filter_by(vlan_id=vlan_id).all())
+ filters = dict(port_id=port_id, switch_ip=switch_ip, vlan_id=vlan_id)
+ bindings = (session.query(nexus_models_v2.NexusPortBinding).
+ filter_by(**filters).all())
+ if not bindings:
+ raise c_exc.NexusPortBindingNotFound(**filters)
+
+ return bindings
raise cisco_exc.NexusComputeHostNotConfigured(host=host)
# Check if this network is already in the DB
- binding = nxos_db.get_port_vlan_switch_binding(
- port_id, vlan_id, switch_ip)
vlan_created = False
vlan_enabled = False
- if not binding:
+
+ try:
+ nxos_db.get_port_vlan_switch_binding(port_id, vlan_id, switch_ip)
+ except cisco_exc.NexusPortBindingNotFound:
_nexus_ip = switch_ip
_nexus_ports = (port_id,)
_nexus_ssh_port = \
_nexus_username = _nexus_creds['username']
_nexus_password = _nexus_creds['password']
# Check for vlan/switch binding
- vbinding = nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
- if not vbinding:
+ try:
+ nxos_db.get_nexusvlan_binding(vlan_id, switch_ip)
+ except cisco_exc.NexusPortBindingNotFound:
# Create vlan and trunk vlan on the port
self._client.create_vlan(
vlan_name, str(vlan_id), _nexus_ip,
"""
LOG.debug(_("NexusPlugin:delete_port() called"))
# Delete DB row for this port
- row = nxos_db.get_nexusvm_binding(vlan_id, device_id)
- if row:
- nxos_db.remove_nexusport_binding(row['port_id'], row['vlan_id'],
- row['switch_ip'],
- row['instance_id'])
- # Check for any other bindings with the same vlan_id and switch_ip
- bindings = nxos_db.get_nexusvlan_binding(
- row['vlan_id'], row['switch_ip'])
-
- if not bindings:
+ try:
+ row = nxos_db.get_nexusvm_binding(vlan_id, device_id)
+ except cisco_exc.NexusPortBindingNotFound:
+ return
+
+ nxos_db.remove_nexusport_binding(row['port_id'], row['vlan_id'],
+ row['switch_ip'],
+ row['instance_id'])
+ # Check for any other bindings with the same vlan_id and switch_ip
+ try:
+ nxos_db.get_nexusvlan_binding(row['vlan_id'], row['switch_ip'])
+ except cisco_exc.NexusPortBindingNotFound:
+ try:
+ # Delete this vlan from this switch
+ _nexus_ip = row['switch_ip']
+ _nexus_ports = (row['port_id'],)
+ _nexus_ssh_port = (self._nexus_switches[_nexus_ip,
+ 'ssh_port'])
+ _nexus_creds = self.get_credential(_nexus_ip)
+ _nexus_username = _nexus_creds['username']
+ _nexus_password = _nexus_creds['password']
+ self._client.delete_vlan(
+ str(row['vlan_id']), _nexus_ip,
+ _nexus_username, _nexus_password,
+ _nexus_ports, _nexus_ssh_port)
+ except Exception as e:
+ # The delete vlan operation on the Nexus failed,
+ # so this delete_port request has failed. For
+ # consistency, roll back the Nexus database to what
+ # it was before this request.
try:
- # Delete this vlan from this switch
- _nexus_ip = row['switch_ip']
- _nexus_ports = (row['port_id'],)
- _nexus_ssh_port = (self._nexus_switches[_nexus_ip,
- 'ssh_port'])
- _nexus_creds = self.get_credential(_nexus_ip)
- _nexus_username = _nexus_creds['username']
- _nexus_password = _nexus_creds['password']
- self._client.delete_vlan(
- str(row['vlan_id']), _nexus_ip,
- _nexus_username, _nexus_password,
- _nexus_ports, _nexus_ssh_port)
- except Exception as e:
- # The delete vlan operation on the Nexus failed,
- # so this delete_port request has failed. For
- # consistency, roll back the Nexus database to what
- # it was before this request.
- try:
- nxos_db.add_nexusport_binding(row['port_id'],
- row['vlan_id'],
- row['switch_ip'],
- row['instance_id'])
- finally:
- # Raise the original exception
- raise e
-
- return row['instance_id']
+ nxos_db.add_nexusport_binding(row['port_id'],
+ row['vlan_id'],
+ row['switch_ip'],
+ row['instance_id'])
+ finally:
+ # Raise the original exception
+ raise e
+
+ return row['instance_id']
def update_port(self, tenant_id, net_id, port_id, port_state, **kwargs):
"""Update port.
that tests the database api method calls
"""
+import mock
+import sqlalchemy
+
from quantum.openstack.common import log as logging
+import quantum.plugins.cisco.common.cisco_exceptions as c_exc
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_v2 as nexus_db
self.assertTrue(count == 1)
self.tearDown_nexusportbinding()
+ def test_get_nexusport_binding_no_result_found_handling(self):
+ with mock.patch('sqlalchemy.orm.Query.all') as mock_all:
+ mock_all.return_value = []
+
+ with self.assertRaises(c_exc.NexusPortBindingNotFound):
+ nexus_db.get_nexusport_binding(port_id=10,
+ vlan_id=20,
+ switch_ip='10.0.0.1',
+ instance_id=1)
+
+ def test_get_nexusvlan_binding_no_result_found_handling(self):
+ with mock.patch('sqlalchemy.orm.Query.all') as mock_all:
+ mock_all.return_value = []
+
+ with self.assertRaises(c_exc.NexusPortBindingNotFound):
+ nexus_db.get_nexusvlan_binding(vlan_id=10,
+ switch_ip='10.0.0.1')
+
+ def test_update_nexusport_binding_no_result_found_handling(self):
+ with mock.patch('sqlalchemy.orm.Query.one') as mock_one:
+ mock_one.side_effect = sqlalchemy.orm.exc.NoResultFound
+
+ with self.assertRaises(c_exc.NexusPortBindingNotFound):
+ nexus_db.update_nexusport_binding(port_id=10,
+ vlan_id=20,
+ switch_ip='10.0.0.1',
+ instance_id=1)
+
+ def test_get_nexusvm_binding_no_result_found_handling(self):
+ with mock.patch('sqlalchemy.orm.Query.first') as mock_first:
+ mock_first.return_value = None
+
+ with self.assertRaises(c_exc.NexusPortBindingNotFound):
+ nexus_db.get_nexusvm_binding(port_id=10,
+ vlan_id=20,
+ switch_ip='10.0.0.1')
+
+ def test_nexusport_binding_not_found_exception_message_formatting(self):
+ try:
+ raise c_exc.NexusPortBindingNotFound(a=1, b='test')
+ except c_exc.NexusPortBindingNotFound as e:
+ self.assertIn('(a=1,b=test)', str(e))
+
def tearDown_nexusportbinding(self):
"""Tear down nexus port binding table."""
LOG.debug("Tearing Down Nexus port Bindings")
Code Review / openstack-build / neutron-build.git / commitdiff