x/vmware-nsx
Code Issues Proposed changes

Pulling in changes from Shweta's branch.

Browse Source
This commit is contained in:
Sumit Naiksatam
2011-08-14 09:35:19 -07:00
parent 8d36031ae2 fa70c6650e
commit 2ba61676a9
3 changed files with 206 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -57,7 +57,7 @@ class DBUtils(object):
results = cursor.fetchall()
db.commit()
LOG.debug("DB query execution succeeded: %s" % sql_query)
db.close()
db.close()
except:
db.rollback()
LOG.debug("DB query execution failed: %s" % sql_query)

15
quantum/plugins/cisco/l2network_plugin.py
View File

@@ -80,7 +80,6 @@ class L2Network(QuantumPluginBase):
new_net_dict = {const.NET_ID: new_net_id,
const.NET_NAME: net_name,
const.NET_PORTS: []}
return new_net_dict
def delete_network(self, tenant_id, net_id):
@@ -143,7 +142,7 @@ class L2Network(QuantumPluginBase):
network = db.network_rename(tenant_id, net_id, new_name)
net_dict = self._make_net_dict(network[const.UUID],
network[const.NETWORKNAME],
[])
None)
return net_dict
def get_all_ports(self, tenant_id, net_id):
@@ -257,7 +256,7 @@ class L2Network(QuantumPluginBase):
pp[const.UUID],
pp[const.PPNAME],
pp[const.PPQOS])
new_pplist.append(new_pp)
new_pplist.append(pp)
return new_pplist
@@ -283,12 +282,12 @@ class L2Network(QuantumPluginBase):
pp = cdb.get_portprofile(profile_id)
except Exception, e:
raise cexc.PortProfileNotFound(tenant_id=tenant_id,
profile_id=profile_id)
portprofile_id=profile_id)
plist = cdb.get_pp_binding(profile_id)
if plist:
raise cexc.PortProfileInvalidDelete(tenant_id=tenant_id,
portprofile_id=profile_id)
profile_id=profile_id)
else:
cdb.remove_portprofile(profile_id)
@@ -297,7 +296,7 @@ class L2Network(QuantumPluginBase):
pp = cdb.get_portprofile(profile_id)
except Exception, e:
raise cexc.PortProfileNotFound(tenant_id=tenant_id,
portprofile_id=profile_id)
profile_id=profile_id)
pp = cdb.update_portprofile(profile_id, new_name)
new_pp = self._make_portprofile_dict(tenant_id,
pp[const.UUID],
@@ -370,12 +369,12 @@ class L2Network(QuantumPluginBase):
return inspect.stack()[1 + offset][3]
def _make_net_dict(self, net_id, net_name, ports):
res = {const.NET_ID: str(net_id), const.NET_NAME: net_name}
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, port_state, net_id, attachment):
res = {const.PORT_ID: str(port_id), const.PORT_STATE: port_state}
res = {const.PORT_ID: port_id, const.PORT_STATE: port_state}
res[const.NET_ID] = net_id
res[const.ATTACHMENT] = attachment
return res

298
quantum/plugins/cisco/tests/unit/test_l2networkApi.py
View File

