openstack/neutron
Code Issues Proposed changes

merge tyler's unit tests for cisco plugin changes lp845140

Browse Source
This commit is contained in:
Dan Wendlandt 2011-09-09 16:38:20 -07:00
commit 6080d44f34
5 changed files with 926 additions and 425 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
quantum/plugins/cisco/common/cisco_constants.py
View File

@ -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"

58
quantum/plugins/cisco/models/l2network_multi_blade.py
View File

@ -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"""

365
quantum/plugins/cisco/tests/unit/test_l2network_multi_blade.py Normal file
View File

@ -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")

201
quantum/plugins/cisco/tests/unit/test_ucs_inventory.py Normal file
View File

@ -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')

725
quantum/plugins/cisco/tests/unit/test_ucs_plugin.py
View File

@ -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)