[testlib] isolation of pytest related code into pytest.py

1 files changed, 463 insertions, 22 deletions

diff --git a/pytest.py b/pytest.py
index 2f71647..d2ec378 100644
--- a/pytest.py
+++ b/pytest.py
@@ -114,15 +114,21 @@ import os, sys, re
 import os.path as osp
 from time import time, clock
 import warnings
+import types
 
 from logilab.common.fileutils import abspath_listdir
 from logilab.common import testlib, STD_BLACKLIST
+from logilab.common.testlib import unittest
 import doctest
-import unittest
-
-
-import imp
 
+# use the same unittest module as testlib
+if sys.version_info < (3, 2):
+    try:
+        import unittest2.suite as unittest_suite
+    except ImportError:
+        sys.exit("You have to install python-unittest2 to use this module")
+else:
+    import unittest.suite as unittest_suite
 
 try:
     import django
@@ -178,17 +184,6 @@ def nocoverage(func):
 ## end of coverage hacks
 
 
-# monkeypatch unittest and doctest (ouch !)
-unittest.TestCase = testlib.TestCase
-unittest.main = testlib.unittest_main
-unittest._TextTestResult = testlib.SkipAwareTestResult
-unittest.TextTestRunner = testlib.SkipAwareTextTestRunner
-unittest.TestLoader = testlib.NonStrictTestLoader
-unittest.TestProgram = testlib.SkipAwareTestProgram
-if sys.version_info >= (2, 4):
-    doctest.DocTestCase.__bases__ = (testlib.TestCase,)
-else:
-    unittest.FunctionTestCase.__bases__ = (testlib.TestCase,)
 
 
 
@@ -260,7 +255,6 @@ class GlobalTestReport(object):
             problems = len(testresult.failures) + len(testresult.errors)
             self.errmodules.append((filename[:-3], problems, ran))
 
-
     def failed_to_test_module(self, filename):
         """called when the test module could not be imported by unittest
         """
@@ -427,7 +421,7 @@ succeeded test file :", osp.join(os.getcwd(), testlib.FILE_RESTART)
         try:
             tstart, cstart = time(), clock()
             try:
