x/vmware-nsx
Code Issues Proposed changes

Changes to test structure. Adding pylint correctons.

Browse Source
This commit is contained in:
Shweta P 2011-08-15 23:02:22 -07:00
parent 4315a65f74
commit 4c71489e9e
10 changed files with 432 additions and 608 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

278
quantum/common/test_lib.py Normal file
View File

@ -0,0 +1,278 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 OpenStack, LLC
# 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.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import gettext
import os
import unittest
import sys
import logging
from nose import result
from nose import core
from nose import config
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
from win32console import GetStdHandle, STD_OUT_HANDLE, \
FOREGROUND_RED, FOREGROUND_BLUE, FOREGROUND_GREEN, \
FOREGROUND_INTENSITY
red, green, blue, bold = (FOREGROUND_RED, FOREGROUND_GREEN,
FOREGROUND_BLUE, FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = GetStdHandle(STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
class QuantumTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
result.TextTestResult.__init__(self, *args, **kw)
self._last_case = None
self.colorizer = None
# NOTE(vish, tfukushima): reset stdout for the terminal check
stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
def getDescription(self, test):
return str(test)
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
if self.showAll:
self.colorizer.write("OK", 'green')
self.stream.writeln()
elif self.dots:
self.stream.write('.')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
if self.showAll:
self.colorizer.write("FAIL", 'red')
self.stream.writeln()
elif self.dots:
self.stream.write('F')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for errorClasses.
If the exception is a registered class, the error will be added
to the list for that class, not errors.
"""
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# This is for compatibility with Python 2.3.
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passwd = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_details(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
if self.showAll:
self.colorizer.write("ERROR", 'red')
self.stream.writeln()
elif self.dots:
stream.write('E')
def startTest(self, test):
unittest.TestResult.startTest(self, test)
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class QuantumTestRunner(core.TextTestRunner):
def _makeResult(self):
return QuantumTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config)
def run_tests(c):
logger = logging.getLogger()
hdlr = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.DEBUG)
runner = QuantumTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c)
return not core.run(config=c, testRunner=runner)
# describes parameters used by different unit/functional tests
# a plugin-specific testing mechanism should import this dictionary
# and override the values in it if needed (e.g., run_tests.py in
# quantum/plugins/openvswitch/ )
test_config = {
"plugin_name": "quantum.plugins.SamplePlugin.FakePlugin",
}

37
quantum/plugins/cisco/README
View File

@ -2,7 +2,7 @@
README: Quantum L2 Network Plugin Framework
============================================
:Author: Sumit Naiksatam, Ram Durairaj, Mark Voelker, Edgar Magana, Shweta Padubidri, Rohit Agarwalla, Ying Liu, Debo Dutta
:Author: Sumit Naiksatam, Ram Durairaj, Mark Voelker, Edgar Magana, Shweta Padubidri, Rohit Agarwalla, Ying Liu
:Contact: netstack@lists.launchpad.net
:Web site: https://launchpad.net/~cisco-openstack
:Copyright: 2011 Cisco Systems, Inc.
@ -125,8 +125,6 @@ nexus_ip_address=10.0.0.1
# Port number on the Nexus switch to which the UCSM 6120 is connected
# Use shortened interface syntax, e.g. "3/23" not "Ethernet3/23".
nexus_port=3/23
#Port number where the SSH will be running at Nexus Switch, e.g.: 22 (Default)
nexus_ssh_port=22
[DRIVER]
name=quantum.plugins.cisco.nexus.cisco_nexus_network_driver.CiscoNEXUSDriver
@ -168,7 +166,8 @@ password=mySecretPasswordForNexus
How to test the installation
----------------------------
The unit tests are located at quantum/plugins/cisco/tests/unit. They can be
executed from quantum/plugins/cisco/ using the run_tests.py script.
executed from the main folder using the run_tests.sh or to get a more detailed
result the quantum/plugins/cisco/run_tests.py script.
1. Testing the core API (without UCS/Nexus/RHEL hardware, and can be run on
Ubuntu):
@ -176,18 +175,40 @@ executed from quantum/plugins/cisco/ using the run_tests.py script.
quantum/plugins/cisco/conf/plugins.ini
Then run the test script:
python run_tests.py unit.test_l2networkApi
Set the environment variable PLUGIN_DIR to the location of the plugin
directory. This is manadatory if the run_tests.sh script is used.
export PLUGIN_DIR=quantum/plugins/cisco
./run_tests.sh quantum.plugins.cisco.tests.unit.test_l2networkApi
or
python quantum/plugins/cisco/run_tests.py
quantum.plugins.cisco.tests.unit.test_l2networkApi
2. Specific Plugin unit test (needs environment setup as indicated in the
pre-requisites):
python run_tests.py unit.<name_of_the file>
export PLUGIN_DIR=quantum/plugins/cisco
./run_tests.sh quantum.plugins.cisco.tests.unit.<name_of_the file>
or
python <path to the plugin directory>/run_tests.py
quantum.plugins.cisco.tests.unit.<name_of_the file>
E.g.:
python run_tests.py unit.test_ucs_plugin.py
python quantum/plugins/cisco/run_tests.py
quantum.plugins.cisco.tests.unit.test_ucs_plugin.py
3. All unit tests (needs environment setup as indicated in the pre-requisites):
python run_tests.py unit
export PLUGIN_DIR=quantum/plugins/cisco
./run_tests.sh quantum.plugins.cisco.tests.unit
or
python quantum/plugins/cisco/run_tests.py quantum.plugins.cisco.tests.unit
Additional installation required on Nova Compute

287
quantum/plugins/cisco/run_tests.py
View File

@ -16,50 +16,34 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Unittest runner for quantum
"""Unittest runner for quantum OVS plugin
This file should be run from the top dir in the quantum directory
To run all test::
python run_tests.py
python quantum/plugins/cisco/run_tests.py
To run all unit tests::
python run_tests.py unit
python quantum/plugins/cisco/run_tests.py quantum.plugins.cisco.tests.unit
To run all functional tests::
python run_tests.py functional
python quantum/plugins/openvswitch/run_tests.py functional
To run a single unit test::
python run_tests.py unit.test_stores:TestSwiftBackend.test_get
python quantum/plugins/openvswitch/run_tests.py \
quantum.plugins.cisco.tests.unit.test_stores:TestSwiftBackend.test_get
To run a single functional test::
python run_tests.py functional.test_service:TestController.test_create
python quantum/plugins/openvswitch/run_tests.py \
quantum.plugins.cisco.tests.functional.test_service \
:TestController.test_create
To run a single unit test module::
python run_tests.py unit.test_stores
python quantum/plugins/openvswitch/run_tests.py unit.test_stores
To run a single functional test module::
python run_tests.py functional.test_stores
python quantum/plugins/openvswitch/run_tests.py functional.test_stores
"""
import gettext
@ -69,233 +53,38 @@ import unittest
import sys
from nose import config
from nose import result
from nose import core
sys.path.append(os.getcwd())
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
from win32console import GetStdHandle, STD_OUT_HANDLE, \
FOREGROUND_RED, FOREGROUND_BLUE, FOREGROUND_GREEN, \
FOREGROUND_INTENSITY
red, green, blue, bold = (FOREGROUND_RED, FOREGROUND_GREEN,
FOREGROUND_BLUE, FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = GetStdHandle(STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
class QuantumTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
result.TextTestResult.__init__(self, *args, **kw)
self._last_case = None
self.colorizer = None
# NOTE(vish, tfukushima): reset stdout for the terminal check
stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
def getDescription(self, test):
return str(test)
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
if self.showAll:
self.colorizer.write("OK", 'green')
self.stream.writeln()
elif self.dots:
self.stream.write('.')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
if self.showAll:
self.colorizer.write("FAIL", 'red')
self.stream.writeln()
elif self.dots:
self.stream.write('F')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for errorClasses.
If the exception is a registered class, the error will be added
to the list for that class, not errors.
"""
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# This is for compatibility with Python 2.3.
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passwd = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_details(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
if self.showAll:
self.colorizer.write("ERROR", 'red')
self.stream.writeln()
elif self.dots:
stream.write('E')
def startTest(self, test):
unittest.TestResult.startTest(self, test)
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class QuantumTestRunner(core.TextTestRunner):
def _makeResult(self):
return QuantumTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config)
from quantum.common.test_lib import run_tests, test_config
#from quantum.plugins.openvswitch.tests.test_vlan_map import VlanMapTest
if __name__ == '__main__':
# Set up test logger.
logger = logging.getLogger()
hdlr = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.DEBUG)
exit_status = False
# if a single test case was specified,
# we should only invoked the tests once
invoke_once = len(sys.argv) > 1
cwd = os.getcwd()
working_dir = os.path.abspath("tests")
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=working_dir)
runner = QuantumTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c)
sys.exit(not core.run(config=c, testRunner=runner))
exit_status = run_tests(c)
if invoke_once:
sys.exit(0)
os.chdir(cwd)
working_dir = os.path.abspath("quantum/plugins/cisco/tests")
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=working_dir)
exit_status = exit_status or run_tests(c)
sys.exit(exit_status)

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

