diff --git a/quantum/plugins/cisco/common/cisco_constants.py b/quantum/plugins/cisco/common/cisco_constants.py index 2780e72bc1..3a99cb9f56 100644 --- a/quantum/plugins/cisco/common/cisco_constants.py +++ b/quantum/plugins/cisco/common/cisco_constants.py @@ -85,7 +85,7 @@ PASSWORD = 'password' LOGGER_COMPONENT_NAME = "cisco_plugin" -BLADE_INTF_DN = "blade-intf-distinguished-name" +BLADE_INTF_DN = "blade_intf_distinguished_name" BLADE_INTF_ORDER = "blade-intf-order" BLADE_INTF_LINK_STATE = "blade-intf-link-state" BLADE_INTF_OPER_STATE = "blade-intf-operational-state" diff --git a/quantum/plugins/cisco/models/l2network_multi_blade.py b/quantum/plugins/cisco/models/l2network_multi_blade.py index d74ac2d601..a3967025ad 100644 --- a/quantum/plugins/cisco/models/l2network_multi_blade.py +++ b/quantum/plugins/cisco/models/l2network_multi_blade.py @@ -73,14 +73,18 @@ class L2NetworkMultiBlade(L2NetworkModelBase): args) device_ips = device_params[const.DEVICE_IP] if not device_ips: - self._invoke_plugin(plugin_key, function_name, args, - device_params) + # Return in a list + return [self._invoke_plugin(plugin_key, function_name, args, + device_params)] else: + # Return a list of return values from each device + output = [] for device_ip in device_ips: new_device_params = deepcopy(device_params) new_device_params[const.DEVICE_IP] = device_ip - self._invoke_plugin(plugin_key, function_name, args, - new_device_params) + output.append(self._invoke_plugin(plugin_key, function_name, + args, new_device_params)) + return output def _invoke_inventory(self, plugin_key, function_name, args): """Invoke only the inventory implementation""" @@ -103,17 +107,25 @@ class L2NetworkMultiBlade(L2NetworkModelBase): def create_network(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) - self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + output = [] + ucs_output = self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), args) + nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + self._func_name(), args) + output.extend(ucs_output or []) + output.extend(nexus_output or []) + return output def delete_network(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) - self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + output = [] + ucs_output = self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), args) + nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + self._func_name(), args) + output.extend(ucs_output or []) + output.extend(nexus_output or []) + return output def get_network_details(self, args): """Not implemented for this model""" @@ -121,10 +133,14 @@ class L2NetworkMultiBlade(L2NetworkModelBase): def rename_network(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) - self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + output = [] + ucs_output = self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), args) + nexus_output = self._invoke_plugin_per_device(const.NEXUS_PLUGIN, + self._func_name(), args) + output.extend(ucs_output or []) + output.extend(nexus_output or []) + return output def get_all_ports(self, args): """Not implemented for this model""" @@ -132,13 +148,13 @@ class L2NetworkMultiBlade(L2NetworkModelBase): def create_port(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) + return self._invoke_plugin_per_device(const.UCS_PLUGIN, + self._func_name(), args) def delete_port(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) + return self._invoke_plugin_per_device(const.UCS_PLUGIN, + self._func_name(), args) def update_port(self, args): """Not implemented for this model""" @@ -150,13 +166,13 @@ class L2NetworkMultiBlade(L2NetworkModelBase): def plug_interface(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) + return self._invoke_plugin_per_device(const.UCS_PLUGIN, + self._func_name(), args) def unplug_interface(self, args): """Support for the Quantum core API call""" - self._invoke_plugin_per_device(const.UCS_PLUGIN, self._func_name(), - args) + return self._invoke_plugin_per_device(const.UCS_PLUGIN, + self._func_name(), args) def schedule_host(self, args): """Provides the hostname on which a dynamic vnic is reserved""" diff --git a/quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py b/quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py new file mode 100644 index 0000000000..6e46381db9 --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py @@ -0,0 +1,365 @@ +""" +# vim: tabstop=4 shiftwidth=4 softtabstop=4 +# +# Copyright 2011 Cisco Systems, Inc. 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. +# +# @author: Shubhangi Satras, Cisco Systems, Inc. +# @author: Peter Strunk, Cisco Systems, Inc. +# @author: Atul Gaikad, Cisco Systems, Inc. +# @author: Tyler Smith, Cisco Systems, Inc. +# +""" + +import unittest +import logging as LOG + +from quantum.common import exceptions as exc +from quantum.common import utils +from quantum.plugins.cisco import l2network_plugin_configuration as conf +from quantum.plugins.cisco.common import cisco_constants as const +from quantum.plugins.cisco.common import cisco_credentials as creds +from quantum.plugins.cisco.models import l2network_multi_blade +from quantum.plugins.cisco.db import api as db +from quantum.plugins.cisco.db import l2network_db as cdb + +LOG.basicConfig(level=LOG.WARN) +LOG.getLogger(__name__) + + +# Set some data to use in tests +tenant_id = "network_admin" +net_name = "TestNetwork1" +new_net_name = "NewTestNetwork1" +net_id = "44" +port_id = "p0005" +port_state = const.PORT_UP +interface_id = "vif-01" +vlan_id = "102" + + +def vlan_name(id): + return "q-%svlan" % id + + +class TestMultiBlade(unittest.TestCase): + """ + Tests for the multi-blade model for the L2Network plugin + """ + _plugins = {} + _inventory = {} + + def setUp(self): + """Setup our tests""" + # Initialize cdb and credentials + db.configure_db({'sql_connection': 'sqlite:///:memory:'}) + cdb.initialize() + creds.Store.initialize() + + # Create a place a store net and port ids for the druation of the test + self.net_id = 0 + self.port_id = 0 + + # Create the multiblade object + self._l2network_multiblade = l2network_multi_blade. \ + L2NetworkMultiBlade() + self.plugin_key = "quantum.plugins.cisco.ucs.cisco_ucs_plugin" + \ + ".UCSVICPlugin" + + # Get UCS inventory to make sure all UCSs are affected by tests + for key in conf.PLUGINS[const.PLUGINS].keys(): + if key in conf.PLUGINS[const.INVENTORY].keys(): + self._inventory[key] = utils.import_object( + conf.PLUGINS[const.INVENTORY][key]) + + self.ucs_count = self._inventory['ucs_plugin'].\ + _inventory.__len__() + + def tearDown(self): + """Tear down our tests""" + try: + port = db.port_get(self.net_id, self.port_id) + self._l2network_multiblade.delete_port([tenant_id, self.net_id, + self.port_id]) + except exc.NetworkNotFound: + # We won't always have a port to remove + pass + except exc.PortNotFound: + # We won't always have a port to remove + pass + + try: + net = db.network_get(self.net_id) + self._l2network_multiblade.delete_network([tenant_id, self.net_id]) + except exc.NetworkNotFound: + # We won't always have a network to remove + pass + db.clear_db() + + def test_create_network(self): + """Support for the Quantum core API call""" + LOG.debug("test_create_network - START") + + # Create the network in the test DB, then with the model + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + networks = self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + + self.assertEqual(networks.__len__(), self.ucs_count) + for network in networks: + self.assertEqual(network[const.NET_ID], self.net_id) + self.assertEqual(network[const.NET_NAME], net_name) + + LOG.debug("test_create_network - END") + + def test_create_networkDNE(self): + """Support for the Quantum core API call""" + LOG.debug("test_create_networkDNE - START") + + self.assertRaises(exc.NetworkNotFound, + self._l2network_multiblade.create_network, + [tenant_id, net_name, net_id, + vlan_name(net_id), vlan_id]) + + LOG.debug("test_create_networkDNE - END") + + def test_delete_network(self): + """Support for the Quantum core API call""" + LOG.debug("test_delete_network - START") + + # Create the network in the test DB, then with the model + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id) + + networks = self._l2network_multiblade.delete_network([tenant_id, + self.net_id]) + + self.assertEqual(networks.__len__(), self.ucs_count) + for network in networks: + self.assertEqual(network[const.NET_ID], self.net_id) + self.assertEqual(network[const.NET_NAME], net_name) + + LOG.debug("test_delete_network - END") + + def test_delete_networkDNE(self): + """Support for the Quantum core API call""" + LOG.debug("test_delete_networkDNE - START") + + self.assertRaises(exc.NetworkNotFound, + self._l2network_multiblade.delete_network, + [tenant_id, net_id]) + + LOG.debug("test_delete_networkDNE - END") + + def test_rename_network(self): + """Support for the Quantum core API call""" + LOG.debug("test_rename_network - START") + + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + + db.network_rename(tenant_id, self.net_id, new_net_name) + networks = self._l2network_multiblade.rename_network([tenant_id, + self.net_id, + new_net_name]) + + self.assertEqual(networks.__len__(), self.ucs_count) + for network in networks: + self.assertEqual(network[const.NET_ID], self.net_id) + self.assertEqual(network[const.NET_NAME], new_net_name) + LOG.debug("test_rename_network - END") + + def test_rename_networkDNE(self): + """Support for the Quantum core API call""" + LOG.debug("test_rename_networkDNE - START") + self.assertRaises(exc.NetworkNotFound, + self._l2network_multiblade.rename_network, + [tenant_id, net_id, new_net_name]) + LOG.debug("test_rename_networkDNE - END") + + def test_get_all_networks(self): + """Not implemented for this model""" + pass + + def test_get_network_details(self): + """Not implemented for this model""" + pass + + def test_create_port(self): + """Support for the Quantum core API call""" + LOG.debug("test_create_port - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + port = self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, + self.port_id]) + + self.assertEqual(self.port_id, port[0][const.PORTID]) + LOG.debug("test_create_port - END") + + def test_delete_port(self): + """Support for the Quantum core API call""" + LOG.debug("test_delete_port - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, self.port_id]) + + port = self._l2network_multiblade.delete_port([tenant_id, + self.net_id, + self.port_id]) + + self.assertEqual(self.port_id, port[0][const.PORTID]) + + # Recreating port so tear down doesn't cause an error + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, self.port_id]) + + LOG.debug("test_delete_port - END") + + def test_get_all_ports(self): + """Not implemented for this model""" + pass + + def test_update_port(self): + """Not implemented for this model""" + pass + + def test_update_portDNE(self): + """Not implemented for this model""" + pass + + def test_update_port_networkDNE(self): + """Not implemented for this model""" + pass + + def test_port_details(self): + """Not implemented for this model""" + pass + + def test_plug_interface(self): + """Support for the Quantum core API call""" + LOG.debug("test_plug_interface - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id) + + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, self.port_id]) + + interface = self._l2network_multiblade.plug_interface([tenant_id, + self.net_id, self.port_id, interface_id]) + port = db.port_set_attachment(self.net_id, self.port_id, interface_id) + + self.assertEqual(self.port_id, interface[0][const.PORTID]) + self.assertEqual(port[const.INTERFACEID], interface_id) + LOG.debug("test_plug_interface - END") + + def test_plug_interface_networkDNE(self): + """Support for the Quantum core API call""" + LOG.debug("test_plug_interface_networkDNE - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id) + + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, self.port_id]) + + self.assertRaises(exc.NetworkNotFound, + self._l2network_multiblade.plug_interface, + [tenant_id, net_id, self.port_id, interface_id]) + + LOG.debug("test_plug_interface_networkDNE - END") + + def test_plug_interface_portDNE(self): + """Support for the Quantum core API call""" + LOG.debug("test_plug_interface_portDNE - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id) + + self.assertRaises(exc.PortNotFound, + self._l2network_multiblade.plug_interface, + [tenant_id, self.net_id, port_id, interface_id]) + + LOG.debug("test_plug_interface_portDNE - START") + + def test_unplug_interface(self): + """Support for the Quantum core API call""" + LOG.debug("test_unplug_interface - START") + self.net_id = db.network_create(tenant_id, net_name)[const.UUID] + self._l2network_multiblade.create_network([tenant_id, + net_name, + self.net_id, + vlan_name(self.net_id), + vlan_id]) + cdb.add_vlan_binding(vlan_id, vlan_name(self.net_id), self.net_id) + + self.port_id = db.port_create(self.net_id, port_state)[const.UUID] + self._l2network_multiblade.create_port([tenant_id, + self.net_id, + port_state, self.port_id]) + + self._l2network_multiblade.plug_interface([tenant_id, self.net_id, + self.port_id, interface_id]) + db.port_set_attachment(self.net_id, self.port_id, interface_id) + interface = self._l2network_multiblade.unplug_interface([tenant_id, + self.net_id, self.port_id]) + + self.assertEqual(self.port_id, interface[0][const.PORTID]) + LOG.debug("test_unplug_interface - END") diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_inventory.py b/quantum/plugins/cisco/tests/unit/test_ucs_inventory.py new file mode 100644 index 0000000000..06f380615f --- /dev/null +++ b/quantum/plugins/cisco/tests/unit/test_ucs_inventory.py @@ -0,0 +1,201 @@ +""" +# vim: tabstop=4 shiftwidth=4 softtabstop=4 +# +# Copyright 2011 Cisco Systems, Inc. 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. +# +# @author: Shubhangi Satras, Cisco Systems, Inc. +# @author: Tyler Smith, Cisco Systems, Inc. +# +""" + +import unittest +import logging as LOG + +from quantum.common import exceptions as exc +from quantum.plugins.cisco.l2network_plugin import L2Network +from quantum.plugins.cisco.common import cisco_constants as const +from quantum.plugins.cisco.common import cisco_credentials as creds +from quantum.plugins.cisco.db import api as db +from quantum.plugins.cisco.db import l2network_db as cdb +from quantum.plugins.cisco.ucs.cisco_ucs_inventory import UCSInventory + +LOG.basicConfig(level=LOG.WARN) +LOG.getLogger(__name__) + +# Set some data to use in tests +tenant = 'shubh' +net_name = 'TestNetwork1' +port_state = const.PORT_UP +interface_id = 'vif-01' + + +class TestUCSInventory(unittest.TestCase): + """ + Tests for the UCS Inventory. Each high-level operation should return + some information about which devices to perform the action on. + """ + + def setUp(self): + """Setup our tests""" + # Initialize cdb and credentials + db.configure_db({'sql_connection': 'sqlite:///:memory:'}) + cdb.initialize() + creds.Store.initialize() + + # Create the ucs inventory object + self._ucs_inventory = UCSInventory() + self.inventory = self._ucs_inventory._inventory + + # Create a plugin instance to create networks/ports + self._l2network = L2Network() + + def assertValidUCM(self, ip_address): + """Asserts that the given ip is in the UCS inventory""" + if ip_address in self.inventory.keys(): + assert(1) + return + assert(0) + + def assertPortNotFound(self, cmd, args): + """Asserts that the given command raises a PortNotFound exception""" + cmd = getattr(self._ucs_inventory, cmd) + self.assertRaises(exc.PortNotFound, cmd, args) + + def _test_get_all_ucms(self, cmd): + """Runs tests for commands that expect a list of all UCMS""" + LOG.debug("test_%s - START", cmd) + results = getattr(self._ucs_inventory, cmd)([]) + self.assertEqual(results[const.DEVICE_IP], self.inventory.keys()) + LOG.debug("test_%s - END", cmd) + + def _test_with_port_creation(self, cmd, params=None): + """Tests commands that requires a port to exist""" + LOG.debug("test_%s - START", cmd) + net = self._l2network.create_network(tenant, net_name) + port = self._l2network.create_port(tenant, net[const.NET_ID], + port_state) + + args = [tenant, net[const.NET_ID], port[const.PORT_ID]] + if params is not None: + args.extend(params) + + ip_address = getattr(self._ucs_inventory, cmd)(args) + ip_address = ip_address[const.DEVICE_IP][0] + self.assertValidUCM(ip_address) + + # Clean up created network and port + try: + self._l2network.unplug_interface(tenant, + net[const.NET_ID], port[const.PORT_ID]) + except: + pass + self._l2network.delete_port(tenant, + net[const.NET_ID], port[const.PORT_ID]) + self._l2network.delete_network(tenant, net[const.NET_ID]) + db.clear_db() + + LOG.debug("test_%s - END", cmd) + + def _test_port_not_found(self, cmd, params=None): + """Tests commands that should raise a PortNotFound exception""" + # Figure out the correct name of this test + name = cmd + if name[-5:] == "_port": + name += "_not_found" + else: + name += "_port_not_found" + + LOG.debug("test_%s - START", name) + args = [tenant, 1, 1] + if params is not None: + args.extend(params) + + self.assertPortNotFound(cmd, args) + LOG.debug("test_%s - END", name) + + def test_create_port(self): + """Test that the UCS Inventory returns the correct devices to use""" + LOG.debug("test_create_port - START") + results = self._ucs_inventory.create_port([]) + results = results[const.LEAST_RSVD_BLADE_DICT] + + ip_address = results[const.LEAST_RSVD_BLADE_UCSM] + chassis = results[const.LEAST_RSVD_BLADE_CHASSIS] + blade = results[const.LEAST_RSVD_BLADE_ID] + + if blade not in self.inventory[ip_address][chassis]: + self.assertEqual(0, 1) + self.assertEqual(1, 1) + LOG.debug("test_create_port - END") + + def test_get_all_networks(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('get_all_networks') + + def test_create_network(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('create_network') + + def test_delete_network(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('delete_network') + + def test_get_network_details(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('get_network_details') + + def test_rename_network(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('rename_network') + + def test_get_all_ports(self): + """Test that the UCS Inventory returns the correct devices to use""" + self._test_get_all_ucms('get_all_ports') + + def test_delete_port(self): + """Test that the UCS Inventory returns a valid UCM""" + self._test_with_port_creation('delete_port') + + def test_get_port_details(self): + """Test that the UCS Inventory returns a valid UCM""" + self._test_with_port_creation('get_port_details') + + def test_update_port(self): + """Test that the UCS Inventory returns a valid UCM""" + self._test_with_port_creation('update_port', [port_state]) + + def test_plug_interface(self): + """Test that the UCS Inventory returns a valid UCM""" + self._test_with_port_creation('plug_interface', [interface_id]) + + def test_unplug_interface(self): + """Test that the UCS Inventory returns a valid UCM""" + self._test_with_port_creation('unplug_interface') + + def test_update_port_not_found(self): + """Test that the UCS Inventory raises a PortNotFound exception""" + self._test_port_not_found('update_port') + + def test_get_port_details_port_not_found(self): + """Test that the UCS Inventory raises a PortNotFound exception""" + self._test_port_not_found('get_port_details') + + def test_plug_interface_port_not_found(self): + """Test that the UCS Inventory raises a PortNotFound exception""" + self._test_port_not_found('plug_interface', [interface_id]) + + def test_unplug_interface_port_not_found(self): + """Test that the UCS Inventory raises a PortNotFound exception""" + self._test_port_not_found('unplug_interface') diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py index 3d6f19845a..6c1e54a701 100644 --- a/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py +++ b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py @@ -14,6 +14,7 @@ # under the License. # # @author: Shubhangi Satras, Cisco Systems, Inc. +# Shweta Padubidri, Cisco Systems, Inc. # import unittest import logging as LOG @@ -21,39 +22,54 @@ from quantum.common import exceptions as exc from quantum.plugins.cisco.common import cisco_constants as const from quantum.plugins.cisco.ucs import cisco_ucs_plugin from quantum.plugins.cisco.ucs import cisco_ucs_configuration as conf +from quantum.plugins.cisco.common import cisco_credentials as cred + +from quantum.plugins.cisco.db import api as db +from quantum.plugins.cisco.db import l2network_db as cdb +from quantum.plugins.cisco.common import cisco_exceptions as c_exc + +from quantum.plugins.cisco.ucs import cisco_ucs_inventory as ucsinv LOG.basicConfig(level=LOG.WARN) -LOG.getLogger("cisco_plugin") +LOG.getLogger("cisco_ucs_plugin") class UCSVICTestPlugin(unittest.TestCase): def setUp(self): - """ - Set up function. - """ self.tenant_id = "test_tenant_cisco12" self.net_name = "test_network_cisco12" self.net_id = 000011 - self.vlan_name = "q-" + str(self.net_id) + "vlan" - self.vlan_id = 266 + self.vlan_name = conf.DEFAULT_VLAN_NAME + self.vlan_id = conf.DEFAULT_VLAN_ID self.port_id = "4" + cdb.initialize() + cred.Store.initialize() self._cisco_ucs_plugin = cisco_ucs_plugin.UCSVICPlugin() + self.device_ip = conf.UCSM_IP_ADDRESS + self._ucs_inventory = ucsinv.UCSInventory() + self.chassis_id = '1' + self.blade_id = '5' + self.blade_intf_distinguished_name = 'sys/chassis-1/blade-5/'\ + 'adaptor-1/host-eth-6' def test_create_network(self): """ Tests creation of new Virtual Network. """ LOG.debug("UCSVICTestPlugin:_test_create_network() called\n") + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) new_net_dict = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) - self.assertEqual(new_net_dict[const.NET_ID], self.net_id) - self.assertEqual(new_net_dict[const.NET_NAME], self.net_name) - self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name) - self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id) - self.tearDownNetwork(self.tenant_id, self.net_id) + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + self.assertEqual(new_net_dict[const.NET_ID], new_network[const.UUID]) + self.assertEqual(new_net_dict[const.NET_NAME], + new_network[const.NETWORKNAME]) + self.tearDownNetwork(self.tenant_id, new_network[const.UUID]) def test_delete_network(self): """ @@ -61,12 +77,16 @@ class UCSVICTestPlugin(unittest.TestCase): belonging to the specified tenant. """ LOG.debug("UCSVICTestPlugin:test_delete_network() called\n") - self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) new_net_dict = self._cisco_ucs_plugin.delete_network( - self.tenant_id, self.net_id) - self.assertEqual(new_net_dict[const.NET_ID], self.net_id) + self.tenant_id, new_network[const.UUID], device_ip=self.device_ip) + self.assertEqual(new_net_dict[const.NET_ID], new_network[const.UUID]) def test_get_network_details(self): """ @@ -74,15 +94,20 @@ class UCSVICTestPlugin(unittest.TestCase): spec """ LOG.debug("UCSVICTestPlugin:test_get_network_details() called\n") - self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) new_net_dict = self._cisco_ucs_plugin.get_network_details( - self.tenant_id, self.net_id) - self.assertEqual(new_net_dict[const.NET_ID], self.net_id) - self.assertEqual(new_net_dict[const.NET_VLAN_NAME], self.vlan_name) - self.assertEqual(new_net_dict[const.NET_VLAN_ID], self.vlan_id) - self.tearDownNetwork(self.tenant_id, self.net_id) + self.tenant_id, new_network[const.UUID], + device_ip=self.device_ip) + self.assertEqual(new_net_dict[const.NET_ID], new_network[const.UUID]) + self.assertEqual(new_net_dict[const.NET_NAME], + new_network[const.NETWORKNAME]) + self.tearDownNetwork(self.tenant_id, new_network[const.UUID]) def test_get_all_networks(self): """ @@ -91,18 +116,28 @@ class UCSVICTestPlugin(unittest.TestCase): the specified tenant. """ LOG.debug("UCSVICTestPlugin:test_get_all_networks() called\n") + new_network1 = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network1[const.UUID]) new_net_dict1 = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) + self.tenant_id, new_network1[const.NETWORKNAME], + new_network1[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_network2 = db.network_create(self.tenant_id, "test_network2") + cdb.add_vlan_binding("6", "q-000006vlan", new_network2[const.UUID]) new_net_dict2 = self._cisco_ucs_plugin.create_network( - self.tenant_id, "test_network2", - 000006, "q-000006vlan", "6") - net_list = self._cisco_ucs_plugin.get_all_networks(self.tenant_id) + self.tenant_id, new_network2[const.NETWORKNAME], + new_network2[const.UUID], "q-000006vlan", "6", + device_ip=self.device_ip) + + net_list = self._cisco_ucs_plugin.get_all_networks( + self.tenant_id, device_ip=self.device_ip) net_id_list = [new_net_dict1, new_net_dict2] + self.assertTrue(net_list[0] in net_id_list) self.assertTrue(net_list[1] in net_id_list) - self.tearDownNetwork(self.tenant_id, new_net_dict1[const.NET_ID]) - self.tearDownNetwork(self.tenant_id, new_net_dict2[const.NET_ID]) + self.tearDownNetwork(self.tenant_id, new_network1[const.UUID]) + self.tearDownNetwork(self.tenant_id, new_network2[const.UUID]) def test_get_all_ports(self): """ @@ -110,74 +145,81 @@ class UCSVICTestPlugin(unittest.TestCase): specified Virtual Network. """ LOG.debug("UCSVICPlugin:get_all_ports() called\n") + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) new_net_dict = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port1 = db.port_create(new_network[const.UUID], const.PORT_UP) port_dict1 = self._cisco_ucs_plugin.create_port( - self.tenant_id, self.net_id, const.PORT_UP, - self.port_id) + self.tenant_id, self.net_id, const.PORT_UP, + new_port1[const.UUID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + new_port2 = db.port_create(new_network[const.UUID], const.PORT_UP) port_dict2 = self._cisco_ucs_plugin.create_port( - self.tenant_id, self.net_id, - const.PORT_UP, "10") + self.tenant_id, self.net_id, const.PORT_UP, + new_port2[const.UUID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) ports_on_net = self._cisco_ucs_plugin.get_all_ports( - self.tenant_id, self.net_id) + self.tenant_id, new_net_dict[const.NET_ID], + device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) port_list = [port_dict1, port_dict2] - self.assertTrue(port_list[0] in ports_on_net) - self.assertTrue(port_list[1] in ports_on_net) - self._cisco_ucs_plugin.delete_port(self.tenant_id, self.net_id, - self.port_id) - self.tearDownNetworkPort(self.tenant_id, new_net_dict[const.NET_ID], - port_dict2[const.PORT_ID]) + self.assertTrue(str(ports_on_net[1]) == str(port_list[1]) or + str(ports_on_net[1]) == str(port_list[0])) + self.assertTrue(str(ports_on_net[0]) == str(port_list[1]) or + str(ports_on_net[0]) == str(port_list[0])) - def _test_rename_network(self, new_name): - """ - Tests whether symbolic name is updated for the particular - Virtual Network. - """ - LOG.debug("UCSVICTestPlugin:_test_rename_network() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - new_net_dict = self._cisco_ucs_plugin.rename_network( - self.tenant_id, self.net_id, new_name) - self.assertEqual(new_net_dict[const.NET_NAME], new_name) - self.tearDownNetwork(self.tenant_id, self.net_id) + self._cisco_ucs_plugin.delete_port( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict1[const.PORTID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + chassis_id=self.chassis_id, blade_id=self.blade_id, + blade_intf_distinguished_name=self.\ + blade_intf_distinguished_name, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + self.tearDownNetworkPort( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict2[const.PORTID]) - def test_rename_network(self): - """ - Tests rename network. - """ - self._test_rename_network("new_test_network1") - - def _test_create_port(self, port_state): + def test_create_port(self): """ Tests creation of a port on the specified Virtual Network. """ LOG.debug("UCSVICTestPlugin:_test_create_port() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - new_port_dict = self._cisco_ucs_plugin.create_port( - self.tenant_id, self.net_id, port_state, self.port_id) - self.assertEqual(new_port_dict[const.PORT_ID], self.port_id) - self.assertEqual(new_port_dict[const.PORT_STATE], port_state) - self.assertEqual(new_port_dict[const.ATTACHMENT], None) - profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id) - new_port_profile = new_port_dict[const.PORT_PROFILE] - self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME], - conf.DEFAULT_VLAN_NAME) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID], - conf.DEFAULT_VLAN_ID) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) - def test_create_port(self): - """ - Tests create port. - """ - self._test_create_port(const.PORT_UP) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + port_dict = self._cisco_ucs_plugin.create_port( + self.tenant_id, self.net_id, const.PORT_UP, + new_port[const.UUID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + self.assertEqual(port_dict[const.PORTID], new_port[const.UUID]) + profile_name = self._cisco_ucs_plugin.\ + _get_profile_name(port_dict[const.PORTID]) + self.assertTrue(profile_name != None) + self.tearDownNetworkPort( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID]) - def _test_delete_port(self, port_state): + def test_delete_port(self): """ Tests Deletion of a port on a specified Virtual Network, if the port contains a remote interface attachment, @@ -185,378 +227,255 @@ class UCSVICTestPlugin(unittest.TestCase): then the port can be deleted. """ LOG.debug("UCSVICTestPlugin:_test_delete_port() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - port_state, self.port_id) - self._cisco_ucs_plugin.delete_port(self.tenant_id, self.net_id, - self.port_id) - net = self._cisco_ucs_plugin._get_network(self.tenant_id, self.net_id) - self.assertEqual(net[const.NET_PORTS], {}) - self.tearDownNetwork(self.tenant_id, self.net_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + port_dict = self._cisco_ucs_plugin.create_port( + self.tenant_id, self.net_id, const.PORT_UP, + new_port[const.UUID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + port_bind = self._cisco_ucs_plugin.delete_port( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + chassis_id=self.chassis_id, blade_id=self.blade_id, + blade_intf_distinguished_name=self.\ + blade_intf_distinguished_name, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) - def test_delete_port(self): - """ - Tests delete port. - """ - self._test_delete_port(const.PORT_UP) + self.assertEqual(port_bind[const.PORTID], new_port[const.UUID]) + self.tearDownNetwork(self.tenant_id, new_net_dict[const.NET_ID]) - def _test_update_port(self, port_state): - """ - Tests Updation of the state of a port on the specified Virtual Network. - """ - LOG.debug("UCSVICTestPlugin:_test_update_port() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - port_state, self.port_id) - port = self._cisco_ucs_plugin.update_port( - self.tenant_id, self.net_id, - self.port_id, port_state) - self.assertEqual(port[const.PORT_STATE], port_state) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) - - def test_update_port_state_up(self): - """ - Tests update port state up - """ - self._test_update_port(const.PORT_UP) - - def test_update_port_state_down(self): - """ - Tests update port state down - """ - self._test_update_port(const.PORT_DOWN) - - def _test_get_port_details_state_up(self, port_state): + def _test_get_port_details(self, port_state): """ Tests whether user is able to retrieve a remote interface that is attached to this particular port when port state is Up. """ - LOG.debug("UCSVICTestPlugin:_test_get_port_details_state_up()" + - "called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - port_state, self.port_id) - port = self._cisco_ucs_plugin.get_port_details( - self.tenant_id, self.net_id, self.port_id) - self.assertEqual(port[const.PORT_ID], self.port_id) - self.assertEqual(port[const.PORT_STATE], port_state) - self.assertEqual(port[const.ATTACHMENT], None) - new_port_profile = port[const.PORT_PROFILE] - profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME], - conf.DEFAULT_VLAN_NAME) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID], - conf.DEFAULT_VLAN_ID) - self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) + LOG.debug("UCSVICTestPlugin:_test_get_port_details() called\n") + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port = db.port_create(new_network[const.UUID], port_state) + port_dict = self._cisco_ucs_plugin.create_port( + self.tenant_id, self.net_id, port_state, + new_port[const.UUID], device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) - def _test_show_port_state_down(self, port_state): - """ - Tests whether user is able to retrieve a remote interface - that is attached to this particular port when port state is down. - """ - LOG.debug("UCSVICTestPlugin:_test_show_port_state_down()" + - "called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - port_state, self.port_id) - port = self._cisco_ucs_plugin.get_port_details(self.tenant_id, - self.net_id, - self.port_id) - self.assertEqual(port[const.PORT_ID], self.port_id) - self.assertNotEqual(port[const.PORT_STATE], port_state) - self.assertEqual(port[const.ATTACHMENT], None) - new_port_profile = port[const.PORT_PROFILE] - profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME], - conf.DEFAULT_VLAN_NAME) - self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID], - conf.DEFAULT_VLAN_ID) - self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) + port_detail = self._cisco_ucs_plugin.get_port_details( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID], device_ip=self.device_ip) + self.assertEqual(str(port_dict), str(port_detail)) + self.tearDownNetworkPort( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID]) def test_get_port_details_state_up(self): - """ - Tests get port details state up - """ - self._test_get_port_details_state_up(const.PORT_UP) + self._test_get_port_details(const.PORT_UP) def test_show_port_state_down(self): - """ - Tests show port state down - """ - self._test_show_port_state_down(const.PORT_DOWN) + self._test_get_port_details(const.PORT_DOWN) def test_create_port_profile(self): - """ - Tests create port profile - """ LOG.debug("UCSVICTestPlugin:test_create_port_profile() called\n") + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + self._cisco_ucs_plugin._set_ucsm(self.device_ip) new_port_profile = self._cisco_ucs_plugin._create_port_profile( - self.tenant_id, self.net_id, self.port_id, - self.vlan_name, self.vlan_id) - profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id) + self.tenant_id, new_network[const.UUID], + new_port[const.UUID], self.vlan_name, + self.vlan_id) + profile_name = self._cisco_ucs_plugin.\ + _get_profile_name(new_port[const.UUID]) self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name) self.assertEqual(new_port_profile[const.PROFILE_VLAN_NAME], - self.vlan_name) + self.vlan_name) self.assertEqual(new_port_profile[const.PROFILE_VLAN_ID], self.vlan_id) - self._cisco_ucs_plugin._delete_port_profile(self.port_id, profile_name) + self._cisco_ucs_plugin._delete_port_profile(new_port[const.UUID], + profile_name) def test_delete_port_profile(self): - """ - Tests delete port profile - """ LOG.debug("UCSVICTestPlugin:test_delete_port_profile() called\n") + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + self._cisco_ucs_plugin._set_ucsm(self.device_ip) self._cisco_ucs_plugin._create_port_profile( - self.tenant_id, self.net_id, self.port_id, self.vlan_name, - self.vlan_id) - profile_name = self._cisco_ucs_plugin._get_profile_name(self.port_id) - counter1 = self._cisco_ucs_plugin._port_profile_counter - self._cisco_ucs_plugin._delete_port_profile(self.port_id, - profile_name) - counter2 = self._cisco_ucs_plugin._port_profile_counter - self.assertNotEqual(counter1, counter2) + self.tenant_id, new_network[const.UUID], + new_port[const.UUID], self.vlan_name, + self.vlan_id) + profile_name = self._cisco_ucs_plugin.\ + _get_profile_name(new_port[const.UUID]) - def _test_plug_interface(self, remote_interface_id): + counter1 = self._cisco_ucs_plugin._port_profile_counter + self._cisco_ucs_plugin._delete_port_profile(new_port[const.UUID], + profile_name) + counter2 = self._cisco_ucs_plugin._port_profile_counter + self.assertEqual(counter1 - 1, counter2) + + def test_plug_interface(self, remote_interface_id=None, + new_vlanid=10, new_vlan_name='new_vlan'): """ Attaches a remote interface to the specified port on the specified Virtual Network. """ LOG.debug("UCSVICTestPlugin:_test_plug_interface() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id, - self.port_id, - remote_interface_id) - port = self._cisco_ucs_plugin._get_port( - self.tenant_id, self.net_id, self.port_id) - self.assertEqual(port[const.ATTACHMENT], remote_interface_id) - port_profile = port[const.PORT_PROFILE] - new_vlan_name = self._cisco_ucs_plugin._get_vlan_name_for_network( - self.tenant_id, self.net_id) - new_vlan_id = self._cisco_ucs_plugin._get_vlan_id_for_network( - self.tenant_id, self.net_id) - self.assertEqual(port_profile[const.PROFILE_VLAN_NAME], new_vlan_name) - self.assertEqual(port_profile[const.PROFILE_VLAN_ID], new_vlan_id) - self.tearDownNetworkPortInterface(self.tenant_id, self.net_id, - self.port_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + port_dict = self._cisco_ucs_plugin.create_port( + self.tenant_id, new_net_dict[const.NET_ID], + const.PORT_UP, new_port[const.UUID], + device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + cdb.update_vlan_binding(new_network[const.UUID], + str(new_vlanid), new_vlan_name) + port_bind = self._cisco_ucs_plugin.plug_interface( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID], remote_interface_id, + device_ip=self.device_ip) + self.assertEqual(port_bind[const.VLANNAME], new_vlan_name) + self.assertEqual(port_bind[const.VLANID], new_vlanid) + self.tearDownNetworkPortInterface( + self.tenant_id, new_net_dict[const.NET_ID], + new_port[const.UUID]) - def test_plug_interface(self): - """ - Tests test plug interface - """ - self._test_plug_interface("4") - - def _test_unplug_interface(self, remote_interface_id): + def test_unplug_interface(self, remote_interface_id=None, + new_vlanid=10, new_vlan_name='new_vlan'): """ Tests whether remote interface detaches from the specified port on the specified Virtual Network. """ LOG.debug("UCSVICTestPlugin:_test_unplug_interface() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id, - self.port_id, - remote_interface_id) - self._cisco_ucs_plugin.unplug_interface(self.tenant_id, self.net_id, - self.port_id) - port = self._cisco_ucs_plugin._get_port( - self.tenant_id, self.net_id, self.port_id) - self.assertEqual(port[const.ATTACHMENT], None) - port_profile = port[const.PORT_PROFILE] - self.assertEqual(port_profile[const.PROFILE_VLAN_NAME], - conf.DEFAULT_VLAN_NAME) - self.assertEqual(port_profile[const.PROFILE_VLAN_ID], - conf.DEFAULT_VLAN_ID) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) + new_port = db.port_create(new_network[const.UUID], const.PORT_UP) + port_dict = self._cisco_ucs_plugin.create_port( + self.tenant_id, new_net_dict[const.NET_ID], + const.PORT_UP, new_port[const.UUID], + device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) + cdb.update_vlan_binding(new_network[const.UUID], + str(new_vlanid), new_vlan_name) + self._cisco_ucs_plugin.plug_interface( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID], remote_interface_id, + device_ip=self.device_ip) - def test_unplug_interface(self): - """ - Tests unplug interface - """ - self._test_unplug_interface("4") + port_bind = self._cisco_ucs_plugin.unplug_interface( + self.tenant_id, new_net_dict[const.NET_ID], + port_dict[const.PORTID], device_ip=self.device_ip) + self.assertEqual(port_bind[const.VLANNAME], self.vlan_name) + self.assertEqual(port_bind[const.VLANID], self.vlan_id) + self.tearDownNetworkPortInterface( + self.tenant_id, new_net_dict[const.NET_ID], + new_port[const.UUID]) def test_get_vlan_name_for_network(self): - """ - Tests get vlan name for network - """ LOG.debug("UCSVICTestPlugin:test_get_vlan_name_for_network() called\n") - net = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, - self.vlan_name, self.vlan_id) - self.assertEqual(net[const.NET_VLAN_NAME], self.vlan_name) - self.tearDownNetwork(self.tenant_id, self.net_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + vlan_bind_name = self._cisco_ucs_plugin._get_vlan_name_for_network( + self.tenant_id, new_network[const.UUID]) + + self.assertEqual(vlan_bind_name, self.vlan_name) def test_get_vlan_id_for_network(self): - """ - Tests get vlan id for network - """ LOG.debug("UCSVICTestPlugin:test_get_vlan_id_for_network() called\n") - net = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, self.vlan_name, - self.vlan_id) - self.assertEqual(net[const.NET_VLAN_ID], self.vlan_id) - self.tearDownNetwork(self.tenant_id, self.net_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + vlan_bind_id = self._cisco_ucs_plugin._get_vlan_id_for_network( + self.tenant_id, new_network[const.UUID]) + self.assertEqual(str(vlan_bind_id), self.vlan_id) - def test_get_network(self): - """ - Tests get network - """ - LOG.debug("UCSVICTestPlugin:test_get_network() called\n") - net = self._cisco_ucs_plugin.create_network( - self.tenant_id, self.net_name, self.net_id, self.vlan_name, - self.vlan_id) - self.assertEqual(net[const.NET_ID], self.net_id) - self.tearDownNetwork(self.tenant_id, self.net_id) - - def test_get_port(self): - """ - Tests get port - """ - LOG.debug("UCSVICTestPlugin:test_get_port() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - new_port_dict = self._cisco_ucs_plugin.create_port( - self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self.assertEqual(new_port_dict[const.PORT_ID], self.port_id) - self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id) - - def test_get_network_NetworkNotFound(self): - """ - Tests get network not found - """ + def test_show_network_NetworkNotFound(self): self.assertRaises(exc.NetworkNotFound, - self._cisco_ucs_plugin._get_network, - self.tenant_id, self.net_id) + self._cisco_ucs_plugin.get_network_details, + self.tenant_id, self.net_id, + device_ip=self.device_ip) def test_delete_network_NetworkNotFound(self): - """ - Tests delete network not found - """ self.assertRaises(exc.NetworkNotFound, self._cisco_ucs_plugin.delete_network, - self.tenant_id, self.net_id) - - def test_delete_port_PortInUse(self): - """ - Tests delete port in use - """ - self._test_delete_port_PortInUse("4") - - def _test_delete_port_PortInUse(self, remote_interface_id): - """ - Tests delete port in use - """ - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id, - self.port_id, - remote_interface_id) - self.assertRaises(exc.PortInUse, self._cisco_ucs_plugin.delete_port, - self.tenant_id, self.net_id, self.port_id) - self.tearDownNetworkPortInterface(self.tenant_id, self.net_id, - self.port_id) + self.tenant_id, self.net_id, + device_ip=self.device_ip) def test_delete_port_PortNotFound(self): - """ - Tests delete port not found - """ - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self.assertRaises(exc.PortNotFound, self._cisco_ucs_plugin.delete_port, - self.tenant_id, self.net_id, self.port_id) - self.tearDownNetwork(self.tenant_id, self.net_id) + new_network = db.network_create(self.tenant_id, self.net_name) + cdb.add_vlan_binding(str(self.vlan_id), self.vlan_name, + new_network[const.UUID]) + new_net_dict = self._cisco_ucs_plugin.create_network( + self.tenant_id, new_network[const.NETWORKNAME], + new_network[const.UUID], self.vlan_name, self.vlan_id, + device_ip=self.device_ip) - def test_plug_interface_PortInUse(self): - """ - Tests plug interface port in use - """ - self._test_plug_interface_PortInUse("6", "5") + self.assertRaises(c_exc.PortVnicNotFound, + self._cisco_ucs_plugin.delete_port, + self.tenant_id, new_net_dict[const.NET_ID], + self.port_id, device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + chassis_id=self.chassis_id, blade_id=self.blade_id, + blade_intf_distinguished_name=self.\ + blade_intf_distinguished_name, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) - def _test_plug_interface_PortInUse(self, remote_interface_id1, - remote_interface_id2): - """ - Tests plug interface port in use - """ - LOG.debug("UCSVICTestPlugin:_test_plug_interface_PortInUse() called\n") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id, - self.port_id, - remote_interface_id1) - self.assertRaises(exc.PortInUse, self._cisco_ucs_plugin.plug_interface, - self.tenant_id, self.net_id, self.port_id, - remote_interface_id2) - self.tearDownNetworkPortInterface(self.tenant_id, self.net_id, - self.port_id) + self.tearDownNetwork(self.tenant_id, new_net_dict[const.NET_ID]) - def test_attachment_exists(self): - """ - Tests attachment exists - """ - LOG.debug("UCSVICTestPlugin:testValidateAttachmentAlreadyAttached") - self._test_attachment_exists("4") - - def _test_attachment_exists(self, remote_interface_id): - """ - Tests attachment exists - """ - LOG.debug("UCSVICTestPlugin:_test_validate_attachmentAlreadyAttached") - self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name, - self.net_id, self.vlan_name, - self.vlan_id) - self._cisco_ucs_plugin.create_port(self.tenant_id, self.net_id, - const.PORT_UP, self.port_id) - self._cisco_ucs_plugin.plug_interface(self.tenant_id, self.net_id, - self.port_id, - remote_interface_id) - self.assertRaises( - exc.PortInUse, self._cisco_ucs_plugin._validate_attachment, - self.tenant_id, self.net_id, self.port_id, remote_interface_id) - self.tearDownNetworkPortInterface(self.tenant_id, self.net_id, - self.port_id) + def tearDown(self): + """Clear the test environment""" + # Remove database contents + db.clear_db() def tearDownNetwork(self, tenant_id, net_id): - """ - Tear down network - """ - self._cisco_ucs_plugin.delete_network(tenant_id, net_id) + self._cisco_ucs_plugin.delete_network(tenant_id, net_id, + device_ip=self.device_ip) def tearDownNetworkPort(self, tenant_id, net_id, port_id): - """ - Tear down network port - """ - self._cisco_ucs_plugin.delete_port(tenant_id, net_id, - port_id) + self._cisco_ucs_plugin.delete_port( + tenant_id, net_id, port_id, device_ip=self.device_ip, + ucs_inventory=self._ucs_inventory, + chassis_id=self.chassis_id, blade_id=self.blade_id, + blade_intf_distinguished_name=self.\ + blade_intf_distinguished_name, + least_rsvd_blade_dict=self._ucs_inventory.\ + _get_least_reserved_blade()) self.tearDownNetwork(tenant_id, net_id) def tearDownNetworkPortInterface(self, tenant_id, net_id, port_id): - """ - Tear down network port interface - """ - self._cisco_ucs_plugin.unplug_interface(tenant_id, net_id, - port_id) + self._cisco_ucs_plugin.unplug_interface( + tenant_id, net_id, port_id, + device_ip=self.device_ip) self.tearDownNetworkPort(tenant_id, net_id, port_id)