From 4c71489e9e940361ea2e014defb58ec78217bc01 Mon Sep 17 00:00:00 2001 From: Shweta P Date: Mon, 15 Aug 2011 23:02:22 -0700 Subject: [PATCH] Changes to test structure. Adding pylint correctons. --- quantum/common/test_lib.py | 278 +++++++++++++++++ quantum/plugins/cisco/README | 37 ++- quantum/plugins/cisco/run_tests.py | 281 +++--------------- .../cisco/tests/unit/test_l2networkApi.py | 37 +-- .../cisco/tests/unit/test_nexus_plugin.py | 7 +- .../cisco/tests/unit/test_ucs_driver.py | 55 ++-- .../cisco/tests/unit/test_ucs_plugin.py | 36 +-- run_tests.py | 246 +-------------- run_tests.sh | 15 +- tests/unit/test_api.py | 42 +-- 10 files changed, 429 insertions(+), 605 deletions(-) create mode 100644 quantum/common/test_lib.py diff --git a/quantum/common/test_lib.py b/quantum/common/test_lib.py new file mode 100644 index 000000000..2ccd681ce --- /dev/null +++ b/quantum/common/test_lib.py @@ -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", +} diff --git a/quantum/plugins/cisco/README b/quantum/plugins/cisco/README index aa8a2880e..47b63b717 100755 --- a/quantum/plugins/cisco/README +++ b/quantum/plugins/cisco/README @@ -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. + + export PLUGIN_DIR=quantum/plugins/cisco + ./run_tests.sh quantum.plugins.cisco.tests.unit. + + or + + python /run_tests.py + quantum.plugins.cisco.tests.unit. 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 diff --git a/quantum/plugins/cisco/run_tests.py b/quantum/plugins/cisco/run_tests.py index d63cc34a4..0b9d98f61 100644 --- a/quantum/plugins/cisco/run_tests.py +++ b/quantum/plugins/cisco/run_tests.py @@ -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 - - -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 +sys.path.append(os.getcwd()) - def supported(cls, stream=sys.stdout): - return True - supported = classmethod(supported) +from quantum.common.test_lib import run_tests, test_config +#from quantum.plugins.openvswitch.tests.test_vlan_map import VlanMapTest - 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 +if __name__ == '__main__': + exit_status = False - self.stream.write( - ' %s' % str(test.test._testMethodName).ljust(60)) - self.stream.flush() + # if a single test case was specified, + # we should only invoked the tests once + invoke_once = len(sys.argv) > 1 + cwd = os.getcwd() -class QuantumTestRunner(core.TextTestRunner): - def _makeResult(self): - return QuantumTestResult(self.stream, - self.descriptions, - self.verbosity, - self.config) + working_dir = os.path.abspath("tests") + c = config.Config(stream=sys.stdout, + env=os.environ, + verbosity=3, + workingDir=working_dir) + exit_status = run_tests(c) + if invoke_once: + sys.exit(0) -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) + os.chdir(cwd) - working_dir = os.path.abspath("tests") + working_dir = os.path.abspath("quantum/plugins/cisco/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 = exit_status or run_tests(c) + + sys.exit(exit_status) diff --git a/quantum/plugins/cisco/tests/unit/test_l2networkApi.py b/quantum/plugins/cisco/tests/unit/test_l2networkApi.py index 366bb4df2..7ba966fb8 100644 --- a/quantum/plugins/cisco/tests/unit/test_l2networkApi.py +++ b/quantum/plugins/cisco/tests/unit/test_l2networkApi.py @@ -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, diff --git a/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py b/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py index 013d79894..d77985b17 100644 --- a/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py +++ b/quantum/plugins/cisco/tests/unit/test_nexus_plugin.py @@ -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): diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_driver.py b/quantum/plugins/cisco/tests/unit/test_ucs_driver.py index ab93d778f..0e71a8a0a 100644 --- a/quantum/plugins/cisco/tests/unit/test_ucs_driver.py +++ b/quantum/plugins/cisco/tests/unit/test_ucs_driver.py @@ -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 = " "\ " "\ "" -create_profile_output = " "\ " " -change_vlan_output = " "\ " " -delete_vlan_output = " "\ " "\ " " -delete_profile_output = " "\ " " -associate_profile_output = " "\ " 0) LOG.debug("test_get_next_dynamic_nic - END") diff --git a/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py index 978c02648..7fabe4267 100644 --- a/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py +++ b/quantum/plugins/cisco/tests/unit/test_ucs_plugin.py @@ -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) diff --git a/run_tests.py b/run_tests.py index d63cc34a4..d73c0d518 100644 --- a/run_tests.py +++ b/run_tests.py @@ -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)) diff --git a/run_tests.sh b/run_tests.sh index 9c603c982..b9079e9ca 100755 --- a/run_tests.sh +++ b/run_tests.sh @@ -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 diff --git a/tests/unit/test_api.py b/tests/unit/test_api.py index aee8f73f3..e56a2cdd7 100644 --- a/tests/unit/test_api.py +++ b/tests/unit/test_api.py @@ -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') -- 2.45.2