--- /dev/null
+# 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",
+}
import logging as LOG
import os
import sys
-import unittest
from quantum.quantum_plugin_base import QuantumPluginBase
from optparse import OptionParser
def get_interface_details(self, tenant_id, net_id, port_id):
res = db.port_get(port_id)
return res.interface_id
-
-
-class VlanMapTest(unittest.TestCase):
-
- def setUp(self):
- self.vmap = VlanMap()
-
- def tearDown(self):
- pass
-
- def testAddVlan(self):
- vlan_id = self.vmap.acquire("foobar")
- self.assertTrue(vlan_id == 2)
-
- def testReleaseVlan(self):
- vlan_id = self.vmap.acquire("foobar")
- self.vmap.release("foobar")
- self.assertTrue(self.vmap.get(vlan_id) == None)
-
-
-# TODO(bgh): Make the tests use a sqlite database instead of mysql
-class OVSPluginTest(unittest.TestCase):
-
- def setUp(self):
- self.quantum = OVSQuantumPlugin()
- self.tenant_id = "testtenant"
-
- def testCreateNetwork(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- self.assertTrue(net1["net-name"] == "plugin_test1")
-
- def testGetNetworks(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- net2 = self.quantum.create_network(self.tenant_id, "plugin_test2")
- nets = self.quantum.get_all_networks(self.tenant_id)
- count = 0
- for x in nets:
- if "plugin_test" in x["net-name"]:
- count += 1
- self.assertTrue(count == 2)
-
- def testDeleteNetwork(self):
- net = self.quantum.create_network(self.tenant_id, "plugin_test1")
- self.quantum.delete_network(self.tenant_id, net["net-id"])
- nets = self.quantum.get_all_networks(self.tenant_id)
- count = 0
- for x in nets:
- if "plugin_test" in x["net-name"]:
- count += 1
- self.assertTrue(count == 0)
-
- def testRenameNetwork(self):
- net = self.quantum.create_network(self.tenant_id, "plugin_test1")
- net = self.quantum.rename_network(self.tenant_id, net["net-id"],
- "plugin_test_renamed")
- self.assertTrue(net["net-name"] == "plugin_test_renamed")
-
- def testCreatePort(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- port = self.quantum.create_port(self.tenant_id, net1["net-id"])
- ports = self.quantum.get_all_ports(self.tenant_id, net1["net-id"])
- count = 0
- for p in ports:
- count += 1
- self.assertTrue(count == 1)
-
- def testDeletePort(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- port = self.quantum.create_port(self.tenant_id, net1["net-id"])
- ports = self.quantum.get_all_ports(self.tenant_id, net1["net-id"])
- count = 0
- for p in ports:
- count += 1
- self.assertTrue(count == 1)
- for p in ports:
- self.quantum.delete_port(self.tenant_id, id, p["port-id"])
- ports = self.quantum.get_all_ports(self.tenant_id, net1["net-id"])
- count = 0
- for p in ports:
- count += 1
- self.assertTrue(count == 0)
-
- def testGetPorts(self):
- pass
-
- def testPlugInterface(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- port = self.quantum.create_port(self.tenant_id, net1["net-id"])
- self.quantum.plug_interface(self.tenant_id, net1["net-id"],
- port["port-id"], "vif1.1")
- port = self.quantum.get_port_details(self.tenant_id, net1["net-id"],
- port["port-id"])
- self.assertTrue(port["attachment"] == "vif1.1")
-
- def testUnPlugInterface(self):
- net1 = self.quantum.create_network(self.tenant_id, "plugin_test1")
- port = self.quantum.create_port(self.tenant_id, net1["net-id"])
- self.quantum.plug_interface(self.tenant_id, net1["net-id"],
- port["port-id"], "vif1.1")
- port = self.quantum.get_port_details(self.tenant_id, net1["net-id"],
- port["port-id"])
- self.assertTrue(port["attachment"] == "vif1.1")
- self.quantum.unplug_interface(self.tenant_id, net1["net-id"],
- port["port-id"])
- port = self.quantum.get_port_details(self.tenant_id, net1["net-id"],
- port["port-id"])
- self.assertTrue(port["attachment"] == "")
-
- def tearDown(self):
- networks = self.quantum.get_all_networks(self.tenant_id)
- # Clean up any test networks lying around
- for net in networks:
- id = net["net-id"]
- name = net["net-name"]
- if "plugin_test" in name:
- # Clean up any test ports lying around
- ports = self.quantum.get_all_ports(self.tenant_id, id)
- for p in ports:
- self.quantum.delete_port(self.tenant_id, id, p["port-id"])
- self.quantum.delete_network(self.tenant_id, id)
-
-
-if __name__ == "__main__":
- usagestr = "Usage: %prog [OPTIONS] <command> [args]"
- parser = OptionParser(usage=usagestr)
- parser.add_option("-v", "--verbose", dest="verbose",
- action="store_true", default=False, help="turn on verbose logging")
-
- options, args = parser.parse_args()
-
- if options.verbose:
- LOG.basicConfig(level=LOG.DEBUG)
- else:
- LOG.basicConfig(level=LOG.WARN)
-
- # Make sqlalchemy quieter
- LOG.getLogger('sqlalchemy.engine').setLevel(LOG.WARN)
- # Run the tests
- suite = unittest.TestLoader().loadTestsFromTestCase(OVSPluginTest)
- unittest.TextTestRunner(verbosity=2).run(suite)
- suite = unittest.TestLoader().loadTestsFromTestCase(VlanMapTest)
- unittest.TextTestRunner(verbosity=2).run(suite)
--- /dev/null
+#!/usr/bin/env python
+# 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.
+
+
+"""Unittest runner for quantum OVS plugin
+
+This file should be run from the top dir in the quantum directory
+
+To run all test::
+ python quantum/plugins/openvswitch/run_tests.py
+
+To run all unit tests::
+ python quantum/plugins/openvswitch/run_tests.py unit
+
+To run all functional tests::
+ python quantum/plugins/openvswitch/run_tests.py functional
+
+To run a single unit test::
+ python quantum/plugins/openvswitch/run_tests.py \
+ unit.test_stores:TestSwiftBackend.test_get
+
+To run a single functional test::
+ python quantum/plugins/openvswitch/run_tests.py \
+ functional.test_service:TestController.test_create
+
+To run a single unit test module::
+ python quantum/plugins/openvswitch/run_tests.py unit.test_stores
+
+To run a single functional test module::
+ python quantum/plugins/openvswitch/run_tests.py functional.test_stores
+"""
+
+import gettext
+import logging
+import os
+import unittest
+import sys
+
+from nose import config
+
+sys.path.append(os.getcwd())
+
+from quantum.common.test_lib import run_tests, test_config
+from quantum.plugins.openvswitch.tests.test_vlan_map import VlanMapTest
+
+if __name__ == '__main__':
+ exit_status = False
+
+ cwd = os.getcwd()
+
+ 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)
+
+ os.chdir(cwd)
+
+ working_dir = os.path.abspath("quantum/plugins/openvswitch/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)
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2011 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.
+
+# See http://code.google.com/p/python-nose/issues/detail?id=373
+# The code below enables nosetests to work with i18n _() blocks
+import __builtin__
+import unittest
+setattr(__builtin__, '_', lambda x: x)
+
+
+class BaseTest(unittest.TestCase):
+
+ def setUp(self):
+ pass
+
+
+def setUp():
+ pass
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+# Copyright 2011 Nicira Networks, Inc.
+# All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from quantum.plugins.openvswitch.ovs_quantum_plugin import VlanMap
+
+
+class VlanMapTest(unittest.TestCase):
+
+ def setUp(self):
+ self.vmap = VlanMap()
+
+ def tearDown(self):
+ pass
+
+ def testAddVlan(self):
+ vlan_id = self.vmap.acquire("foobar")
+ self.assertTrue(vlan_id == 2)
+
+ def testReleaseVlan(self):
+ vlan_id = self.vmap.acquire("foobar")
+ self.vmap.release("foobar")
+ self.assertTrue(self.vmap.get(vlan_id) == None)
# 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
"""
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))
#!/bin/bash
function usage {
- echo "Usage: $0 [OPTION]..."
+ echo "Usage: $0 [OPTION]"
echo "Run Quantum's test suite(s)"
echo ""
echo " -V, --virtual-env Always use virtualenv. Install automatically if not present"
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
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')
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"
Code Review / openstack-build / neutron-build.git / commitdiff