-                testprog = testlib.unittest_main(modname, batchmode=batchmode, cvg=self.cvg,
+                testprog = SkipAwareTestProgram(modname, batchmode=batchmode, cvg=self.cvg,
                                                  options=self.options, outstream=sys.stderr)
             except KeyboardInterrupt:
                 raise
@@ -486,7 +480,6 @@ class DjangoTester(PyTester):
         create_test_db(verbosity=0)
         self.dbname = self.settings.TEST_DATABASE_NAME
 
-
     def after_testfile(self):
         # Those imports must be done **after** setup_environ was called
         from django.test.utils import teardown_test_environment
@@ -495,7 +488,6 @@ class DjangoTester(PyTester):
         print 'destroying', self.dbname
         destroy_test_db(self.dbname, verbosity=0)
 
-
     def testall(self, exitfirst=False):
         """walks through current working directory, finds something
         which can be considered as a testdir and runs every test there
@@ -517,7 +509,6 @@ class DjangoTester(PyTester):
                         break
                     dirs[:] = []
 
-
     def testonedir(self, testdir, exitfirst=False):
         """finds each testfile in the `testdir` and runs it
 
@@ -542,7 +533,6 @@ class DjangoTester(PyTester):
         remove_local_modules_from_sys(testdir)
         return True
 
-
     def testfile(self, filename, batchmode=False):
         """runs every test in `filename`
 
@@ -559,7 +549,7 @@ class DjangoTester(PyTester):
             try:
                 tstart, cstart = time(), clock()
                 self.before_testfile()
-                testprog = testlib.unittest_main(modname, batchmode=batchmode, cvg=self.cvg)
+                testprog = SkipAwareTestProgram(modname, batchmode=batchmode, cvg=self.cvg)
                 tend, cend = time(), clock()
                 ttime, ctime = (tend - tstart), (cend - cstart)
                 self.report.feed(filename, testprog.result, ttime, ctime)
@@ -746,3 +736,454 @@ def run():
         if covermode:
             print 'coverage information stored, use it with pycoverage -ra'
         sys.exit(tester.errcode)
+
+class SkipAwareTestProgram(unittest.TestProgram):
+    # XXX: don't try to stay close to unittest.py, use optparse
+    USAGE = """\
+Usage: %(progName)s [options] [test] [...]
+
+Options:
+  -h, --help       Show this message
+  -v, --verbose    Verbose output
+  -i, --pdb        Enable test failure inspection
+  -x, --exitfirst  Exit on first failure
+  -c, --capture    Captures and prints standard out/err only on errors
+  -p, --printonly  Only prints lines matching specified pattern
+                   (implies capture)
+  -s, --skip       skip test matching this pattern (no regexp for now)
+  -q, --quiet      Minimal output
+  --color          colorize tracebacks
+
+  -m, --match      Run only test whose tag match this pattern
+
+  -P, --profile    FILE: Run the tests using cProfile and saving results
+                   in FILE
+
+Examples:
+  %(progName)s                               - run default set of tests
+  %(progName)s MyTestSuite                   - run suite 'MyTestSuite'
+  %(progName)s MyTestCase.testSomething      - run MyTestCase.testSomething
+  %(progName)s MyTestCase                    - run all 'test*' test methods
+                                               in MyTestCase
+"""
+    def __init__(self, module='__main__', defaultTest=None, batchmode=False,
+                 cvg=None, options=None, outstream=sys.stderr):
+        self.batchmode = batchmode
+        self.cvg = cvg
+        self.options = options
+        self.outstream = outstream
+        super(SkipAwareTestProgram, self).__init__(
+            module=module, defaultTest=defaultTest,
+            testLoader=NonStrictTestLoader())
+
+    def parseArgs(self, argv):
+        self.pdbmode = False
+        self.exitfirst = False
+        self.capture = 0
+        self.printonly = None
+        self.skipped_patterns = []
+        self.test_pattern = None
+        self.tags_pattern = None
+        self.colorize = False
+        self.profile_name = None
+        import getopt
+        try:
+            options, args = getopt.getopt(argv[1:], 'hHvixrqcp:s:m:P:',
+                                          ['help', 'verbose', 'quiet', 'pdb',
+                                           'exitfirst', 'restart', 'capture', 'printonly=',
+                                           'skip=', 'color', 'match=', 'profile='])
+            for opt, value in options:
+                if opt in ('-h', '-H', '--help'):
+                    self.usageExit()
+                if opt in ('-i', '--pdb'):
+                    self.pdbmode = True
+                if opt in ('-x', '--exitfirst'):
+                    self.exitfirst = True
+                if opt in ('-r', '--restart'):
+                    self.restart = True
+                    self.exitfirst = True
+                if opt in ('-q', '--quiet'):
+                    self.verbosity = 0
+                if opt in ('-v', '--verbose'):
+                    self.verbosity = 2
+                if opt in ('-c', '--capture'):
+                    self.capture += 1
+                if opt in ('-p', '--printonly'):
+                    self.printonly = re.compile(value)
+                if opt in ('-s', '--skip'):
+                    self.skipped_patterns = [pat.strip() for pat in
+                                             value.split(', ')]
+                if opt == '--color':
+                    self.colorize = True
+                if opt in ('-m', '--match'):
+                    #self.tags_pattern = value
+                    self.options["tag_pattern"] = value
+                if opt in ('-P', '--profile'):
+                    self.profile_name = value
+            self.testLoader.skipped_patterns = self.skipped_patterns
+            if self.printonly is not None:
+                self.capture += 1
+            if len(args) == 0 and self.defaultTest is None:
+                suitefunc = getattr(self.module, 'suite', None)
+                if isinstance(suitefunc, (types.FunctionType,
+                        types.MethodType)):
+                    self.test = self.module.suite()
+                else:
+                    self.test = self.testLoader.loadTestsFromModule(self.module)
+                return
+            if len(args) > 0:
+                self.test_pattern = args[0]
+                self.testNames = args
+            else:
+                self.testNames = (self.defaultTest, )
+            self.createTests()
+        except getopt.error, msg:
+            self.usageExit(msg)
+
+    def runTests(self):
+        if self.profile_name:
+            import cProfile
+            cProfile.runctx('self._runTests()', globals(), locals(), self.profile_name )
+        else:
+            return self._runTests()
+
+    def _runTests(self):
+        if hasattr(self.module, 'setup_module'):
+            try:
+                self.module.setup_module(self.options)
+            except Exception, exc:
+                print 'setup_module error:', exc
+                sys.exit(1)
+        self.testRunner = SkipAwareTextTestRunner(verbosity=self.verbosity,
+                                                  stream=self.outstream,
+                                                  exitfirst=self.exitfirst,
+                                                  capture=self.capture,
+                                                  printonly=self.printonly,
+                                                  pdbmode=self.pdbmode,
+                                                  cvg=self.cvg,
+                                                  test_pattern=self.test_pattern,
+                                                  skipped_patterns=self.skipped_patterns,
+                                                  colorize=self.colorize,
+                                                  batchmode=self.batchmode,
+                                                  options=self.options)
+
+        def removeSucceededTests(obj, succTests):
+            """ Recursive function that removes succTests from
+            a TestSuite or TestCase
+            """
+            if isinstance(obj, TestSuite):
+                removeSucceededTests(obj._tests, succTests)
+            if isinstance(obj, list):
+                for el in obj[:]:
+                    if isinstance(el, TestSuite):
+                        removeSucceededTests(el, succTests)
+                    elif isinstance(el, TestCase):
+                        descr = '.'.join((el.__class__.__module__,
+                                el.__class__.__name__,
+                                el._testMethodName))
+                        if descr in succTests:
+                            obj.remove(el)
+        # take care, self.options may be None
+        if getattr(self.options, 'restart', False):
+            # retrieve succeeded tests from FILE_RESTART
+            try:
+                restartfile = open(FILE_RESTART, 'r')
+                try:
+                    succeededtests = list(elem.rstrip('\n\r') for elem in
+                                          restartfile.readlines())
+                    removeSucceededTests(self.test, succeededtests)
+                finally:
+                    restartfile.close()
+            except Exception, ex:
+                raise Exception("Error while reading succeeded tests into %s: %s"
+                                % (osp.join(os.getcwd(), FILE_RESTART), ex))
+
+        result = self.testRunner.run(self.test)
+        # help garbage collection: we want TestSuite, which hold refs to every
+        # executed TestCase, to be gc'ed
+        del self.test
+        if hasattr(self.module, 'teardown_module'):
+            try:
+                self.module.teardown_module(self.options, result)
+            except Exception, exc:
+                print 'teardown_module error:', exc
+                sys.exit(1)
+        if getattr(result, "debuggers", None) and \
+           getattr(self, "pdbmode", None):
+            start_interactive_mode(result)
+        if not getattr(self, "batchmode", None):
+            sys.exit(not result.wasSuccessful())
+        self.result = result
+
+
+class SkipAwareTextTestRunner(unittest.TextTestRunner):
+
+    def __init__(self, stream=sys.stderr, verbosity=1,
+                 exitfirst=False, capture=False, printonly=None,
+                 pdbmode=False, cvg=None, test_pattern=None,
+                 skipped_patterns=(), colorize=False, batchmode=False,
+                 options=None):
+        super(SkipAwareTextTestRunner, self).__init__(stream=stream,
+                                                      verbosity=verbosity)
+        self.exitfirst = exitfirst
+        self.capture = capture
+        self.printonly = printonly
+        self.pdbmode = pdbmode
+        self.cvg = cvg
+        self.test_pattern = test_pattern
+        self.skipped_patterns = skipped_patterns
+        self.colorize = colorize
+        self.batchmode = batchmode
+        self.options = options
+
+    def _this_is_skipped(self, testedname):
+        return any([(pat in testedname) for pat in self.skipped_patterns])
+
+    def _runcondition(self, test, skipgenerator=True):
+        if isinstance(test, testlib.InnerTest):
+            testname = test.name
+        else:
+            if isinstance(test, testlib.TestCase):
+                meth = test._get_test_method()
+                func = meth.im_func
+                testname = '%s.%s' % (meth.im_class.__name__, func.__name__)
+            elif isinstance(test, types.FunctionType):
+                func = test
+                testname = func.__name__
+            elif isinstance(test, types.MethodType):
+                func = test.im_func
+                testname = '%s.%s' % (test.im_class.__name__, func.__name__)
+            else:
+                return True # Not sure when this happens
+            if testlib.is_generator(func) and skipgenerator:
+                return self.does_match_tags(func) # Let inner tests decide at run time
+        if self._this_is_skipped(testname):
+            return False # this was explicitly skipped
+        if self.test_pattern is not None:
+            try:
+                classpattern, testpattern = self.test_pattern.split('.')
+                klass, name = testname.split('.')
+                if classpattern not in klass or testpattern not in name:
+                    return False
+            except ValueError:
+                if self.test_pattern not in testname:
+                    return False
+
+        return self.does_match_tags(test)
+
+    def does_match_tags(self, test):
+        if self.options is not None:
+            tags_pattern = getattr(self.options, 'tags_pattern', None)
+            if tags_pattern is not None:
+                tags = getattr(test, 'tags', testlib.Tags())
+                if tags.inherit and isinstance(test, types.MethodType):
+                    tags = tags | getattr(test.im_class, 'tags', testlib.Tags())
+                return tags.match(tags_pattern)
+        return True # no pattern
+
+    def _makeResult(self):
+        return testlib.SkipAwareTestResult(self.stream, self.descriptions,
+                                   self.verbosity, self.exitfirst, self.capture,
+                                   self.printonly, self.pdbmode, self.cvg,
+                                   self.colorize)
+
+    def run(self, test):
+        "Run the given test case or test suite."
+        result = self._makeResult()
+        startTime = time()
+        test(result, runcondition=self._runcondition, options=self.options)
+        stopTime = time()
+        timeTaken = stopTime - startTime
+        result.printErrors()
+        if not self.batchmode:
+            self.stream.writeln(result.separator2)
+            run = result.testsRun
+            self.stream.writeln("Ran %d test%s in %.3fs" %
+                                (run, run != 1 and "s" or "", timeTaken))
+            self.stream.writeln()
+            if not result.wasSuccessful():
+                if self.colorize:
+                    self.stream.write(textutils.colorize_ansi("FAILED", color='red'))
+                else:
+                    self.stream.write("FAILED")
+            else:
+                if self.colorize:
+                    self.stream.write(textutils.colorize_ansi("OK", color='green'))
+                else:
+                    self.stream.write("OK")
+            failed, errored, skipped = map(len, (result.failures,
+                                                 result.errors,
+                                                 result.skipped))
+
+            det_results = []
+            for name, value in (("failures", result.failures),
+                                ("errors",result.errors),
+                                ("skipped", result.skipped)):
+                if value:
+                    det_results.append("%s=%i" % (name, len(value)))
+            if det_results:
+                self.stream.write(" (")
+                self.stream.write(', '.join(det_results))
+                self.stream.write(")")
+            self.stream.writeln("")
+        return result
+
+class NonStrictTestLoader(unittest.TestLoader):
+    """
+    Overrides default testloader to be able to omit classname when
+    specifying tests to run on command line.
+
+    For example, if the file test_foo.py contains ::
+
+        class FooTC(TestCase):
+            def test_foo1(self): # ...
+            def test_foo2(self): # ...
+            def test_bar1(self): # ...
+
+        class BarTC(TestCase):
+            def test_bar2(self): # ...
+
+    'python test_foo.py' will run the 3 tests in FooTC
+    'python test_foo.py FooTC' will run the 3 tests in FooTC
+    'python test_foo.py test_foo' will run test_foo1 and test_foo2
+    'python test_foo.py test_foo1' will run test_foo1
+    'python test_foo.py test_bar' will run FooTC.test_bar1 and BarTC.test_bar2
+    """
+
+    def __init__(self):
+        self.skipped_patterns = ()
+
+    # some magic here to accept empty list by extending
+    # and to provide callable capability
+    def loadTestsFromNames(self, names, module=None):
+        suites = []
+        for name in names:
+            suites.extend(self.loadTestsFromName(name, module))
+        return self.suiteClass(suites)
+
+    def _collect_tests(self, module):
+        tests = {}
+        for obj in vars(module).values():
+            if (issubclass(type(obj), (types.ClassType, type)) and
+                 issubclass(obj, unittest.TestCase)):
+                classname = obj.__name__
+                if classname[0] == '_' or self._this_is_skipped(classname):
+                    continue
+                methodnames = []
+                # obj is a TestCase class
+                for attrname in dir(obj):
+                    if attrname.startswith(self.testMethodPrefix):
+                        attr = getattr(obj, attrname)
+                        if callable(attr):
+                            methodnames.append(attrname)
+                # keep track of class (obj) for convenience
+                tests[classname] = (obj, methodnames)
+        return tests
+
+    def loadTestsFromSuite(self, module, suitename):
+        try:
+            suite = getattr(module, suitename)()
+        except AttributeError:
+            return []
+        assert hasattr(suite, '_tests'), \
+               "%s.%s is not a valid TestSuite" % (module.__name__, suitename)
+        # python2.3 does not implement __iter__ on suites, we need to return
+        # _tests explicitly
+        return suite._tests
+
+    def loadTestsFromName(self, name, module=None):
+        parts = name.split('.')
+        if module is None or len(parts) > 2:
+            # let the base class do its job here
+            return [super(NonStrictTestLoader, self).loadTestsFromName(name)]
+        tests = self._collect_tests(module)
+        collected = []
+        if len(parts) == 1:
+            pattern = parts[0]
+            if callable(getattr(module, pattern, None)
+                    )  and pattern not in tests:
+                # consider it as a suite
+                return self.loadTestsFromSuite(module, pattern)
+            if pattern in tests:
+                # case python unittest_foo.py MyTestTC
+                klass, methodnames = tests[pattern]
+                for methodname in methodnames:
+                    collected = [klass(methodname)
+                        for methodname in methodnames]
+            else:
+                # case python unittest_foo.py something
+                for klass, methodnames in tests.values():
+                    # skip methodname if matched by skipped_patterns
+                    for skip_pattern in self.skipped_patterns:
+                        methodnames = [methodname
+                                      for methodname in methodnames
+                                      if skip_pattern not in methodname]
+                    collected += [klass(methodname)
+                                  for methodname in methodnames
+                                  if pattern in methodname]
+        elif len(parts) == 2:
+            # case "MyClass.test_1"
+            classname, pattern = parts
+            klass, methodnames = tests.get(classname, (None, []))
+            for methodname in methodnames:
+                collected = [klass(methodname) for methodname in methodnames
+                            if pattern in methodname]
+        return collected
+
+    def _this_is_skipped(self, testedname):
+        return any([(pat in testedname) for pat in self.skipped_patterns])
+
+    def getTestCaseNames(self, testCaseClass):
+        """Return a sorted sequence of method names found within testCaseClass
+        """
+        is_skipped = self._this_is_skipped
+        classname = testCaseClass.__name__
+        if classname[0] == '_' or is_skipped(classname):
+            return []
+        testnames = super(NonStrictTestLoader, self).getTestCaseNames(
+                testCaseClass)
+        return [testname for testname in testnames if not is_skipped(testname)]
+
+def _ts_run(self, result, runcondition=None, options=None):
+    self._wrapped_run(result,runcondition=runcondition, options=options)
+    self._tearDownPreviousClass(None, result)
+    self._handleModuleTearDown(result)
+    return result
+
+def _ts_wrapped_run(self, result, debug=False, runcondition=None, options=None):
+    for test in self:
+        if result.shouldStop:
+            break
+        if unittest_suite._isnotsuite(test):
+            self._tearDownPreviousClass(test, result)
+            self._handleModuleFixture(test, result)
+            self._handleClassSetUp(test, result)
+            result._previousTestClass = test.__class__
+            if (getattr(test.__class__, '_classSetupFailed', False) or 
+                getattr(result, '_moduleSetUpFailed', False)):
+                continue
+
+        if hasattr(test, '_wrapped_run'):
+            test._wrapped_run(result, debug)
+        elif not debug:
+            try:
+                test(result, runcondition, options)
+            except TypeError:
+                test(result)
+        else:
+            test.debug()
+
+
+
+
+# monkeypatch unittest and doctest (ouch !)
+unittest._TextTestResult = testlib.SkipAwareTestResult
+unittest.TextTestRunner = SkipAwareTextTestRunner
+unittest.TestLoader = NonStrictTestLoader
+unittest.TestProgram = SkipAwareTestProgram
+if sys.version_info >= (2, 4):
+    doctest.DocTestCase.__bases__ = (testlib.TestCase,)
+else:
+    unittest.FunctionTestCase.__bases__ = (testlib.TestCase,)
+unittest.TestSuite.run = _ts_run
+unittest.TestSuite._wrapped_run = _ts_wrapped_run