@@ -19,11 +19,15 @@
import logging
import unittest
import time
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.common import cisco_utils as utils
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')
@@ -47,6 +51,8 @@ class CoreAPITestFunc(unittest.TestCase):
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")
@@ -64,6 +70,9 @@ class CoreAPITestFunc(unittest.TestCase):
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(net, None)
self.assertEqual(
new_net_dict[const.NET_ID], delete_net_dict[const.NET_ID])
LOG.debug("test_delete_network - END")
@@ -117,6 +126,8 @@ class CoreAPITestFunc(unittest.TestCase):
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])
@@ -152,6 +163,8 @@ class CoreAPITestFunc(unittest.TestCase):
tenant_id, self.network_name)
rename_net_dict = self._l2network_plugin.rename_network(
tenant_id, new_net_dict[const.NET_ID], 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_rename_network - END")
@@ -184,9 +197,18 @@ class CoreAPITestFunc(unittest.TestCase):
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")
@@ -206,9 +228,20 @@ class CoreAPITestFunc(unittest.TestCase):
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])
@@ -227,7 +260,12 @@ class CoreAPITestFunc(unittest.TestCase):
tenant_id, self.network_name)
port_dict = self._l2network_plugin.create_port(
tenant_id, new_net_dict[const.NET_ID], port_state)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(port_dict[const.PORT_STATE], port_state)
self.assertEqual(port_dict[const.NET_ID], new_net_dict[const.NET_ID])
self.assertEqual(port[const.PORTSTATE], port_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")
@@ -263,8 +301,16 @@ class CoreAPITestFunc(unittest.TestCase):
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.assertEqual(port, None)
# self.assertEqual(delete_port_dict[const.PORT_STATE], port_state)
# self.assertEqual(delete_port_dict[const.NET_ID], new_net_dict[NET_ID])
# self.assertEqual(delete_port_dict[const.PORT_ID],
# new_net_dict[PORT_ID])
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
self.assertEqual(delete_port_dict, None)
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',
@@ -327,6 +373,9 @@ class CoreAPITestFunc(unittest.TestCase):
update_port_dict = self._l2network_plugin.update_port(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_state)
new_port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(new_port[const.PORTSTATE], port_state)
self.assertEqual(update_port_dict[const.PORT_STATE], port_state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
@@ -341,7 +390,7 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_update_port_networkDNE - START")
self.assertRaises(exc.NetworkNotFound,
self._l2network_plugin.update_port, tenant_id,
net_id, port_id, self.port_state)
net_id, port_id, 'ACTIVE')
LOG.debug("test_update_port_networkDNE - END")
def test_update_portDNE(self, tenant_id='test_tenant', port_id='p0005'):
@@ -371,6 +420,9 @@ class CoreAPITestFunc(unittest.TestCase):
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.port_state)
self.assertEqual(get_port_dict[const.PORT_STATE], self.port_state)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
@@ -416,10 +468,13 @@ class CoreAPITestFunc(unittest.TestCase):
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.assertEqual(
self._l2network_plugin._networks[new_net_dict[const.NET_ID]]
[const.NET_PORTS][port_dict[const.PORT_ID]]
[const.ATTACHMENT], remote_interface)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
# self.assertEqual(
# self._l2network_plugin._networks[new_net_dict[const.NET_ID]]
# [const.NET_PORTS][port_dict[const.PORT_ID]]
# [const.ATTACHMENT], remote_interface)
self.assertEqual(port[const.INTERFACEID], remote_interface)
self.tearDownNetworkPortInterface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
@@ -470,7 +525,7 @@ class CoreAPITestFunc(unittest.TestCase):
self._l2network_plugin.plug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
self.assertRaises(exc.AlreadyAttached,
self.assertRaises(exc.PortInUse,
self._l2network_plugin.plug_interface, tenant_id,
new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], remote_interface)
@@ -496,9 +551,12 @@ class CoreAPITestFunc(unittest.TestCase):
self._l2network_plugin.unplug_interface(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
self.assertEqual(self._l2network_plugin._networks
[new_net_dict[const.NET_ID]][const.NET_PORTS]
[port_dict[const.PORT_ID]][const.ATTACHMENT], None)
port = db.port_get(new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
# self.assertEqual(self._l2network_plugin._networks
# [new_net_dict[const.NET_ID]][const.NET_PORTS]
# [port_dict[const.PORT_ID]][const.ATTACHMENT], None)
self.assertEqual(port[const.INTERFACEID], None)
self.tearDownNetworkPort(tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_unplug_interface - END")
@@ -533,7 +591,7 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_unplug_interface_portDNE - END")
def test_create_portprofile(self, net_tenant_id=None,
net_profile_name=None, net_vlan_id=None):
net_profile_name=None, net_qos=None):
"""
Tests creation of a port-profile
"""
@@ -548,19 +606,22 @@ class CoreAPITestFunc(unittest.TestCase):
profile_name = net_profile_name
else:
profile_name = self.profile_name
if net_vlan_id:
vlan_id = net_vlan_id
if net_qos:
qos = net_qos
else:
vlan_id = self.vlan_id
qos = self.qos
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, profile_name, vlan_id)
tenant_id, profile_name, qos)
port_profile_id = port_profile_dict['profile-id']
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]['vlan-id'],
vlan_id)
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]
['profile-name'], profile_name)
port_profile = cdb.get_portprofile(port_profile_id)
self.assertEqual(port_profile[const.PPNAME], profile_name)
self.assertEqual(port_profile[const.PPQOS], qos)
# self.assertEqual(
# self._l2network_plugin._portprofiles[port_profile_id]['vlan-id'],
# vlan_id)
#self.assertEqual(
# self._l2network_plugin._portprofiles[port_profile_id]
# ['profile-name'], profile_name)
self.tearDownPortProfile(tenant_id, port_profile_id)
LOG.debug("test_create_portprofile - tenant id: %s - END",
net_tenant_id)
@@ -577,10 +638,13 @@ class CoreAPITestFunc(unittest.TestCase):
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
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.assertEqual(self._l2network_plugin._portprofiles, {})
# port_profile = cdb.get_portprofile(port_profile_id)
self.assertRaises(Exception, cdb.get_portprofile, port_profile_id)
# self.assertEqual(port_profile, {})
# self.assertEqual(self._l2network_plugin._portprofiles, {})
LOG.debug("test_delete_portprofile - tenant id: %s - END",
net_tenant_id)
@@ -604,7 +668,7 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_delete_portprofileAssociated - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(
tenant_id, self.net_id, self.port_id, port_profile_id)
@@ -622,23 +686,34 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_list_portprofile - tenant id: %s - START", tenant_id)
profile_name2 = tenant_id + '_port_profile2'
vlan_id2 = tenant_id + '201'
qos2 = tenant_id + 'qos2'
port_profile_dict1 = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
tenant_id, self.profile_name, self.qos)
port_profile_dict2 = self._l2network_plugin.create_portprofile(
tenant_id, profile_name2, vlan_id2)
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)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id1]['vlan-id'], self.vlan_id)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id1]['profile-name'], self.profile_name)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id2]['vlan-id'], vlan_id2)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id2]['profile-name'], profile_name2)
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.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id1]['vlan-id'], self.vlan_id)
# self.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id1]['profile-name'], self.profile_name)
# self.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id2]['vlan-id'], vlan_id2)
# self.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id2]['profile-name'], profile_name2)
LOG.debug("test_create_portprofile - tenant id: %s - END", tenant_id)
def test_show_portprofile(self, net_tenant_id=None):
@@ -652,12 +727,15 @@ class CoreAPITestFunc(unittest.TestCase):
else:
tenant_id = self.tenant_id
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
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)
self.assertEqual(result_port_profile[const.PROFILE_VLAN_ID],
self.vlan_id)
port_profile = cdb.get_portprofile(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)
@@ -670,7 +748,7 @@ class CoreAPITestFunc(unittest.TestCase):
"""
LOG.debug("test_show_portprofileDNE - START")
self.assertRaises(cexc.PortProfileNotFound,
self.assertRaises(Exception,
self._l2network_plugin.get_portprofile_details,
tenant_id, profile_id)
LOG.debug("test_show_portprofileDNE - END")
@@ -683,10 +761,12 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_rename_portprofile - START")
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
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(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)
@@ -712,16 +792,29 @@ class CoreAPITestFunc(unittest.TestCase):
"""
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.port_state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(
tenant_id, net_id, port_id, port_profile_id)
self.assertEqual(
self._l2network_plugin._portprofiles[port_profile_id]
[const.PROFILE_ASSOCIATIONS][0], port_id)
tenant_id, net_id, port_dict[const.PORT_ID],
port_profile_id)
port_profile_associate = cdb.get_pp_binding(port_profile_id)
self.assertEqual(port_profile_associate[const.PORTID],
port_dict[const.PORT_ID])
#self.assertEqual(
# self._l2network_plugin._portprofiles[port_profile_id]
# [const.PROFILE_ASSOCIATIONS][0], port_id)
self.tearDownAssociatePortProfile(tenant_id, net_id,
port_id, port_profile_id)
port_dict[const.PORT_ID],
port_profile_id)
self.tearDownNetworkPortInterface(
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',
@@ -744,16 +837,28 @@ class CoreAPITestFunc(unittest.TestCase):
"""
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.port_state)
port_profile_dict = self._l2network_plugin.create_portprofile(
tenant_id, self.profile_name, self.vlan_id)
tenant_id, self.profile_name, self.qos)
port_profile_id = port_profile_dict['profile-id']
self._l2network_plugin.associate_portprofile(tenant_id, net_id,
port_id, port_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, net_id, port_id, port_profile_id)
self.assertEqual(self._l2network_plugin._portprofiles
[port_profile_id][const.PROFILE_ASSOCIATIONS], [])
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID], port_profile_id)
port_profile_associate = cdb.get_pp_binding(port_profile_id)
self.assertEqual(port_profile_associate, [])
# self.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id][const.PROFILE_ASSOCIATIONS], [])
self.tearDownPortProfile(tenant_id, port_profile_id)
self.tearDownNetworkPortInterface(
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',
@@ -769,23 +874,7 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_disassociate_portprofileDNE - END")
# def test_disassociate_portprofile_Unassociated
def test_get_tenant(self, net_tenant_id=None):
"""
Tests get tenant
"""
LOG.debug("test_get_tenant - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
tenant_dict = self._l2network_plugin._get_tenant(tenant_id)
self.assertEqual(tenant_dict[const.TENANT_ID], tenant_id)
self.assertEqual(tenant_dict[const.TENANT_NAME], tenant_id)
LOG.debug("test_get_tenant - END")
def test_get_vlan_name(self, net_tenant_id=None, vlan_name="NewVlan",
def test_get_vlan_name(self, net_tenant_id=None, vlan_id="NewVlan",
vlan_prefix=conf.VLAN_NAME_PREFIX):
"""
Tests get vlan name
@@ -797,8 +886,8 @@ class CoreAPITestFunc(unittest.TestCase):
else:
tenant_id = self.tenant_id
result_vlan_name = self._l2network_plugin._get_vlan_name(tenant_id,
vlan_name)
expected_output = vlan_prefix + tenant_id + "-" + vlan_name
vlan_id)
expected_output = vlan_prefix + vlan_id
self.assertEqual(result_vlan_name, expected_output)
LOG.debug("test_get_vlan_name - END")
@@ -823,43 +912,23 @@ class CoreAPITestFunc(unittest.TestCase):
port_state)
LOG.debug("test_validate_port_state - END")
def test_validate_attachment(self, net_tenant_id=None,
remote_interface_id="new_interface"):
"""
Tests validate attachment
"""
LOG.debug("test_validate_attachment - START")
if net_tenant_id:
tenant_id = net_tenant_id
else:
tenant_id = self.tenant_id
net_name = self.network_name
new_network_dict = self._l2network_plugin.create_network(tenant_id,
net_name)
network_id = new_network_dict[const.NET_ID]
new_port_dict = self._l2network_plugin.create_port(tenant_id,
network_id)
port_id = new_port_dict[const.PORT_ID]
self._l2network_plugin.plug_interface(
tenant_id, new_network_dict[const.NET_ID], port_id,
remote_interface_id)
self.assertRaises(exc.AlreadyAttached,
self._l2network_plugin._validate_attachment,
tenant_id, network_id, port_id, remote_interface_id)
self.tearDownNetworkPortInterface(
tenant_id, new_network_dict[const.NET_ID], port_id)
LOG.debug("test_validate_attachment - END")
def setUp(self):
self.tenant_id = "test_tenant"
self.network_name = "test_network"
self.profile_name = "test_tenant_port_profile"
self.vlan_id = "test_tenant_vlanid300"
# self.vlan_id = "test_tenant_vlanid300"
self.qos = "test_qos"
self.port_state = const.PORT_UP
self.net_id = '00005'
self.port_id = 'p0005'
self.remote_interface = 'new_interface'
#sql_query = "drop database quantum_l2network"
#sql_query_2 = "create database quantum_l2network"
#self._utils = utils.DBUtils()
#self._utils.execute_db_query(sql_query)
#time.sleep(10)
#self._utils.execute_db_query(sql_query_2)
#time.sleep(10)
self._l2network_plugin = l2network_plugin.L2Network()
"""
@@ -890,3 +959,32 @@ class CoreAPITestFunc(unittest.TestCase):
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, port_state, net_id, attachment):
res = {const.PORT_ID: port_id, const.PORT_STATE: port_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(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, profile_id):
plist = cdb.get_pp_binding(profile_id)
assc_list = []
for port in plist:
assc_list.append(port[const.PORTID])
return assc_list