# Copyright (c) 2013 OpenStack Foundation # All Rights Reserved. # # 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. import collections import testtools from neutron.db import api as db from neutron.plugins.cisco.common import cisco_exceptions as c_exc from neutron.plugins.cisco.db import nexus_db_v2 as nxdb from neutron.tests import base class CiscoNexusDbTest(base.BaseTestCase): """Unit tests for cisco.db.nexus_models_v2.NexusPortBinding model.""" NpbObj = collections.namedtuple('NpbObj', 'port vlan switch instance') def setUp(self): super(CiscoNexusDbTest, self).setUp() db.configure_db() self.session = db.get_session() self.addCleanup(db.clear_db) def _npb_test_obj(self, pnum, vnum, switch=None, instance=None): """Create a Nexus port binding test object from a pair of numbers.""" if pnum is 'router': port = pnum else: port = '1/%s' % str(pnum) vlan = str(vnum) if switch is None: switch = '10.9.8.7' if instance is None: instance = 'instance_%s_%s' % (str(pnum), str(vnum)) return self.NpbObj(port, vlan, switch, instance) def _assert_equal(self, npb, npb_obj): self.assertEqual(npb.port_id, npb_obj.port) self.assertEqual(int(npb.vlan_id), int(npb_obj.vlan)) self.assertEqual(npb.switch_ip, npb_obj.switch) self.assertEqual(npb.instance_id, npb_obj.instance) def _add_to_db(self, npbs): for npb in npbs: nxdb.add_nexusport_binding( npb.port, npb.vlan, npb.switch, npb.instance) def test_nexusportbinding_add_remove(self): npb11 = self._npb_test_obj(10, 100) npb = nxdb.add_nexusport_binding( npb11.port, npb11.vlan, npb11.switch, npb11.instance) self._assert_equal(npb, npb11) npb = nxdb.remove_nexusport_binding( npb11.port, npb11.vlan, npb11.switch, npb11.instance) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb11) with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.remove_nexusport_binding( npb11.port, npb11.vlan, npb11.switch, npb11.instance) def test_nexusportbinding_get(self): npb11 = self._npb_test_obj(10, 100) npb21 = self._npb_test_obj(20, 100) npb22 = self._npb_test_obj(20, 200) self._add_to_db([npb11, npb21, npb22]) npb = nxdb.get_nexusport_binding( npb11.port, npb11.vlan, npb11.switch, npb11.instance) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb11) npb = nxdb.get_nexusport_binding( npb21.port, npb21.vlan, npb21.switch, npb21.instance) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb21) npb = nxdb.get_nexusport_binding( npb22.port, npb22.vlan, npb22.switch, npb22.instance) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb22) with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.get_nexusport_binding( npb21.port, npb21.vlan, npb21.switch, "dummyInstance") def test_nexusvlanbinding_get(self): npb11 = self._npb_test_obj(10, 100) npb21 = self._npb_test_obj(20, 100) npb22 = self._npb_test_obj(20, 200) self._add_to_db([npb11, npb21, npb22]) npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, npb11.switch) self.assertEqual(len(npb_all_v100), 2) npb_v200 = nxdb.get_nexusvlan_binding(npb22.vlan, npb22.switch) self.assertEqual(len(npb_v200), 1) self._assert_equal(npb_v200[0], npb22) with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.get_nexusvlan_binding(npb21.vlan, "dummySwitch") def test_nexusvmbinding_get(self): npb11 = self._npb_test_obj(10, 100) npb21 = self._npb_test_obj(20, 100) npb22 = self._npb_test_obj(20, 200) self._add_to_db([npb11, npb21, npb22]) npb = nxdb.get_nexusvm_bindings(npb21.vlan, npb21.instance)[0] self._assert_equal(npb, npb21) npb = nxdb.get_nexusvm_bindings(npb22.vlan, npb22.instance)[0] self._assert_equal(npb, npb22) with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.get_nexusvm_bindings(npb21.vlan, "dummyInstance") def test_nexusportvlanswitchbinding_get(self): npb11 = self._npb_test_obj(10, 100) npb21 = self._npb_test_obj(20, 100) self._add_to_db([npb11, npb21]) npb = nxdb.get_port_vlan_switch_binding( npb11.port, npb11.vlan, npb11.switch) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb11) with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.get_port_vlan_switch_binding( npb21.port, npb21.vlan, "dummySwitch") def test_nexusportswitchbinding_get(self): npb11 = self._npb_test_obj(10, 100) npb21 = self._npb_test_obj(20, 100, switch='2.2.2.2') npb22 = self._npb_test_obj(20, 200, switch='2.2.2.2') self._add_to_db([npb11, npb21, npb22]) npb = nxdb.get_port_switch_bindings(npb11.port, npb11.switch) self.assertEqual(len(npb), 1) self._assert_equal(npb[0], npb11) npb_all_p20 = nxdb.get_port_switch_bindings(npb21.port, npb21.switch) self.assertEqual(len(npb_all_p20), 2) npb = nxdb.get_port_switch_bindings(npb21.port, "dummySwitch") self.assertIsNone(npb) def test_nexussvibinding_get(self): npbr1 = self._npb_test_obj('router', 100) npb21 = self._npb_test_obj(20, 100) self._add_to_db([npbr1, npb21]) npb_svi = nxdb.get_nexussvi_bindings() self.assertEqual(len(npb_svi), 1) self._assert_equal(npb_svi[0], npbr1) npbr2 = self._npb_test_obj('router', 200) self._add_to_db([npbr2]) npb_svi = nxdb.get_nexussvi_bindings() self.assertEqual(len(npb_svi), 2) def test_nexusbinding_update(self): npb11 = self._npb_test_obj(10, 100, switch='1.1.1.1', instance='test') npb21 = self._npb_test_obj(20, 100, switch='1.1.1.1', instance='test') self._add_to_db([npb11, npb21]) npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, '1.1.1.1') self.assertEqual(len(npb_all_v100), 2) npb22 = self._npb_test_obj(20, 200, switch='1.1.1.1', instance='test') npb = nxdb.update_nexusport_binding(npb21.port, 200) self._assert_equal(npb, npb22) npb_all_v100 = nxdb.get_nexusvlan_binding(npb11.vlan, '1.1.1.1') self.assertEqual(len(npb_all_v100), 1) self._assert_equal(npb_all_v100[0], npb11) npb = nxdb.update_nexusport_binding(npb21.port, 0) self.assertIsNone(npb) npb33 = self._npb_test_obj(30, 300, switch='1.1.1.1', instance='test') with testtools.ExpectedException(c_exc.NexusPortBindingNotFound): nxdb.update_nexusport_binding(npb33.port, 200)