python-proboscis 1.2.6.0-1 / usr / share / pyshared / proboscis / case.py

# 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