# Copyright (c) 2011 Rackspace # 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. """Creates TestCases from a list of TestEntries. This module mainly exists to translate Proboscis classes and concepts into the unittest equivalents. """ import os import pydoc import types import unittest import sys from collections import deque from functools import wraps from proboscis import compatability from proboscis import dependencies from proboscis import SkipTest from proboscis.sorting import TestGraph from proboscis.core import TestMethodClassEntry from proboscis.decorators import DEFAULT_REGISTRY # This is here so Proboscis own test harness can change it while still calling # TestProgram normally. Its how the examples are tested. OVERRIDE_DEFAULT_STREAM = None class TestPlan(object): """Grabs information from the TestRegistry and creates a test plan.""" def __init__(self, groups, test_entries, factories): test_cases = self.create_cases(test_entries, factories) graph = TestGraph(groups, test_entries, test_cases) self.tests = graph.sort() @staticmethod def create_from_registry(registry): """Returns a sorted TestPlan from a TestRegistry instance.""" return TestPlan(registry.groups, registry.tests, registry.factories) @staticmethod def create_cases_from_instance(factory, instance): if isinstance(instance, type): raise RuntimeError("Factory %s returned type %s (rather than an " "instance), which is not allowed." % (factory, instance)) if isinstance(instance, types.MethodType): home = compatability.get_method_function(instance) elif isinstance(instance, types.FunctionType): home = instance else: home = type(instance) if issubclass(home, unittest.TestCase): raise RuntimeError("Factory %s returned a unittest.TestCase " "instance %s, which is not legal.") try: entry = home._proboscis_entry_ except AttributeError: raise RuntimeError("Factory method %s returned an instance %s " "which was not tagged as a Proboscis TestEntry." % (factory, instance)) entry.mark_as_used_by_factory() # Don't iterate this since a # function is creating it. if entry.is_child: raise RuntimeError("Function %s, which exists as a bound method " "in a decorated class may not be returned from a factory." % instance) # There is potentially an issue in that a different Registry might # register an entry, and we could then read that in with a factory. # Later the entry would not be found in the dictionary of entries. if isinstance(instance, types.MethodType): try: state = TestMethodState(instance.im_self) except AttributeError: raise RuntimeError("Only bound methods may be returned from " "factories. %s is not bound." % instance) else: state = TestMethodState(entry, instance) return TestPlan._create_test_cases_for_entry(entry, state) @staticmethod def create_cases(test_entries, factories): tests = [] entries = {} for factory in factories: list = factory() for item in list: cases = TestPlan.create_cases_from_instance(factory, item) tests += cases for entry in test_entries: if not entry.is_child and not entry.used_by_factory: test_cases = TestPlan._create_test_cases_for_entry(entry) entries[entry] = test_cases tests += test_cases return tests @staticmethod def _create_test_cases_for_entry(entry, state=None): """Processes a test case entry.""" if not hasattr(entry, 'children'): # function or unittest.TestCase return [TestCase(entry)] state = state or TestMethodState(entry) cases = [] for child_entry in entry.children: case = TestCase(child_entry, state=state) cases.append(case) return cases def create_test_suite(self, config, loader): """Transforms the plan into a Nose test suite.""" creator = TestSuiteCreator(loader) if dependencies.use_nose: from nose.suite import ContextSuiteFactory suite = ContextSuiteFactory(config)([]) else: suite = unittest.TestSuite() for case in self.tests: if case.entry.info.enabled and case.entry.home is not None: tests = creator.loadTestsFromTestEntry(case) for test in tests: suite.addTest(test) return suite def filter(self, group_names=None, classes=None, functions=None): """Whittles down test list to those matching criteria.""" test_homes = [] classes = classes or [] functions = functions or [] for cls in classes: test_homes.append(cls) for function in functions: test_homes.append(function) group_names = group_names or [] filtered_list = [] while self.tests: case = self.tests.pop() if case.entry.contains(group_names, test_homes): filtered_list.append(case) # Add any groups this depends on so they will run as well. for group_name in case.entry.info.depends_on_groups: if not group_name in group_names: group_names.append(group_name) for test_home in case.entry.info.depends_on: if not test_home in test_homes: test_homes.append(test_home) self.tests = list(reversed(filtered_list)) class TestCase(object): """Represents an instance of a TestEntry. This class is also used to store status information, such as the dependent TestEntry objects (discovered when this test is sorted) and any failure in the dependencies of this test (used to raise SkipTest if needed). There may be multiple TestCase instances for each TestEntry instance. """ def __init__(self, entry, state=None): self.entry = entry self.dependents = [] # This is populated when we sort the tests. self.dependency_failure = None self.state = state def check_dependencies(self, test_self): """If a dependency has failed, SkipTest is raised.""" if self.dependency_failure is not None and \ self.dependency_failure != self and not self.entry.info.always_run: home = self.dependency_failure.entry.home dependencies.skip_test(test_self, "Failure in %s" % home) def fail_test(self, dependency_failure=None): """Called when this entry fails to notify dependents.""" if not dependency_failure: dependency_failure = self if not self.dependency_failure: # Do NOT overwrite the first cause self.dependency_failure = dependency_failure for dependent in self.dependents: if dependent.critical: dependent.case.fail_test( dependency_failure=dependency_failure) def write_doc(self, file): file.write(str(self.entry.home) + "\n") doc = pydoc.getdoc(self.entry.home) if doc: file.write(doc + "\n") for field in str(self.entry.info).split(', '): file.write("\t" + field + "\n") def __repr__(self): return "TestCase(" + repr(self.entry.home) + ", " + \ repr(self.entry.info) + ", " + object.__repr__(self) + ")" def __str__(self): return "Home = " + str(self.entry.home) + ", Info(" + \ str(self.entry.info) + ")" class TestResultListener(): """Implements methods of TestResult to be informed of test failures.""" def __init__(self, chain_to_cls): self.chain_to_cls = chain_to_cls def addError(self, test, err): self.onError(test) self.chain_to_cls.addError(self, test, err) def addFailure(self, test, err): self.onError(test) self.chain_to_cls.addFailure(self, test, err) def onError(self, test): """Notify a test entry and its dependents of failure.""" if dependencies.use_nose: root = test.test else: root = test if hasattr(root, "__proboscis_case__"): case = root.__proboscis_case__ case.fail_test() class TestResult(TestResultListener, dependencies.TextTestResult): """Adds Proboscis skip on dependency failure functionality. Extends either Nose or unittest's TextTestResult class. If a program needs to use its own TestResult class it must inherit from this class and call "onError" at the start of both the addError and addFailure functions, passing the "test" parameter, to keep Proboscis's skip on depdendency failure functionality. """ # I had issues extending TextTestResult directly so resorted to this. def __init__(self, *args, **kwargs): TestResultListener.__init__(self, dependencies.TextTestResult) dependencies.TextTestResult.__init__(self, *args, **kwargs) def test_runner_cls(wrapped_cls, cls_name): """Creates a test runner class which uses Proboscis TestResult.""" new_dict = wrapped_cls.__dict__.copy() if dependencies.use_nose: def cb_make_result(self): return TestResult(self.stream, self.descriptions, self.verbosity, self.config) else: def cb_make_result(self): return TestResult(self.stream, self.descriptions, self.verbosity) new_dict["_makeResult"] = cb_make_result return type(cls_name, (wrapped_cls,), new_dict) def skippable_func(test_case, func): """Gives free functions a Nose independent way of skipping a test. The unittest module TestCase class has a skipTest method, but to run it you need access to the TestCase class. This wraps the runTest method of the underlying unittest.TestCase subclass to invoke the skipTest method if it catches the SkipTest exception. """ s_func = None if dependencies.use_nose: s_func = func else: @wraps(func) def skip_capture_func(): st = compatability.capture_exception(func, SkipTest) if st is not None: dependencies.skip_test(test_case, st.message) s_func = skip_capture_func @wraps(s_func) def testng_method_mistake_capture_func(): compatability.capture_type_error(s_func) return testng_method_mistake_capture_func class FunctionTest(unittest.FunctionTestCase): """Wraps a single function as a test runnable by unittest / nose.""" def __init__(self, test_case): func = test_case.entry.home _old_setup = None if hasattr(func, 'setup'): # Don't destroy nose-style setup _old_setup = func.setup def cb_check(cb_self=None): test_case.check_dependencies(self) if _old_setup is not None: _old_setup() self.__proboscis_case__ = test_case sfunc = skippable_func(self, func) unittest.FunctionTestCase.__init__(self, testFunc=sfunc, setUp=cb_check) class TestMethodState(object): """Manages a test class instance used by one or more test methods.""" def __init__(self, entry, instance=None): self.entry = entry # This would be a simple "isinstance" but due to the reloading mania # needed for Proboscis's documentation tests it has to be a bit # weirder. if not str(type(self.entry)) == str(TestMethodClassEntry): raise RuntimeError("%s is not a TestMethodClassEntry but is a %s." % (self.entry, type(self.entry))) self.instance = instance def get_state(self): if not self.instance: self.instance = self.entry.home() return self.instance class MethodTest(unittest.FunctionTestCase): """Wraps a method as a test runnable by unittest.""" def __init__(self, test_case): assert test_case.state is not None #TODO: Figure out how to attach calls to BeforeMethod and BeforeClass, # AfterMethod and AfterClass. It should be easy enough to # just find them using the TestEntry parent off test_case Entrty. def cb_check(cb_self=None): test_case.check_dependencies(self) @wraps(test_case.entry.home) def func(self=None): # Called by FunctionTestCase func = test_case.entry.home func(test_case.state.get_state()) self.__proboscis_case__ = test_case sfunc = skippable_func(self, func) unittest.FunctionTestCase.__init__(self, testFunc=sfunc, setUp=cb_check) def decorate_class(setUp_method=None, tearDown_method=None): """Inserts method calls in the setUp / tearDown methods of a class.""" def return_method(cls): """Returns decorated class.""" new_dict = cls.__dict__.copy() if setUp_method: if hasattr(cls, "setUp"): @wraps(setUp_method) def _setUp(self): setUp_method(self) cls.setUp(self) else: @wraps(setUp_method) def _setUp(self): setUp_method(self) new_dict["setUp"] = _setUp if tearDown_method: if hasattr(cls, "tearDown"): @wraps(tearDown_method) def _tearDown(self): tearDown_method(self) cls.setUp(self) else: @wraps(tearDown_method) def _tearDown(self): tearDown_method(self) new_dict["tearDown"] = _tearDown return type(cls.__name__, (cls,), new_dict) return return_method class TestSuiteCreator(object): """Turns Proboscis test cases into elements to be run by unittest.""" def __init__(self, loader): self.loader = loader def loadTestsFromTestEntry(self, test_case): """Wraps a test class in magic so it will skip on dependency failures. Decorates the testEntry class's setUp method to raise SkipTest if tests this test was dependent on failed or had errors. """ home = test_case.entry.home if home is None: return [] if isinstance(home, type): return self.wrap_unittest_test_case_class(test_case) if isinstance(home, types.FunctionType): if home._proboscis_entry_.is_child: return self.wrap_method(test_case) else: return self.wrap_function(test_case) raise RuntimeError("Unknown test type:" + str(type(home))) def wrap_function(self, test_case): return [FunctionTest(test_case)] def wrap_method(self, test_case): return [MethodTest(test_case)] def wrap_unittest_test_case_class(self, test_case): original_cls = test_case.entry.home def cb_check(cb_self): test_case.check_dependencies(cb_self) testCaseClass = decorate_class(setUp_method=cb_check)(original_cls) testCaseNames = self.loader.getTestCaseNames(testCaseClass) if not testCaseNames and hasattr(testCaseClass, 'runTest'): testCaseNames = ['runTest'] suite = [] if issubclass(original_cls, unittest.TestCase): for name in testCaseNames: test_instance = testCaseClass(name) setattr(test_instance, "__proboscis_case__", test_case) suite.append(test_instance) return suite class TestProgram(dependencies.TestProgram): """Use this to run Proboscis. Translates the Proboscis test registry into types used by Nose or unittest in order to run the program. Most arguments to this are simply passed to Nose or unittest's TestProgram class. For most cases using the default arguments works fine. :param registry: The test registry to use. If unset uses the default global registry. :param groups: A list of strings representing the groups of tests to run. The list is added to by parsing the argv argument. If unset then all groups are run. :param testLoader: The test loader. By default, its unittest.TestLoader. :param config: The config passed to Nose or unittest.TestProgram. The config determines things such as plugins or output streams, so it may be necessary to create this for advanced use cases. :param plugins: Nose plugins. Similar to config it may be necessary to set this in an advanced setup. :param env: By default is os.environ. This is used only by Nose. :param testRunner: By default Proboscis uses its own. If this is set however care must be taken to avoid breaking Proboscis's automatic skipping of tests on dependency failures. In particular, _makeResult must return a subclass of proboscis.TestResult which calls proboscis.TestResult.onError at the start of the addFailure and addError methods. :param stream: By default this is standard out. :param argv: By default this is sys.argv. Proboscis parses this for the --group argument. """ def __init__(self, registry=DEFAULT_REGISTRY, groups=None, testLoader=None, config=None, plugins=None, env=None, testRunner=None, stream=None, argv=None, *args, **kwargs): groups = groups or [] argv = argv or sys.argv argv = self.extract_groups_from_argv(argv, groups) if "suite" in kwargs: raise ValueError("'suite' is not a valid argument, as Proboscis " \ "creates the suite.") self.__loader = testLoader or unittest.TestLoader() if OVERRIDE_DEFAULT_STREAM: stream = OVERRIDE_DEFAULT_STREAM if env is None: env = os.environ if dependencies.use_nose and config is None: config = self.makeConfig(env, plugins) if not stream: stream = config.stream stream = stream or sys.stdout if testRunner is None: runner_cls = test_runner_cls(dependencies.TextTestRunner, "ProboscisTestRunner") if dependencies.use_nose: testRunner = runner_cls(stream, verbosity=3, # config.verbosity, config=config) else: testRunner = runner_cls(stream, verbosity=3) #registry.sort() self.plan = TestPlan.create_from_registry(registry) if len(groups) > 0: self.plan.filter(group_names=groups) self.cases = self.plan.tests if "--show-plan" in argv: self.__run = self.show_plan else: self.__suite = self.create_test_suite_from_entries(config, self.cases) def run(): if dependencies.use_nose: dependencies.TestProgram.__init__( self, suite=self.__suite, config=config, env=env, plugins=plugins, testLoader=testLoader, # Pass arg, not what we create testRunner=testRunner, argv=argv, *args, **kwargs ) else: dependencies.TestProgram.__init__( self, suite=self.__suite, config=config, testLoader=testLoader, # Pass arg, not what we create testRunner=testRunner, argv=argv, *args, **kwargs ) self.__run = run def create_test_suite_from_entries(self, config, cases): """Creates a suite runnable by unittest.""" return self.plan.create_test_suite(config, self.__loader) def extract_groups_from_argv(self, argv, groups): """Given argv, records the "--group" options. :param argv: A list of arguments, such as sys.argv. :param groups: A list of strings for each group to run which is added to. Returns a copy of param argv with the --group options removed. This is useful if argv needs to be passed to another program such as Nose. """ new_argv = [argv[0]] for arg in argv[1:]: if arg[:8] == "--group=": groups.append(arg[8:]) else: new_argv.append(arg) return new_argv def run_and_exit(self): """Calls unittest or Nose to run all tests. unittest will call sys.exit on completion. """ self.__run() def show_plan(self): """Prints information on test entries and the order they will run.""" print(" * * * Test Plan * * *") for case in self.cases: case.write_doc(sys.stdout) @property def test_suite(self): return self.__suite