nose_testresource.py

"""
This plugin replaces the built-in FinalizingSuiteWrapper with
testresources.OptimisingTestSuite.
"""
from nose.plugins.base import Plugin
from nose.suite import FinalizingSuiteWrapper
from nose.case import Test
from nose.plugins.manager import DefaultPluginManager
from testresources import OptimisingTestSuite
import unittest2

import sys
err = sys.stderr

class NoseTestResource(Plugin):
    """
    Replaces the built-in FinalizingSuiteWrapper with
    testresources.OptimisingTestSuite.
    """
    name = "nose-testresources"
    enableOpt = 'testresources'

    def options(self, parser, env):
        """Register commandline options."""
        parser.add_option('--testresources',
                          action='store_true',
                          dest=self.enableOpt,
                          default=False,
                          help="Enable nose-testresources")

    def prepareTestLoader(self, loader):
        """Install OptimisingTestSuite suite class in TestLoader."""
        # Disable context awareness
        loader.suiteClass = NoseOptimisingFactory(self.conf)


class NoseOptimisingFactory:
    """Factory for producing configured test suites."""

    def __init__(self, conf):
        self.conf = conf

    def __call__(self, tests=(), **kw):
        plugins = DefaultPluginManager()
        return NoseOptimisingTestSuite(tests, plugins.finalize, conf=self.conf)


class NoseOptimisingTestSuite(OptimisingTestSuite, FinalizingSuiteWrapper):

    def __init__(self, tests, finalize, conf=None):
        #err.write("initializing NoseOptimisingTestSuite\n")
        self.conf = conf
        # Exec lazy suites: makes discovery depth-first
        if callable(tests):
            tests = tests()
        self.suite = tests
        self.finalize = finalize
        OptimisingTestSuite.__init__(self, tests)

    def addTest(self, test_case_or_suite):
        try:
            tests = iter(test_case_or_suite)
        except TypeError:
            # add bare test cases directly
            self._tests.append(test_case_or_suite)
            return
        if test_case_or_suite.__class__ in (unittest2.TestSuite,
                                            self.__class__):
            # remove test cases from TestSuite or NoseOptimisingTestSuite
            # and add them recursively
            for test in tests:
                self.addTest(test)
        else:
            for test in tests:
                # do one layer of un-suite-ing, then add
                self._tests.append(test)

    def run(self, result):
        try:
            self.sortTests()
            current_resources = set()
            for test in self._tests:
                if result.shouldStop:
                    break
                resources = getattr(test, 'resources', [])
                #err.write("desired resources for next test: %s\n" % resources)
                new_resources = set()
                for name, resource in resources:
                    new_resources.update(resource.neededResources())
                self.switch(current_resources, new_resources, result)
                test(result)
                current_resources = new_resources
            self.switch(current_resources, set(), result)
            return result
        finally:
            self.finalize(result)

    def sort_tests(self):
        #err.write("before sorting: %s\n" % self._tests)
        super(NoseOptimisingTestSuite, self).sort_tests()
        #err.write("after sorting: %s\n" % self._tests)

    def switch(self, old_resource_set, new_resource_set, result):
        """Switch from 'old_resource_set' to 'new_resource_set'.

        Tear down resources in old_resource_set that aren't in
        new_resource_set and set up resources that are in new_resource_set but
        not in old_resource_set.

        :param result: TestResult object to report activity on.
        """
        #err.write("incoming old resources: %s\n" % old_resource_set)
        #err.write("incoming new resources: %s\n" % new_resource_set)
        new_resources = new_resource_set - old_resource_set
        #err.write("resources to be created: %s\n" % new_resources)
        old_resources = old_resource_set - new_resource_set
        #err.write("resources to be killed: %s\n" % old_resources)
        for resource in old_resources:
            resource.finishedWith(resource._currentResource, result)
        for resource in new_resources:
            resource.getResource(result)

setup.py

import sys
try:
    import ez_setup
    ez_setup.use_setuptools()
except ImportError:
    pass

from setuptools import setup

setup(
    name='nose-testresources',
    version='0.01',
    author='Leah Klearman',
    author_email='lklrmn@gmail.com',
    description='Replaces the built-in FinalizingSuiteWrapper with '
                'testresources.OptimisingTestSuite.',
    py_modules=['nose_testresources'],
    entry_points = {
        'nose.plugins.0.10': [
            'nose-testresources = nose_testresources:NoseTestResource'
        ]
    },
    license='Mozilla Public License 2.0 (MPL 2.0)',
    keywords='nose selenium testresources',
    classifiers=[
        'Development Status :: 2 - Pre-Alpha',
        'Intended Audience :: Developers',
        'License :: OSI Approved :: Mozilla Public License 2.0 (MPL 2.0)',
        'Operating System :: POSIX',
        'Operating System :: Microsoft :: Windows',
        'Operating System :: MacOS :: MacOS X',
        'Topic :: Software Development :: Quality Assurance',
        'Topic :: Software Development :: Testing',
        'Topic :: Utilities',
        'Programming Language :: Python',
        'Programming Language :: Python :: 2.6',
        'Programming Language :: Python :: 2.7'],
    install_requires=['nose', 'testresources', 'unittest2'],
)