From 8f5f4537af7790be386974628c804a7bc719b738 Mon Sep 17 00:00:00 2001 From: "Bala.FA" Date: Fri, 7 Mar 2014 18:28:09 +0530 Subject: Initial commit Change-Id: Ie8fdd046d111a4a46abe0e162985e833323bfd7d Signed-off-by: Bala.FA --- tests/Makefile.am | 51 ++++++++ tests/README | 11 ++ tests/run_tests.sh.in | 9 ++ tests/run_tests_local.sh.in | 6 + tests/testValidation.py | 141 ++++++++++++++++++++ tests/testrunner.py | 308 ++++++++++++++++++++++++++++++++++++++++++++ tests/utilsTests.py | 78 +++++++++++ 7 files changed, 604 insertions(+) create mode 100644 tests/Makefile.am create mode 100644 tests/README create mode 100644 tests/run_tests.sh.in create mode 100644 tests/run_tests_local.sh.in create mode 100644 tests/testValidation.py create mode 100644 tests/testrunner.py create mode 100644 tests/utilsTests.py (limited to 'tests') diff --git a/tests/Makefile.am b/tests/Makefile.am new file mode 100644 index 0000000..a4fc465 --- /dev/null +++ b/tests/Makefile.am @@ -0,0 +1,51 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +test_modules = \ + utilsTests.py \ + $(NULL) + +dist_glusternagioscommontests_DATA = \ + $(NULL) + +dist_glusternagioscommontests_PYTHON = \ + $(test_modules) \ + testrunner.py \ + testValidation.py \ + $(NULL) + +dist_glusternagioscommontests_SCRIPTS = \ + run_tests.sh \ + $(NULL) + +dist_noinst_DATA = \ + run_tests_local.sh \ + $(NULL) + +CLEANFILES = \ + $(NULL) + +all-local: \ + $(nodist_glusternagioscommontests_PYTHON) + +check-local: + @echo '*** Running tests. To skip this step place NOSE_EXCLUDE=.* ***' + @echo '*** into your environment. Do not submit untested code! ***' + $(top_srcdir)/tests/run_tests_local.sh $(test_modules) diff --git a/tests/README b/tests/README new file mode 100644 index 0000000..3681917 --- /dev/null +++ b/tests/README @@ -0,0 +1,11 @@ +This test suite is built on the Nose framework. For more information see: + + http://readthedocs.org/docs/nose/en/latest/ + +Running the test suite: +----------------------- + +Running these tests is easy. + +~$ python testrunner.py +~$ python testrunner.py *.py diff --git a/tests/run_tests.sh.in b/tests/run_tests.sh.in new file mode 100644 index 0000000..ff91cdb --- /dev/null +++ b/tests/run_tests.sh.in @@ -0,0 +1,9 @@ +#!/bin/sh +if [ -z "$PYTHON_EXE" ]; then + PYTHON_EXE="@PYTHON@" +fi + +prefix="@prefix@" +exec_prefix="@exec_prefix@" +pyexecdir="@pyexecdir@" +LC_ALL=C PYTHONPATH="@glusternagioscommonpylibdir@:" "$PYTHON_EXE" @top_srcdir@/tests/testrunner.py $@ diff --git a/tests/run_tests_local.sh.in b/tests/run_tests_local.sh.in new file mode 100644 index 0000000..39fc36a --- /dev/null +++ b/tests/run_tests_local.sh.in @@ -0,0 +1,6 @@ +#!/bin/sh +if [ -z "$PYTHON_EXE" ]; then + PYTHON_EXE="@PYTHON@" +fi + +PYTHONDONTWRITEBYTECODE=1 LC_ALL=C PYTHONPATH="@top_srcdir@:$PYTHONPATH" "$PYTHON_EXE" @top_srcdir@/tests/testrunner.py $@ diff --git a/tests/testValidation.py b/tests/testValidation.py new file mode 100644 index 0000000..b46c9f8 --- /dev/null +++ b/tests/testValidation.py @@ -0,0 +1,141 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +## This framework is mostly copied from vdsm test framework + +import os +from nose.plugins.skip import SkipTest +from functools import wraps +from nose.plugins import Plugin +import subprocess + + +class SlowTestsPlugin(Plugin): + """Skips tests that might be too slow to be run for quick iteration + builds""" + name = 'slowtests' + enabled = False + + def add_options(self, parser, env=os.environ): + env_opt = 'NOSE_SKIP_SLOW_TESTS' + if env is None: + default = False + else: + default = env.get(env_opt) + + parser.add_option('--without-slow-tests', + action='store_true', + default=default, + dest='disable_slow_tests', + help='Some tests might take a long time to run, ' + + 'use this to skip slow tests automatically.' + + ' [%s]' % env_opt) + + def configure(self, options, conf): + Plugin.configure(self, options, conf) + if options.disable_slow_tests: + SlowTestsPlugin.enabled = True + + +class StressTestsPlugin(Plugin): + """ + Denotes a test which stresses the resources of the system under test. Such + tests should probably not be run in parallel. This plugin provides a + mechanism for parallel testing applications to skip stress tests. + """ + name = 'nonparalleltests' + enabled = False + + def add_options(self, parser, env=os.environ): + env_opt = 'NOSE_SKIP_STRESS_TESTS' + if env is None: + default = False + else: + default = env.get(env_opt) + + parser.add_option('--without-stress-tests', + action='store_true', + default=default, + dest='disable_stress_tests', + help='Some tests stress the resources of the ' + + 'system under test. Use this option to skip' + + 'these tests (eg. when doing parallel' + + 'testing [%s]' % env_opt) + + def configure(self, options, conf): + Plugin.configure(self, options, conf) + if options.disable_stress_tests: + StressTestsPlugin.enabled = True + + +def ValidateRunningAsRoot(f): + @wraps(f) + def wrapper(*args, **kwargs): + if os.geteuid() != 0: + raise SkipTest("This test must be run as root") + + return f(*args, **kwargs) + + return wrapper + + +def slowtest(f): + @wraps(f) + def wrapper(*args, **kwargs): + if SlowTestsPlugin.enabled: + raise SkipTest("Slow tests have been disabled") + + return f(*args, **kwargs) + + return wrapper + + +def brokentest(msg="Test failed but it is known to be broken"): + def wrap(f): + @wraps(f) + def wrapper(*args, **kwargs): + try: + return f(*args, **kwargs) + except: + raise SkipTest(msg) + return wrapper + + return wrap + + +def stresstest(f): + @wraps(f) + def wrapper(*args, **kwargs): + if StressTestsPlugin.enabled: + raise SkipTest("Stress tests have been disabled") + + return f(*args, **kwargs) + + return wrapper + + +def checkSudo(cmd): + p = subprocess.Popen(['sudo', '-l', '-n'] + cmd, + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + out, err = p.communicate() + + if p.returncode != 0: + raise SkipTest("Test requires SUDO configuration (%s)" % err.strip()) diff --git a/tests/testrunner.py b/tests/testrunner.py new file mode 100644 index 0000000..56f1416 --- /dev/null +++ b/tests/testrunner.py @@ -0,0 +1,308 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +## This framework is mostly copied from vdsm test framework + +import logging +import sys +import os +import unittest +import re +import shutil +import tempfile +from contextlib import contextmanager +from glusternagios import utils +# Monkey patch pthreading in necessary +if sys.version_info[0] == 2: + # as long as we work with Python 2, we need to monkey-patch threading + # module before it is ever used. + import pthreading + pthreading.monkey_patch() +from nose import config +from nose import core +from nose import result + +from testValidation import SlowTestsPlugin, StressTestsPlugin + + +class TermColor(object): + black = 30 + red = 31 + green = 32 + yellow = 33 + blue = 34 + magenta = 35 + cyan = 36 + white = 37 + + +def colorWrite(stream, text, color): + if os.isatty(stream.fileno()) or os.environ.get("NOSE_COLOR", False): + stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text)) + else: + stream.write(text) + + +@contextmanager +def temporaryPath(perms=None, data=None): + fd, src = tempfile.mkstemp() + if data is not None: + f = os.fdopen(fd, "wb") + f.write(data) + f.flush() + f.close() + else: + os.close(fd) + if perms is not None: + os.chmod(src, perms) + try: + yield src + finally: + os.unlink(src) + + +@contextmanager +def namedTemporaryDir(): + tmpDir = tempfile.mkdtemp() + try: + yield tmpDir + finally: + shutil.rmtree(tmpDir) + + +# FIXME: This is a forward port of the assertRaises from python +# 2.7, remove when no longer supporting earlier versions +class _AssertRaisesContext(object): + """A context manager used to implement TestCase.assertRaises* methods.""" + + def __init__(self, expected, test_case, expected_regexp=None): + self.expected = expected + self.failureException = test_case.failureException + self.expected_regexp = expected_regexp + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc_value, tb): + if exc_type is None: + try: + exc_name = self.expected.__name__ + except AttributeError: + exc_name = str(self.expected) + raise self.failureException( + "{0} not raised".format(exc_name)) + if not issubclass(exc_type, self.expected): + # let unexpected exceptions pass through + return False + self.exception = exc_value # store for later retrieval + if self.expected_regexp is None: + return True + + expected_regexp = self.expected_regexp + if isinstance(expected_regexp, basestring): + expected_regexp = re.compile(expected_regexp) + if not expected_regexp.search(str(exc_value)): + raise self.failureException('"%s" does not match "%s"' % + (expected_regexp.pattern, + str(exc_value))) + return True + + +# FIXME: This is a forward port of the assertIn from python +# 2.7, remove when no longer supporting earlier versions +def safe_repr(obj, short=False): + _MAX_LENGTH = 80 + try: + result = repr(obj) + except Exception: + result = object.__repr__(obj) + if not short or len(result) < _MAX_LENGTH: + return result + return result[:_MAX_LENGTH] + ' [truncated]...' + + +class GlusterNagiosTestCase(unittest.TestCase): + def __init__(self, *args, **kwargs): + unittest.TestCase.__init__(self, *args, **kwargs) + self.log = logging.getLogger(self.__class__.__name__) + + def retryAssert(self, *args, **kwargs): + '''Keep retrying an assertion if AssertionError is raised. + See function utils.retry for the meaning of the arguments. + ''' + return utils.retry(expectedException=AssertionError, *args, **kwargs) + + # FIXME: This is a forward port of the assertRaises from python + # 2.7, remove when no longer supporting earlier versions + def assertRaises(self, excClass, callableObj=None, *args, **kwargs): + context = _AssertRaisesContext(excClass, self) + if callableObj is None: + return context + with context: + callableObj(*args, **kwargs) + + # FIXME: This is a forward port of the assertIn from python + # 2.7, remove when no longer supporting earlier versions + def assertIn(self, member, container, msg=None): + """ + Just like self.assertTrue(a in b), but with a nicer default message. + """ + if member not in container: + if msg is None: + msg = '%s not found in %s' % (safe_repr(member), + safe_repr(container)) + raise self.failureException(msg) + + # FIXME: This is a forward port of the assertNotIn from python + # 2.7, remove when no longer supporting earlier versions + def assertNotIn(self, member, container, msg=None): + """ + Just like self.assertTrue(a not in b), but with a nicer default message + """ + if member in container: + if msg is None: + msg = '%s unexpectedly found in %s' % (safe_repr(member), + safe_repr(container)) + raise self.failureException(msg) + + +class GlusterNagiosTestResult(result.TextTestResult): + def __init__(self, *args, **kwargs): + result.TextTestResult.__init__(self, *args, **kwargs) + self._last_case = None + + def getDescription(self, test): + return str(test) + + def _writeResult(self, test, long_result, color, short_result, success): + if self.showAll: + colorWrite(self.stream, long_result, color) + self.stream.writeln() + elif self.dots: + self.stream.write(short_result) + self.stream.flush() + + def addSuccess(self, test): + unittest.TestResult.addSuccess(self, test) + self._writeResult(test, 'OK', TermColor.green, '.', True) + + def addFailure(self, test, err): + unittest.TestResult.addFailure(self, test, err) + self._writeResult(test, 'FAIL', TermColor.red, 'F', False) + + def addSkip(self, test, reason): + # 2.7 skip compat + from nose.plugins.skip import SkipTest + if SkipTest in self.errorClasses: + storage, label, isfail = self.errorClasses[SkipTest] + storage.append((test, reason)) + self._writeResult(test, 'SKIP : %s' % reason, TermColor.blue, 'S', + True) + + def addError(self, test, err): + stream = getattr(self, 'stream', None) + ec, ev, tb = err + try: + exc_info = self._exc_info_to_string(err, test) + except TypeError: + # 2.3 compat + 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.passed = 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_detail(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: + self._writeResult(test, 'ERROR', TermColor.red, 'E', False) + + 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 GlusterNagiosTestRunner(core.TextTestRunner): + def __init__(self, *args, **kwargs): + core.TextTestRunner.__init__(self, *args, **kwargs) + + def _makeResult(self): + return GlusterNagiosTestResult(self.stream, + self.descriptions, + self.verbosity, + self.config) + + def run(self, test): + result_ = core.TextTestRunner.run(self, test) + return result_ + + +def run(): + argv = sys.argv + stream = sys.stdout + verbosity = 3 + testdir = os.path.dirname(os.path.abspath(__file__)) + + conf = config.Config(stream=stream, + env=os.environ, + verbosity=verbosity, + workingDir=testdir, + plugins=core.DefaultPluginManager()) + conf.plugins.addPlugin(SlowTestsPlugin()) + conf.plugins.addPlugin(StressTestsPlugin()) + + runner = GlusterNagiosTestRunner(stream=conf.stream, + verbosity=conf.verbosity, + config=conf) + + sys.exit(not core.run(config=conf, testRunner=runner, argv=argv)) + + +def findRemove(listR, value): + """used to test if a value exist, if it is, return true and remove it.""" + try: + listR.remove(value) + return True + except ValueError: + return False + + +if __name__ == '__main__': + run() diff --git a/tests/utilsTests.py b/tests/utilsTests.py new file mode 100644 index 0000000..a608dd9 --- /dev/null +++ b/tests/utilsTests.py @@ -0,0 +1,78 @@ +# +# Copyright 2014 Red Hat, Inc. +# +# This program is free software; you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation; either version 2 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA +# +# Refer to the README and COPYING files for full details of the license +# + +import errno + +from testrunner import GlusterNagiosTestCase as TestCaseBase +from glusternagios import utils + + +class RetryTests(TestCaseBase): + def testStopCallback(self): + counter = [0] + limit = 4 + + def stopCallback(): + counter[0] += 1 + if counter[0] == limit: + return True + + return False + + def foo(): + raise RuntimeError("If at first you don't succeed, try, try again." + "Then quit. There's no point in being a damn" + "fool about it.") + # W. C. Fields + + self.assertRaises(RuntimeError, utils.retry, foo, tries=(limit + 10), + sleep=0, stopCallback=stopCallback) + # Make sure we had the proper amount of iterations before failing + self.assertEquals(counter[0], limit) + + +class CommandPathTests(TestCaseBase): + def testExisting(self): + cp = utils.CommandPath('sh', 'utter nonsense', '/bin/sh') + self.assertEquals(cp.cmd, '/bin/sh') + + def testMissing(self): + NAME = 'nonsense' + try: + utils.CommandPath(NAME, 'utter nonsense').cmd + except OSError as e: + self.assertEquals(e.errno, errno.ENOENT) + self.assertIn(NAME, e.strerror) + + +class ExecCmdTests(TestCaseBase): + def testSuccess(self): + (rc, out, err) = utils.execCmd(["true"]) + self.assertEquals(rc, 0) + + def testFailure(self): + (rc, out, err) = utils.execCmd(["false"]) + self.assertEquals(rc, 1) + + def testOSError(self): + def _runUnknown(): + (rc, out, err) = utils.execCmd(["unknown"]) + + self.assertRaises(OSError, _runUnknown) -- cgit