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