Files
vmware-nsx/quantum/plugins/cisco/tests/unit/test_l2networkApi.py
Brad Hall 9bab0960b6 Second round of packaging changes
This change condenses the directory structure to something more similar to
what we had before while producing similar packages.

It also introduces version.py which allows us to get the version from git tags
(or a fallback version if not available).

Fixes lp bug 889336
Fixes lp bug 888795

Change-Id: I86136bd9dbabb5eb1f8366ed665ed9b54f695124
2011-11-28 10:33:52 -08:00

1067 lines
47 KiB
Python

# 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: Shweta Padubidri, Cisco Systems, Inc.
#
import logging
import unittest
from quantum.common import exceptions as exc
from quantum.plugins.cisco.common import cisco_constants as const
from quantum.plugins.cisco.common import cisco_exceptions as cexc
from quantum.plugins.cisco import l2network_plugin
from quantum.plugins.cisco import l2network_plugin_configuration as conf
from quantum.plugins.cisco.db import api as db
from quantum.plugins.cisco.db import l2network_db as cdb
LOG = logging.getLogger('quantum.tests.test_core_api_func')
class CoreAPITestFunc(unittest.TestCase):
def test_create_network(self, net_tenant_id=None, net_name=None):
"""
Tests creation of new Virtual Network.
"""
LOG.debug("test_create_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if net_name:
network_name = net_name
else:
network_name = self.network_name
new_net_dict = self._l2network_plugin.create_network(
tenant_id, network_name)
net = db.network_get(new_net_dict[const.NET_ID])
self.assertEqual(net[const.NETWORKNAME], network_name)
self.assertEqual(new_net_dict[const.NET_NAME], network_name)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_create_network - END")
def test_delete_network(self, net_tenant_id=None):
"""
Tests deletion of a Virtual Network.
"""
LOG.debug("test_delete_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
delete_net_dict = self._l2network_plugin.delete_network(
tenant_id, new_net_dict[const.NET_ID])
self.assertRaises(exc.NetworkNotFound, db.network_get,
new_net_dict[const.NET_ID])
self.assertEqual(
new_net_dict[const.NET_ID], delete_net_dict[const.NET_ID])
LOG.debug("test_delete_network - END")
def test_delete_networkDNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests deletion of a Virtual Network when Network does not exist.
"""
LOG.debug("test_delete_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(
exc.NetworkNotFound, self._l2network_plugin.delete_network,
tenant_id, net_id)
LOG.debug("test_delete_network_not_found - END")
def test_delete_networkInUse(
self, tenant_id='test_tenant', instance_tenant_id='nova',
nova_user_id='novaadmin', instance_id=10,
vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests deletion of a Virtual Network when Network is in Use.
"""
LOG.debug("test_delete_networkInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
instance_id,
instance_desc)
instance_vif_desc = {'project_id': tenant_id,
'user_id': nova_user_id,
'vif_id': vif_id}
vif_description = self._l2network_plugin.associate_port(
instance_tenant_id, instance_id,
instance_vif_desc)
self.assertRaises(exc.NetworkInUse,
self._l2network_plugin.delete_network, tenant_id,
new_net_dict[const.NET_ID])
self.tearDownNetworkPortInterface(
tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_delete_networkInUse - END")
def test_show_network(self, net_tenant_id=None):
"""
Tests display of details of a Virtual Network .
"""
LOG.debug("test_show_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
result_net_dict = self._l2network_plugin.get_network_details(
tenant_id, new_net_dict[const.NET_ID])
net = db.network_get(new_net_dict[const.NET_ID])
self.assertEqual(net[const.UUID], new_net_dict[const.NET_ID])
self.assertEqual(
new_net_dict[const.NET_ID], result_net_dict[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_show_network - END")
def test_show_networkDNE(self, net_tenant_id=None, net_id='0005'):
"""
Tests display of a Virtual Network when Network does not exist.
"""
LOG.debug("test_show_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.get_network_details,
tenant_id, net_id)
LOG.debug("test_show_network_not_found - END")
def test_update_network(self, net_tenant_id=None,
new_name='new_test_network'):
"""
Tests rename of a Virtual Network .
"""
LOG.debug("test_update_network - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
rename_net_dict = self._l2network_plugin.update_network(
tenant_id, new_net_dict[const.NET_ID],
name=new_name)
net = db.network_get(new_net_dict[const.NET_ID])
self.assertEqual(net[const.NETWORKNAME], new_name)
self.assertEqual(new_name, rename_net_dict[const.NET_NAME])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_update_network - END")
def test_update_networkDNE(self, net_tenant_id=None,
net_id='0005', new_name='new_test_network'):
"""
Tests update of a Virtual Network when Network does not exist.
"""
LOG.debug("test_update_network_not_found - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_network,
tenant_id, net_id, name=new_name)
LOG.debug("test_update_network_not_found - END")
def test_list_networks(self, tenant_id='test_network'):
"""
Tests listing of all the Virtual Networks .
"""
LOG.debug("test_list_networks - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
new_net_dict2 = self._l2network_plugin.create_network(
tenant_id, 'test_net2')
net_list = self._l2network_plugin.get_all_networks(tenant_id)
net_temp_list = [new_net_dict, new_net_dict2]
networks_list = db.network_list(tenant_id)
new_networks_list = []
for network in networks_list:
new_network_dict = self._make_net_dict(network[const.UUID],
network[const.NETWORKNAME],
[])
new_networks_list.append(new_network_dict)
self.assertEqual(len(net_list), 2)
self.assertTrue(net_list[0] in net_temp_list)
self.assertTrue(net_list[1] in net_temp_list)
self.assertTrue(new_networks_list[0] in net_temp_list)
self.assertTrue(new_networks_list[1] in net_temp_list)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
self.tearDownNetwork(tenant_id, new_net_dict2[const.NET_ID])
LOG.debug("test_list_networks - END")
def test_list_ports(self, tenant_id='test_network'):
"""
Tests listing of all the Ports.
"""
LOG.debug("test_list_ports - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
port_dict2 = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
port_list = self._l2network_plugin.get_all_ports(
tenant_id, new_net_dict[const.NET_ID])
port_temp_list = [port_dict, port_dict2]
network = db.network_get(new_net_dict[const.NET_ID])
ports_list = network[const.NETWORKPORTS]
ports_on_net = []
for port in ports_list:
new_port = self._make_port_dict(port[const.UUID],
port[const.PORTSTATE],
port[const.NETWORKID],
port[const.INTERFACEID])
ports_on_net.append(new_port)
self.assertEqual(len(port_list), 2)
self.assertTrue(port_list[0] in port_temp_list)
self.assertTrue(port_list[1] in port_temp_list)
self.assertTrue(ports_on_net[0] in port_temp_list)
self.assertTrue(ports_on_net[1] in port_temp_list)
self.tearDownPortOnly(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict2[const.PORT_ID])
LOG.debug("test_list_ports - END")
def test_create_port(self, tenant_id='test_network',
state=const.PORT_UP):
"""
Tests creation of Ports.
"""
LOG.debug("test_create_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], state)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(port_dict[const.PORT_STATE], state)
self.assertEqual(port_dict[const.NET_ID], new_net_dict[const.NET_ID])
self.assertEqual(port[const.PORTSTATE], state)
self.assertEqual(port[const.NETWORKID], new_net_dict[const.NET_ID])
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_create_port - END")
def test_create_port_network_DNE(
self, net_tenant_id=None, net_id='0005', state=const.PORT_UP):
"""
Tests creation of Ports when network does not exist.
"""
LOG.debug("test_create_port_network_DNE - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.create_port,
tenant_id, net_id, state)
LOG.debug("test_create_port_network_DNE - END:")
def test_delete_port(self, tenant_id='test_tenant',
state=const.PORT_UP):
"""
Tests deletion of Ports
"""
LOG.debug("test_delete_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
state)
delete_port_dict = self._l2network_plugin.delete_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertRaises(exc.PortNotFound, db.port_get,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(delete_port_dict[const.PORT_ID],
port_dict[const.PORT_ID])
LOG.debug("test_delete_port - END")
def test_delete_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests deletion of Ports when network does not exist.
"""
LOG.debug("test_delete_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.delete_port, tenant_id,
net_id, port_id)
LOG.debug("test_delete_port_networkDNE - END")
def test_delete_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests deletion of Ports when port does not exist.
"""
LOG.debug("test_delete_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(exc.PortNotFound, self._l2network_plugin.delete_port,
tenant_id, new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_delete_portDNE - END")
def test_delete_portInUse(
self, tenant_id='test_tenant', instance_tenant_id='nova',
nova_user_id='novaadmin', instance_id=10,
vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests deletion of Ports when port is in Use.
"""
LOG.debug("test_delete_portInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
instance_id,
instance_desc)
instance_vif_desc = {'project_id': tenant_id,
'user_id': nova_user_id,
'vif_id': vif_id}
vif_description = self._l2network_plugin.associate_port(
instance_tenant_id, instance_id,
instance_vif_desc)
self.assertRaises(exc.PortInUse,
self._l2network_plugin.delete_port, tenant_id,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
self.tearDownNetworkPortInterface(
tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_delete_portInUse - END")
def test_update_port(self, tenant_id='test_tenant',
state=const.PORT_DOWN):
"""
Tests updation of Ports.
"""
LOG.debug("test_update_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
update_port_dict = self._l2network_plugin.update_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], state=state)
new_port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(new_port[const.PORTSTATE], state)
self.assertEqual(update_port_dict[const.PORT_STATE], state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_update_port - END")
def test_update_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests updation of Ports when network does not exist.
"""
LOG.debug("test_update_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_port, tenant_id,
net_id, port_id, state=const.PORT_UP)
LOG.debug("test_update_port_networkDNE - END")
def test_update_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests updation of Ports when port does not exist.
"""
LOG.debug("test_update_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(
exc.PortNotFound, self._l2network_plugin.update_port, tenant_id,
new_net_dict[const.NET_ID], port_id, state=const.PORT_UP)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_update_portDNE - END")
def test_show_port(self, tenant_id='test_tenant'):
"""
Tests display of Ports
"""
LOG.debug("test_show_port - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
get_port_dict = self._l2network_plugin.get_port_details(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(port[const.PORTSTATE], self.state)
self.assertEqual(get_port_dict[const.PORT_STATE], self.state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_show_port - END")
def test_show_port_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests display of Ports when network does not exist
"""
LOG.debug("test_show_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.get_port_details,
tenant_id, net_id, port_id)
LOG.debug("test_show_port_networkDNE - END")
def test_show_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
"""
Tests display of Ports when port does not exist
"""
LOG.debug("test_show_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(exc.PortNotFound,
self._l2network_plugin.get_port_details, tenant_id,
new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_show_portDNE - END")
def test_plug_interface(
self, tenant_id='test_tenant', instance_tenant_id='nova',
nova_user_id='novaadmin', instance_id=10,
vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests attachment of interface to the port
"""
LOG.debug("test_plug_interface - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
instance_id,
instance_desc)
instance_vif_desc = {'project_id': tenant_id,
'user_id': nova_user_id,
'vif_id': vif_id}
vif_description = self._l2network_plugin.associate_port(
instance_tenant_id, instance_id,
instance_vif_desc)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], vif_id)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(port[const.INTERFACEID], vif_id)
self.tearDownNetworkPortInterface(
tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_plug_interface - END")
def test_plug_interface_networkDNE(
self, tenant_id='test_tenant', net_id='0005',
port_id='p0005', remote_interface='new_interface'):
"""
Tests attachment of interface network does not exist
"""
LOG.debug("test_plug_interface_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.plug_interface, tenant_id,
net_id, port_id, remote_interface)
LOG.debug("test_plug_interface_networkDNE - END")
def test_plug_interface_portDNE(self, tenant_id='test_tenant',
port_id='p0005',
remote_interface='new_interface'):
"""
Tests attachment of interface port does not exist
"""
LOG.debug("test_plug_interface_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
self.assertRaises(
exc.PortNotFound, self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID], port_id, remote_interface)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_plug_interface_portDNE - END")
def test_plug_interface_portInUse(
self, tenant_id='test_tenant', instance_tenant_id='nova',
nova_user_id='novaadmin', instance_id=10,
vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc',
remote_interface='new_interface'):
"""
Tests attachment of new interface to the port when there is an
existing attachment
"""
LOG.debug("test_plug_interface_portInUse - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
instance_id,
instance_desc)
instance_vif_desc = {'project_id': tenant_id,
'user_id': nova_user_id,
'vif_id': vif_id}
vif_description = self._l2network_plugin.associate_port(
instance_tenant_id, instance_id,
instance_vif_desc)
self.assertRaises(exc.PortInUse,
self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.tearDownNetworkPortInterface(
tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_plug_interface_portInUse - END")
def test_unplug_interface(
self, tenant_id='test_tenant', instance_tenant_id='nova',
nova_user_id='novaadmin', instance_id=10,
vif_id='fe701ddf-26a2-42ea-b9e6-7313d1c522cc'):
"""
Tests detaachment of an interface to a port
"""
LOG.debug("test_unplug_interface - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
instance_desc = {'project_id': tenant_id,
'user_id': nova_user_id}
host_list = self._l2network_plugin.schedule_host(instance_tenant_id,
instance_id,
instance_desc)
instance_vif_desc = {'project_id': tenant_id,
'user_id': nova_user_id,
'vif_id': vif_id}
vif_description = self._l2network_plugin.associate_port(
instance_tenant_id, instance_id,
instance_vif_desc)
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], vif_id)
self._l2network_plugin.unplug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
vif_id_unplugged = vif_id + '(detached)'
self.assertEqual(port[const.INTERFACEID], vif_id_unplugged)
self.tearDownNetworkPortInterface(
tenant_id, instance_tenant_id, instance_id, instance_vif_desc,
new_net_dict[const.NET_ID], port_dict[const.PORT_ID])
LOG.debug("test_unplug_interface - END")
def test_unplug_interface_networkDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p0005'):
"""
Tests detaachment of an interface to a port, when the network does
not exist
"""
LOG.debug("test_unplug_interface_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.unplug_interface,
tenant_id, net_id, port_id)
LOG.debug("test_unplug_interface_networkDNE - END")
def test_unplug_interface_portDNE(self, tenant_id='test_tenant',
port_id='p0005'):
"""
Tests detaachment of an interface to a port, when the port does
not exist
"""
LOG.debug("test_unplug_interface_portDNE - START")
new_net_dict = self._l2network_plugin.create_network(tenant_id,
self.network_name)
self.assertRaises(exc.PortNotFound,
self._l2network_plugin.unplug_interface, tenant_id,
new_net_dict[const.NET_ID], port_id)
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_unplug_interface_portDNE - END")
def test_create_portprofile(self, net_tenant_id=None,
net_profile_name=None, net_qos=None):
"""
Tests creation of a port-profile
"""
LOG.debug("test_create_portprofile - tenant id: %s - START",
net_tenant_id)
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
if net_profile_name:
profile_name = net_profile_name
else:
profile_name = self.profile_name
if net_qos:
qos = net_qos
else:
qos = self.qos
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, profile_name, qos)
port_profile_id = port_profile_dict['profile_id']
port_profile = cdb.get_portprofile(tenant_id, port_profile_id)
self.assertEqual(port_profile[const.PPNAME], profile_name)
self.assertEqual(port_profile[const.PPQOS], qos)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_create_portprofile - tenant id: %s - END",
net_tenant_id)
def test_delete_portprofile(self, net_tenant_id=None):
"""
Tests deletion of a port-profile
"""
LOG.debug("test_delete_portprofile - tenant id: %s - START",
net_tenant_id)
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
self._l2network_plugin.delete_portprofile(tenant_id, port_profile_id)
self.assertRaises(Exception, cdb.get_portprofile, port_profile_id)
LOG.debug("test_delete_portprofile - tenant id: %s - END",
net_tenant_id)
def test_delete_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005'):
"""
Tests deletion of a port-profile when netowrk does not exist
"""
LOG.debug("test_delete_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.delete_portprofile,
tenant_id, profile_id)
LOG.debug("test_delete_portprofileDNE - END")
def test_delete_portprofileAssociated(self, tenant_id='test_tenant'):
"""
Tests deletion of an associatedport-profile
"""
LOG.debug("test_delete_portprofileAssociated - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
new_net_dict = self._l2network_plugin.create_network(
tenant_id, 'test_network')
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
const.PORT_UP)
self._l2network_plugin.associate_portprofile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
self.assertRaises(cexc.PortProfileInvalidDelete,
self._l2network_plugin.delete_portprofile,
tenant_id, port_profile_id)
self.tearDownAssociatePortProfile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_delete_portprofileAssociated - END")
def test_list_portprofile(self, tenant_id='test_tenant'):
"""
Tests listing of port-profiles
"""
LOG.debug("test_list_portprofile - tenant id: %s - START", tenant_id)
profile_name2 = tenant_id + '_port_profile2'
qos2 = tenant_id + 'qos2'
port_profile_dict1 = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_dict2 = self._l2network_plugin.create_portprofile(
tenant_id, profile_name2, qos2)
port_profile_id1 = port_profile_dict1['profile_id']
port_profile_id2 = port_profile_dict2['profile_id']
list_all_portprofiles = self._l2network_plugin.get_all_portprofiles(
tenant_id)
port_profile_list = [port_profile_dict1, port_profile_dict2]
pplist = cdb.get_all_portprofiles()
new_pplist = []
for pp in pplist:
new_pp = self._make_portprofile_dict(tenant_id,
pp[const.UUID],
pp[const.PPNAME],
pp[const.PPQOS])
new_pplist.append(new_pp)
self.assertTrue(new_pplist[0] in port_profile_list)
self.assertTrue(new_pplist[1] in port_profile_list)
self.tearDownPortProfile(tenant_id, port_profile_id1)
self.tearDownPortProfile(tenant_id, port_profile_id2)
LOG.debug("test_list_portprofile - tenant id: %s - END", tenant_id)
def test_show_portprofile(self, net_tenant_id=None):
"""
Tests display of a port-profile
"""
LOG.debug("test_show_portprofile - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
result_port_profile = self._l2network_plugin.get_portprofile_details(
tenant_id, port_profile_id)
port_profile = cdb.get_portprofile(tenant_id, port_profile_id)
self.assertEqual(port_profile[const.PPQOS], self.qos)
self.assertEqual(port_profile[const.PPNAME], self.profile_name)
self.assertEqual(result_port_profile[const.PROFILE_QOS],
self.qos)
self.assertEqual(result_port_profile[const.PROFILE_NAME],
self.profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_show_portprofile - tenant id: %s - END", net_tenant_id)
def test_show_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005'):
"""
Tests display of a port-profile when network does not exist
"""
LOG.debug("test_show_portprofileDNE - START")
self.assertRaises(Exception,
self._l2network_plugin.get_portprofile_details,
tenant_id, profile_id)
LOG.debug("test_show_portprofileDNE - END")
def test_rename_portprofile(self, tenant_id='test_tenant',
new_profile_name='new_profile_name'):
"""
Tests rename of a port-profile
"""
LOG.debug("test_rename_portprofile - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
result_port_profile_dict = self._l2network_plugin.rename_portprofile(
tenant_id, port_profile_id, new_profile_name)
port_profile = cdb.get_portprofile(tenant_id, port_profile_id)
self.assertEqual(port_profile[const.PPNAME], new_profile_name)
self.assertEqual(result_port_profile_dict[const.PROFILE_NAME],
new_profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_show_portprofile - tenant id: %s - END")
def test_rename_portprofileDNE(self, tenant_id='test_tenant',
profile_id='pr0005',
new_profile_name='new_profile_name'):
"""
Tests rename of a port-profile when network does not exist
"""
LOG.debug("test_rename_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.rename_portprofile,
tenant_id, profile_id, new_profile_name)
LOG.debug("test_rename_portprofileDNE - END")
def test_associate_portprofile(self, tenant_id='test_tenant'):
"""
Tests association of a port-profile
"""
LOG.debug("test_associate_portprofile - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
self._l2network_plugin.associate_portprofile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
port_profile_associate = cdb.get_pp_binding(tenant_id, port_profile_id)
self.assertEqual(port_profile_associate[const.PORTID],
port_dict[const.PORT_ID])
self.tearDownAssociatePortProfile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
self.tearDownNetworkPort(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_associate_portprofile - END")
def test_associate_portprofileDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005',
profile_id='pr0005'):
"""
Tests association of a port-profile when a network does not exist
"""
LOG.debug("test_associate_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.associate_portprofile,
tenant_id, net_id, port_id, profile_id)
LOG.debug("test_associate_portprofileDNE - END")
def test_disassociate_portprofile(self, tenant_id='test_tenant',
):
"""
Tests disassociation of a port-profile
"""
LOG.debug("test_disassociate_portprofile - START")
new_net_dict = self._l2network_plugin.create_network(
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID],
self.state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile_id']
self._l2network_plugin.associate_portprofile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
self._l2network_plugin.disassociate_portprofile(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
port_profile_associate = cdb.get_pp_binding(tenant_id, port_profile_id)
self.assertEqual(port_profile_associate, [])
self.tearDownPortProfile(tenant_id, port_profile_id)
self.tearDownNetworkPort(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_disassociate_portprofile - END")
def test_disassociate_portprofileDNE(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005', profile_id='pr0005'):
"""
Tests disassociation of a port-profile when network does not exist
"""
LOG.debug("test_disassociate_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self._l2network_plugin.disassociate_portprofile,
tenant_id, net_id, port_id, profile_id)
LOG.debug("test_disassociate_portprofileDNE - END")
def test_get_vlan_name(self, net_tenant_id=None, vlan_id="NewVlan",
vlan_prefix=conf.VLAN_NAME_PREFIX):
"""
Tests get vlan name
"""
LOG.debug("test_get_vlan_name - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
result_vlan_name = self._l2network_plugin._get_vlan_name(tenant_id,
vlan_id)
expected_output = vlan_prefix + vlan_id
self.assertEqual(result_vlan_name, expected_output)
LOG.debug("test_get_vlan_name - END")
def test_validate_port_state(self, state=const.PORT_UP):
"""
Tests validate port state
"""
LOG.debug("test_validate_port_state - START")
result = self._l2network_plugin._validate_port_state(state)
self.assertEqual(result, True)
LOG.debug("test_validate_port_state - END")
def test_invalid_port_state(self, state="BADSTATE"):
"""
Tests invalidate port state
"""
LOG.debug("test_validate_port_state - START")
self.assertRaises(exc.StateInvalid,
self._l2network_plugin._validate_port_state,
state)
LOG.debug("test_validate_port_state - END")
def setUp(self):
"""
Set up function
"""
self.tenant_id = "test_tenant"
self.network_name = "test_network"
self.profile_name = "test_tenant_port_profile"
self.qos = "test_qos"
self.state = const.PORT_UP
self.net_id = '00005'
self.port_id = 'p0005'
self.remote_interface = 'new_interface'
self._l2network_plugin = l2network_plugin.L2Network()
"""
Clean up functions after the tests
"""
def tearDown(self):
"""Clear the test environment"""
# Remove database contents
db.clear_db()
def tearDownNetwork(self, tenant_id, network_dict_id):
"""
Tear down the Network
"""
self._l2network_plugin.delete_network(tenant_id, network_dict_id)
def tearDownPortOnly(self, tenant_id, network_dict_id, port_id):
"""
Tear doen the port
"""
self._l2network_plugin.delete_port(tenant_id, network_dict_id, port_id)
def tearDownNetworkPort(self, tenant_id, network_dict_id, port_id):
"""
Tear down Network Port
"""
self._l2network_plugin.delete_port(tenant_id, network_dict_id, port_id)
self.tearDownNetwork(tenant_id, network_dict_id)
def tearDownNetworkPortInterface(self, tenant_id, instance_tenant_id,
instance_id, instance_desc, network_dict_id,
port_id):
"""
Tear down Network Port Interface
"""
self._l2network_plugin.detach_port(instance_tenant_id, instance_id,
instance_desc)
self.tearDownNetworkPort(tenant_id, network_dict_id, port_id)
def tearDownPortProfile(self, tenant_id, port_profile_id):
"""
Tear down Port Profile
"""
self._l2network_plugin.delete_portprofile(tenant_id, port_profile_id)
def tearDownPortProfileBinding(self, tenant_id, port_profile_id):
"""
Tear down port profile binding
"""
self._l2network_plugin.delete_portprofile(tenant_id, port_profile_id)
def tearDownAssociatePortProfile(self, tenant_id, net_id, port_id,
port_profile_id):
"""
Tear down associate port profile
"""
self._l2network_plugin.disassociate_portprofile(
tenant_id, net_id, port_id, port_profile_id)
self.tearDownPortProfile(tenant_id, port_profile_id)
def _make_net_dict(self, net_id, net_name, ports):
res = {const.NET_ID: net_id, const.NET_NAME: net_name}
res[const.NET_PORTS] = ports
return res
def _make_port_dict(self, port_id, state, net_id, attachment):
res = {const.PORT_ID: port_id, const.PORT_STATE: state}
res[const.NET_ID] = net_id
res[const.ATTACHMENT] = attachment
return res
def _make_portprofile_dict(self, tenant_id, profile_id, profile_name,
qos):
profile_associations = self._make_portprofile_assc_list(
tenant_id, profile_id)
res = {const.PROFILE_ID: str(profile_id),
const.PROFILE_NAME: profile_name,
const.PROFILE_ASSOCIATIONS: profile_associations,
const.PROFILE_VLAN_ID: None,
const.PROFILE_QOS: qos}
return res
def _make_portprofile_assc_list(self, tenant_id, profile_id):
plist = cdb.get_pp_binding(tenant_id, profile_id)
assc_list = []
for port in plist:
assc_list.append(port[const.PORTID])
return assc_list