merge tyler's unit tests for cisco plugin changes lp845140
This commit is contained in:
Dan Wendlandt
commit
77a9a71866
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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"""
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
@ -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')
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Reference in New Issue