@ -19,13 +19,13 @@
import logging
import unittest
import time
#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.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
@ -675,13 +675,16 @@ class CoreAPITestFunc(unittest.TestCase):
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)
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])
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'):
@ -793,8 +796,7 @@ class CoreAPITestFunc(unittest.TestCase):
tenant_id, profile_id, new_profile_name)
LOG.debug("test_rename_portprofileDNE - END")
def test_associate_portprofile(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005'):
def test_associate_portprofile(self, tenant_id='test_tenant'):
"""
Tests association of a port-profile
"""
@ -809,18 +811,18 @@ class CoreAPITestFunc(unittest.TestCase):
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_dict[const.PORT_ID],
port_profile_id)
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.assertEqual(
# self._l2network_plugin._portprofiles[port_profile_id]
# [const.PROFILE_ASSOCIATIONS][0], port_id)
self.tearDownAssociatePortProfile(tenant_id, net_id,
port_dict[const.PORT_ID],
port_profile_id)
self.tearDownNetworkPortInterface(
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")
@ -839,7 +841,7 @@ class CoreAPITestFunc(unittest.TestCase):
LOG.debug("test_associate_portprofileDNE - END")
def test_disassociate_portprofile(self, tenant_id='test_tenant',
net_id='0005', port_id='p00005'):
):
"""
Tests disassociation of a port-profile
"""
@ -864,7 +866,7 @@ class CoreAPITestFunc(unittest.TestCase):
# self.assertEqual(self._l2network_plugin._portprofiles
# [port_profile_id][const.PROFILE_ASSOCIATIONS], [])
self.tearDownPortProfile(tenant_id, port_profile_id)
self.tearDownNetworkPortInterface(
self.tearDownNetworkPort(
tenant_id, new_net_dict[const.NET_ID],
port_dict[const.PORT_ID])
LOG.debug("test_disassociate_portprofile - END")
@ -988,7 +990,8 @@ class CoreAPITestFunc(unittest.TestCase):
def _make_portprofile_dict(self, tenant_id, profile_id, profile_name,
qos):
profile_associations = self._make_portprofile_assc_list(tenant_id, profile_id)
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,

7
quantum/plugins/cisco/tests/unit/test_nexus_plugin.py
View File

@ -259,11 +259,10 @@ class TestNexusPlugin(unittest.TestCase):
self.tearDownNetwork(tenant_id, new_net_dict[const.NET_ID])
LOG.debug("test_get_vlan_id_for_network - END")
"""
Clean up functions after the tests
"""
def tearDownNetwork(self, tenant_id, network_dict_id):
"""
Clean up functions after the tests
"""
self._cisco_nexus_plugin.delete_network(tenant_id, network_dict_id)
# def test_create_network(self):

55
quantum/plugins/cisco/tests/unit/test_ucs_driver.py
View File

@ -24,13 +24,13 @@ from quantum.plugins.cisco.ucs import cisco_ucs_network_driver
LOG = logging.getLogger('quantum.tests.test_ucs_driver')
create_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
CREATE_VLAN_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/net-New Vlan\"> "\
"<fabricVlan defaultNet=\"no\" dn=\"fabric/lan/net-New Vlan\" id=\"200\" "\
"name=\"New Vlan\" status=\"created\"></fabricVlan> </pair> </inConfigs> "\
"</configConfMos>"
create_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
CREATE_PROFILE_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/profiles/vnic-"\
"New Profile\"> <vnicProfile descr=\"Profile created by Cisco OpenStack "\
"Quantum Plugin\" dn=\"fabric/lan/profiles/vnic-New Profile\" maxPorts="\
@ -39,7 +39,7 @@ create_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"name=\"New Vlan\" rn=\"if-New Vlan\" > </vnicEtherIf> </vnicProfile> "\
"</pair> </inConfigs> </configConfMos>"
change_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
CHANGE_VLAN_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\""\
"fabric/lan/profiles/vnic-New Profile\"> <vnicProfile descr=\"Profile "\
"created by Cisco OpenStack Quantum Plugin\" "\
@ -50,18 +50,18 @@ change_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
"<vnicEtherIf defaultNet=\"yes\" name=\"New Vlan\" rn=\"if-New Vlan\" > "\
"</vnicEtherIf> </vnicProfile> </pair></inConfigs> </configConfMos>"
delete_vlan_output = "<configConfMos cookie=\"cookie_placeholder\" "\
DELETE_VLAN_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs><pair key=\"fabric/lan/net-New Vlan\"> "\
"<fabricVlan dn=\"fabric/lan/net-New Vlan\" status=\"deleted\"> "\
"</fabricVlan> </pair> </inConfigs></configConfMos>"
delete_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
DELETE_PROFILE_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"false\"> <inConfigs><pair key=\""\
"fabric/lan/profiles/vnic-New Profile\"> <vnicProfile "\
"dn=\"fabric/lan/profiles/vnic-New Profile\" status=\"deleted\"> "\
"</vnicProfile></pair> </inConfigs> </configConfMos>"
associate_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
ASSOCIATE_PROFILE_OUTPUT = "<configConfMos cookie=\"cookie_placeholder\" "\
"inHierarchical=\"true\"> <inConfigs> <pair key="\
"\"fabric/lan/profiles/vnic-New Profile/cl-New Profile Client\">"\
" <vmVnicProfCl dcName=\".*\" descr=\"\" dn=\"fabric/lan/profiles/vnic-"\
@ -73,83 +73,86 @@ associate_profile_output = "<configConfMos cookie=\"cookie_placeholder\" "\
class TestUCSDriver(unittest.TestCase):
def setUp(self):
self._ucsmDriver = cisco_ucs_network_driver.CiscoUCSMDriver()
self.ucsm_driver = cisco_ucs_network_driver.CiscoUCSMDriver()
self.vlan_name = 'New Vlan'
self.vlan_id = '200'
self.profile_name = 'New Profile'
self.old_vlan_name = 'Old Vlan'
self.profile_client_name = 'New Profile Client'
def test_create_vlan_post_data(self, expected_output=create_vlan_output):
def test_create_vlan_post_data(self, expected_output=CREATE_VLAN_OUTPUT):
"""
Tests creation of vlan post Data
"""
LOG.debug("test_create_vlan")
vlan_details = self._ucsmDriver._create_vlan_post_data(
vlan_details = self.ucsm_driver._create_vlan_post_data(
self.vlan_name, self.vlan_id)
self.assertEqual(vlan_details, expected_output)
LOG.debug("test_create_vlan - END")
def test_create_profile_post_data(
self, expected_output=create_profile_output):
self, expected_output=CREATE_PROFILE_OUTPUT):
"""
Tests creation of profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._create_profile_post_data(
profile_details = self.ucsm_driver._create_profile_post_data(
self.profile_name, self.vlan_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_change_vlan_in_profile_post_data(
self, expected_output=change_vlan_output):
def test_change_vlan_profile_data(
self, expected_output=CHANGE_VLAN_OUTPUT):
"""
Tests creation of change vlan in profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._change_vlan_in_profile_post_data(
profile_details = self.ucsm_driver._change_vlan_in_profile_post_data(
self.profile_name, self.old_vlan_name, self.vlan_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_delete_vlan_post_data(self, expected_output=delete_vlan_output):
LOG.debug("test_create_profile_post_data - START")
def test_delete_vlan_post_data(self, expected_output=DELETE_VLAN_OUTPUT):
"""
Tests deletion of vlan post Data
"""
vlan_details = self._ucsmDriver._create_vlan_post_data(
LOG.debug("test_create_profile_post_data - START")
self.ucsm_driver._create_vlan_post_data(
self.vlan_name, self.vlan_id)
vlan_delete_details = self._ucsmDriver._delete_vlan_post_data(
vlan_delete_details = self.ucsm_driver._delete_vlan_post_data(
self.vlan_name)
self.assertEqual(vlan_delete_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_delete_profile_post_data(
self, expected_output=delete_profile_output):
self, expected_output=DELETE_PROFILE_OUTPUT):
"""
Tests deletion of profile post Data
"""
LOG.debug("test_create_profile_post_data - START")
profile_details = self._ucsmDriver._create_profile_post_data(
#profile_details = self.ucsm_driver._create_profile_post_data(
# self.profile_name, self.vlan_name)
self.ucsm_driver._create_profile_post_data(
self.profile_name, self.vlan_name)
profile_delete_details = self._ucsmDriver._delete_profile_post_data(
profile_delete_details = self.ucsm_driver._delete_profile_post_data(
self.profile_name)
self.assertEqual(profile_delete_details, expected_output)
LOG.debug("test_create_profile_post - END")
def test_create_profile_client_post_data(
self, expected_output=associate_profile_output):
def test_create_profile_client_data(
self, expected_output=ASSOCIATE_PROFILE_OUTPUT):
"""
Tests creation of profile client post Data
"""
LOG.debug("test_create_profile_client_post_data - START")
profile_details = self._ucsmDriver._create_profile_client_post_data(
LOG.debug("test_create_profile_client_data - START")
profile_details = self.ucsm_driver._create_profile_client_post_data(
self.profile_name, self.profile_client_name)
self.assertEqual(profile_details, expected_output)
LOG.debug("test_create_profile_post - END")
@ -160,6 +163,6 @@ class TestUCSDriver(unittest.TestCase):
"""
LOG.debug("test_get_next_dynamic_nic - START")
dynamic_nic_id = self._ucsmDriver._get_next_dynamic_nic()
dynamic_nic_id = self.ucsm_driver._get_next_dynamic_nic()
self.assertTrue(len(dynamic_nic_id) > 0)
LOG.debug("test_get_next_dynamic_nic - END")

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

@ -32,7 +32,7 @@ class UCSVICTestPlugin(unittest.TestCase):
self.tenant_id = "test_tenant_cisco12"
self.net_name = "test_network_cisco12"
self.net_id = 000007
self.net_id = 000011
self.vlan_name = "q-" + str(self.net_id) + "vlan"
self.vlan_id = 266
self.port_id = "4"
@ -238,12 +238,12 @@ class UCSVICTestPlugin(unittest.TestCase):
self.assertEqual(new_port_profile[const.PROFILE_NAME], profile_name)
self.tearDownNetworkPort(self.tenant_id, self.net_id, self.port_id)
def _test_get_port_details_state_down(self, port_state):
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_get_port_details_state_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,
@ -268,8 +268,8 @@ class UCSVICTestPlugin(unittest.TestCase):
def test_get_port_details_state_up(self):
self._test_get_port_details_state_up(const.PORT_UP)
def test_get_port_details_state_down(self):
self._test_get_port_details_state_down(const.PORT_DOWN)
def test_show_port_state_down(self):
self._test_show_port_state_down(const.PORT_DOWN)
def test_create_port_profile(self):
LOG.debug("UCSVICTestPlugin:test_create_port_profile() called\n")
@ -313,7 +313,7 @@ class UCSVICTestPlugin(unittest.TestCase):
self.tenant_id, self.net_id, self.port_id)
self.assertEqual(port[const.ATTACHMENT], remote_interface_id)
port_profile = port[const.PORT_PROFILE]
profile_name = port_profile[const.PROFILE_NAME]
#profile_name = port_profile[const.PROFILE_NAME]
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(
@ -346,7 +346,7 @@ class UCSVICTestPlugin(unittest.TestCase):
self.tenant_id, self.net_id, self.port_id)
self.assertEqual(port[const.ATTACHMENT], None)
port_profile = port[const.PORT_PROFILE]
profile_name = port_profile[const.PROFILE_NAME]
#profile_name = port_profile[const.PROFILE_NAME]
self.assertEqual(port_profile[const.PROFILE_VLAN_NAME],
conf.DEFAULT_VLAN_NAME)
self.assertEqual(port_profile[const.PROFILE_VLAN_ID],
@ -394,12 +394,12 @@ class UCSVICTestPlugin(unittest.TestCase):
def test_get_network_NetworkNotFound(self):
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin._get_network,
*(self.tenant_id, self.net_id))
self.tenant_id, self.net_id)
def test_delete_network_NetworkNotFound(self):
self.assertRaises(exc.NetworkNotFound,
self._cisco_ucs_plugin.delete_network,
*(self.tenant_id, self.net_id))
self.tenant_id, self.net_id)
def test_delete_port_PortInUse(self):
self._test_delete_port_PortInUse("4")
@ -414,7 +414,7 @@ class UCSVICTestPlugin(unittest.TestCase):
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.tenant_id, self.net_id, self.port_id)
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
@ -423,7 +423,7 @@ class UCSVICTestPlugin(unittest.TestCase):
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.tenant_id, self.net_id, self.port_id)
self.tearDownNetwork(self.tenant_id, self.net_id)
def test_plug_interface_PortInUse(self):
@ -441,16 +441,16 @@ class UCSVICTestPlugin(unittest.TestCase):
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.tenant_id, self.net_id, self.port_id,
remote_interface_id2)
self.tearDownNetworkPortInterface(self.tenant_id, self.net_id,
self.port_id)
def test_validate_attachment_AlreadyAttached(self):
def test_attachment_exists(self):
LOG.debug("UCSVICTestPlugin:testValidateAttachmentAlreadyAttached")
self._test_validate_attachment_AlreadyAttached("4")
self._test_attachment_exists("4")
def _test_validate_attachment_AlreadyAttached(self, remote_interface_id):
def _test_attachment_exists(self, remote_interface_id):
LOG.debug("UCSVICTestPlugin:_test_validate_attachmentAlreadyAttached")
self._cisco_ucs_plugin.create_network(self.tenant_id, self.net_name,
self.net_id, self.vlan_name,
@ -461,8 +461,8 @@ class UCSVICTestPlugin(unittest.TestCase):
self.port_id,
remote_interface_id)
self.assertRaises(
exc.AlreadyAttached, self._cisco_ucs_plugin._validate_attachment,
*(self.tenant_id, self.net_id, self.port_id, remote_interface_id))
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)

246
run_tests.py
View File

@ -16,27 +16,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Unittest runner for quantum
@ -63,239 +42,18 @@ To run a single functional test module::
"""
import gettext
import logging
import os
import unittest
import sys
from quantum.common.test_lib import run_tests
from nose import config
from nose import result
from nose import core
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
from win32console import GetStdHandle, STD_OUT_HANDLE, \
FOREGROUND_RED, FOREGROUND_BLUE, FOREGROUND_GREEN, \
FOREGROUND_INTENSITY
red, green, blue, bold = (FOREGROUND_RED, FOREGROUND_GREEN,
FOREGROUND_BLUE, FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = GetStdHandle(STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
class QuantumTestResult(result.TextTestResult):
def __init__(self, *args, **kw):
result.TextTestResult.__init__(self, *args, **kw)
self._last_case = None
self.colorizer = None
# NOTE(vish, tfukushima): reset stdout for the terminal check
stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
def getDescription(self, test):
return str(test)
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
if self.showAll:
self.colorizer.write("OK", 'green')
self.stream.writeln()
elif self.dots:
self.stream.write('.')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
if self.showAll:
self.colorizer.write("FAIL", 'red')
self.stream.writeln()
elif self.dots:
self.stream.write('F')
self.stream.flush()
# NOTE(vish, tfukushima): copied from unittest with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for errorClasses.
If the exception is a registered class, the error will be added
to the list for that class, not errors.
"""
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# This is for compatibility with Python 2.3.
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passwd = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_details(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
if self.showAll:
self.colorizer.write("ERROR", 'red')
self.stream.writeln()
elif self.dots:
stream.write('E')
def startTest(self, test):
unittest.TestResult.startTest(self, test)
current_case = test.test.__class__.__name__
if self.showAll:
if current_case != self._last_case:
self.stream.writeln(current_case)
self._last_case = current_case
self.stream.write(
' %s' % str(test.test._testMethodName).ljust(60))
self.stream.flush()
class QuantumTestRunner(core.TextTestRunner):
def _makeResult(self):
return QuantumTestResult(self.stream,
self.descriptions,
self.verbosity,
self.config)
if __name__ == '__main__':
# Set up test logger.
logger = logging.getLogger()
hdlr = logging.StreamHandler()
formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)
logger.setLevel(logging.DEBUG)
working_dir = os.path.abspath("tests")
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=working_dir)
runner = QuantumTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c)
sys.exit(not core.run(config=c, testRunner=runner))
sys.exit(run_tests(c))

15
run_tests.sh
View File

@ -39,11 +39,20 @@ done
function run_tests {
# Just run the test suites in current environment
${wrapper} rm -f tests.sqlite
${wrapper} $NOSETESTS 2> run_tests.err.log
${wrapper} rm -f ./$PLUGIN_DIR/tests.sqlite
${wrapper} $NOSETESTS 2> ./$PLUGIN_DIR/run_tests.err.log
}
NOSETESTS="python run_tests.py $noseargs"
NOSETESTS="python ./$PLUGIN_DIR/run_tests.py $noseargs"
if [ -n "$PLUGIN_DIR" ]
then
if ! [ -f ./$PLUGIN_DIR/run_tests.py ]
then
echo "Could not find run_tests.py in plugin directory $PLUGIN_DIR"
exit 1
fi
fi
if [ $never_venv -eq 0 ]
then

42
tests/unit/test_api.py
View File

@ -20,13 +20,13 @@
import logging
import unittest
import tests.unit.testlib_api as testlib
from quantum import api as server
from quantum.db import api as db
from quantum.common.test_lib import test_config
from quantum.common.wsgi import Serializer
LOG = logging.getLogger('quantum.tests.test_api')
@ -150,19 +150,6 @@ class APITest(unittest.TestCase):
network_data['network'])
LOG.debug("_test_rename_network - format:%s - END", format)
def _test_rename_network_duplicate(self, format):
LOG.debug("_test_rename_network_duplicate - format:%s - START", format)
content_type = "application/%s" % format
network_id1 = self._create_network(format, name="net1")
network_id2 = self._create_network(format, name="net2")
update_network_req = testlib.update_network_request(self.tenant_id,
network_id2,
"net1",
format)
update_network_res = update_network_req.get_response(self.api)
self.assertEqual(update_network_res.status_int, 422)
LOG.debug("_test_rename_network_duplicate - format:%s - END", format)
def _test_rename_network_badrequest(self, format):
LOG.debug("_test_rename_network_badrequest - format:%s - START",
format)
@ -440,23 +427,6 @@ class APITest(unittest.TestCase):
show_port_res.body, content_type)
self.assertEqual({'id': port_id, 'state': new_port_state},
port_data['port'])
# now set it back to the original value
update_port_req = testlib.update_port_request(self.tenant_id,
network_id, port_id,
port_state,
format)
update_port_res = update_port_req.get_response(self.api)
self.assertEqual(update_port_res.status_int, 200)
show_port_req = testlib.show_port_request(self.tenant_id,
network_id, port_id,
format)
show_port_res = show_port_req.get_response(self.api)
self.assertEqual(show_port_res.status_int, 200)
port_data = self._port_serializer.deserialize(
show_port_res.body, content_type)
self.assertEqual({'id': port_id, 'state': port_state},
port_data['port'])
LOG.debug("_test_set_port_state - format:%s - END", format)
def _test_set_port_state_networknotfound(self, format):
@ -680,7 +650,7 @@ class APITest(unittest.TestCase):
def setUp(self):
options = {}
options['plugin_provider'] = 'quantum.plugins.SamplePlugin.FakePlugin'
options['plugin_provider'] = test_config['plugin_name']
self.api = server.APIRouterV01(options)
self.tenant_id = "test_tenant"
self.network_name = "test_network"
@ -736,12 +706,6 @@ class APITest(unittest.TestCase):
def test_rename_network_xml(self):
self._test_rename_network('xml')
def test_rename_network_duplicate_json(self):
self._test_rename_network_duplicate('json')
def test_rename_network_duplicate_xml(self):
self._test_rename_network_duplicate('xml')
def test_rename_network_badrequest_json(self):
self._test_rename_network_badrequest('json')