feat!: remove deprecated test runnner logilab.common.pytest

logilab/common/testlib.py

@@ -42,7 +42,6 @@
 # disable camel case warning
 # pylint: disable=C0103
 
-from contextlib import contextmanager
 import sys
 import os
 import os.path as osp
@@ -46,5 +45,10 @@
 import sys
 import os
 import os.path as osp
+import types
+import doctest
+import inspect
+import unittest
+import traceback
 import tempfile
 import warnings
@@ -49,6 +53,11 @@
 import tempfile
 import warnings
-from inspect import isgeneratorfunction
-from typing import Any, Iterator, Union, Optional, Callable, Dict, List, Tuple
+
+from inspect import isgeneratorfunction, isclass, FrameInfo
+from functools import wraps
+from itertools import dropwhile
+from contextlib import contextmanager
+
+from typing import Any, Iterator, Union, Optional, Callable, Dict, List, Tuple, Generator
 from mypy_extensions import NoReturn
 
@@ -53,10 +62,6 @@
 from mypy_extensions import NoReturn
 
-import doctest
-
-import unittest as unittest_legacy
-
-from functools import wraps
-
+from logilab.common import textutils
+from logilab.common.debugger import Debugger, colorize_source
 from logilab.common.decorators import cached, classproperty
 
@@ -61,17 +66,5 @@
 from logilab.common.decorators import cached, classproperty
 
-if not getattr(unittest_legacy, "__package__", None):
-    try:
-        import unittest2 as unittest
-        from unittest2 import SkipTest
-    except ImportError:
-        raise ImportError(f"You have to install python-unittest2 to use {__name__}")
-else:
-    # mypy: Name 'unittest' already defined (possibly by an import)
-    # compat
-    import unittest as unittest  # type: ignore
-    from unittest import SkipTest
-
 
 __all__ = ["unittest_main", "find_tests", "nocoverage", "pause_trace"]
 
@@ -216,7 +209,7 @@
 unittest_main = unittest.main
 
 
-class InnerTestSkipped(SkipTest):
+class InnerTestSkipped(unittest.SkipTest):
     """raised when a test is skipped"""
 
 
@@ -445,7 +438,7 @@
         except self.failureException:
             result.addFailure(self, self.__exc_info())
             success = False
-        except SkipTest as e:
+        except unittest.SkipTest as e:
             result.addSkip(self, e)
         except Exception:
             # if an error occurs between two yield
@@ -478,7 +471,7 @@
         except InnerTestSkipped as e:
             result.addSkip(self, e)
             return 1
-        except SkipTest as e:
+        except unittest.SkipTest as e:
             result.addSkip(self, e)
             return 0
         except Exception:
@@ -665,3 +658,377 @@
             return new_f
 
     return check_require_module
+
+
+class SkipAwareTextTestRunner(unittest.TextTestRunner):
+    def __init__(
+        self,
+        stream=sys.stderr,
+        verbosity=1,
+        exitfirst=False,
+        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.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 does_match_tags(self, test: Callable) -> bool:
+        if self.options is not None:
+            tags_pattern = getattr(self.options, "tags_pattern", None)
+            if tags_pattern is not None:
+                tags = getattr(test, "tags", Tags())
+                if tags.inherit and isinstance(test, types.MethodType):
+                    tags = tags | getattr(test.__self__.__class__, "tags", Tags())
+                return tags.match(tags_pattern)
+        return True  # no pattern
+
+    def _makeResult(self) -> "SkipAwareTestResult":
+        return SkipAwareTestResult(
+            self.stream,
+            self.descriptions,
+            self.verbosity,
+            self.exitfirst,
+            self.pdbmode,
+            self.cvg,
+            self.colorize,
+        )
+
+
+class SkipAwareTestResult(unittest._TextTestResult):
+    def __init__(
+        self,
+        stream,
+        descriptions: bool,
+        verbosity: int,
+        exitfirst: bool = False,
+        pdbmode: bool = False,
+        cvg: Optional[Any] = None,
+        colorize: bool = False,
+    ) -> None:
+        super(SkipAwareTestResult, self).__init__(stream, descriptions, verbosity)
+        self.skipped: List[Tuple[Any, Any]] = []
+        self.debuggers: List = []
+        self.fail_descrs: List = []
+        self.error_descrs: List = []
+        self.exitfirst = exitfirst
+        self.pdbmode = pdbmode
+        self.cvg = cvg
+        self.colorize = colorize
+        self.pdbclass = Debugger
+        self.verbose = verbosity > 1
+
+    def descrs_for(self, flavour: str) -> List[Tuple[int, str]]:
+        return getattr(self, f"{flavour.lower()}_descrs")
+
+    def _create_pdb(self, test_descr: str, flavour: str) -> None:
+        self.descrs_for(flavour).append((len(self.debuggers), test_descr))
+        if self.pdbmode:
+            self.debuggers.append(self.pdbclass(sys.exc_info()[2]))
+
+    def _iter_valid_frames(self, frames: List[FrameInfo]) -> Generator[FrameInfo, Any, None]:
+        """only consider non-testlib frames when formatting  traceback"""
+
+        def invalid(fi):
+            return osp.abspath(fi[1]) in (lgc_testlib, std_testlib)
+
+        lgc_testlib = osp.abspath(__file__)
+        std_testlib = osp.abspath(unittest.__file__)
+
+        for frameinfo in dropwhile(invalid, frames):
+            yield frameinfo
+
+    def _exc_info_to_string(self, err, test):
+        """Converts a sys.exc_info()-style tuple of values into a string.
+
+        This method is overridden here because we want to colorize
+        lines if --color is passed, and display local variables if
+        --verbose is passed
+        """
+        exctype, exc, tb = err
+        output = ["Traceback (most recent call last)"]
+        frames = inspect.getinnerframes(tb)
+        colorize = self.colorize
+        frames = enumerate(self._iter_valid_frames(frames))
+        for index, (frame, filename, lineno, funcname, ctx, ctxindex) in frames:
+            filename = osp.abspath(filename)
+            if ctx is None:  # pyc files or C extensions for instance
+                source = "<no source available>"
+            else:
+                source = "".join(ctx)
+            if colorize:
+                filename = textutils.colorize_ansi(filename, "magenta")
+                source = colorize_source(source)
+            output.append(f'  File "{filename}", line {lineno}, in {funcname}')
+            output.append(f"    {source.strip()}")
+            if self.verbose:
+                output.append(f"{dir(frame)!r} == {test.__module__!r}")
+                output.append("")
+                output.append("    " + " local variables ".center(66, "-"))
+                for varname, value in sorted(frame.f_locals.items()):
+                    output.append(f"    {varname}: {value!r}")
+                    if varname == "self":  # special handy processing for self
+                        for varname, value in sorted(vars(value).items()):
+                            output.append(f"      self.{varname}: {value!r}")
+                output.append("    " + "-" * 66)
+                output.append("")
+        output.append("".join(traceback.format_exception_only(exctype, exc)))
+        return "\n".join(output)
+
+    def addError(self, test, err):
+        """err ->  (exc_type, exc, tcbk)"""
+        exc_type, exc, _ = err
+        if isinstance(exc, unittest.SkipTest):
+            assert exc_type == unittest.SkipTest
+            self.addSkip(test, exc)
+        else:
+            if self.exitfirst:
+                self.shouldStop = True
+            descr = self.getDescription(test)
+            super(SkipAwareTestResult, self).addError(test, err)
+            self._create_pdb(descr, "error")
+
+    def addFailure(self, test, err):
+        if self.exitfirst:
+            self.shouldStop = True
+        descr = self.getDescription(test)
+        super(SkipAwareTestResult, self).addFailure(test, err)
+        self._create_pdb(descr, "fail")
+
+    def addSkip(self, test, reason):
+        self.skipped.append((test, reason))
+        if self.showAll:
+            self.stream.writeln("SKIPPED")
+        elif self.dots:
+            self.stream.write("S")
+
+    def printErrors(self) -> None:
+        super(SkipAwareTestResult, self).printErrors()
+        self.printSkippedList()
+
+    def printSkippedList(self) -> None:
+        # format (test, err) compatible with unittest2
+        for test, err in self.skipped:
+            descr = self.getDescription(test)
+            self.stream.writeln(self.separator1)
+            self.stream.writeln(f"{'SKIPPED'}: {descr}")
+            self.stream.writeln(f"\t{err}")
+
+    def printErrorList(self, flavour, errors):
+        for (_, descr), (test, err) in zip(self.descrs_for(flavour), errors):
+            self.stream.writeln(self.separator1)
+            self.stream.writeln(f"{flavour}: {descr}")
+            self.stream.writeln(self.separator2)
+            self.stream.writeln(err)
+            self.stream.writeln("no stdout".center(len(self.separator2)))
+            self.stream.writeln("no stderr".center(len(self.separator2)))
+
+
+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) -> None:
+        self.skipped_patterns = ()
+
+    # some magic here to accept empty list by extending
+    # and to provide callable capability
+    def loadTestsFromNames(self, names: List[str], module: type = None) -> TestSuite:
+        suites = []
+        for name in names:
+            suites.extend(self.loadTestsFromName(name, module))
+        return self.suiteClass(suites)
+
+    def _collect_tests(self, module: type) -> Dict[str, Tuple[type, List[str]]]:
+        tests = {}
+        for obj in vars(module).values():
+            if isclass(obj) 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: str) -> bool:
+        # mypy: Need type annotation for 'pat'
+        # doc doesn't say how to that in list comprehension
+        return any([(pat in testedname) for pat in self.skipped_patterns])  # type: ignore
+
+    def getTestCaseNames(self, testCaseClass: type) -> List[str]:
+        """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)]
+
+
+# The 2 functions below are modified versions of the TestSuite.run method
+# that is provided with unittest2 for python 2.6, in unittest2/suite.py
+# It is used to monkeypatch the original implementation to support
+# extra runcondition and options arguments (see in testlib.py)
+
+
+def _ts_wrapped_run(
+    self: Any,
+    result: SkipAwareTestResult,
+    debug: bool = False,
+    runcondition: Callable = None,
+    options: Optional[Any] = None,
+) -> SkipAwareTestResult:
+    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
+
+        # --- modifications to deal with _wrapped_run ---
+        # original code is:
+        #
+        # if not debug:
+        #     test(result)
+        # else:
+        #     test.debug()
+        if hasattr(test, "_wrapped_run"):
+            try:
+                test._wrapped_run(result, debug, runcondition=runcondition, options=options)
+            except TypeError:
+                test._wrapped_run(result, debug)
+        elif not debug:
+            try:
+                test(result, runcondition, options)
+            except TypeError:
+                test(result)
+        else:
+            test.debug()
+        # --- end of modifications to deal with _wrapped_run ---
+    return result
+
+
+def _ts_run(  # noqa
+    self: Any,
+    result: SkipAwareTestResult,
+    debug: bool = False,
+    runcondition: Callable = None,
+    options: Optional[Any] = None,
+) -> SkipAwareTestResult:
+    topLevel = False
+    if getattr(result, "_testRunEntered", False) is False:
+        result._testRunEntered = topLevel = True
+
+    self._wrapped_run(result, debug, runcondition, options)
+
+    if topLevel:
+        self._tearDownPreviousClass(None, result)
+        self._handleModuleTearDown(result)
+        result._testRunEntered = False
+    return result
+
+
+# monkeypatch unittest and doctest (ouch !)
+unittest._TextTestResult = SkipAwareTestResult
+unittest.TextTestRunner = SkipAwareTextTestRunner
+unittest.TestLoader = NonStrictTestLoader
+
+unittest.FunctionTestCase.__bases__ = (TestCase,)
+unittest.TestSuite.run = _ts_run
+
+unittest.TestSuite._wrapped_run = _ts_wrapped_run

test/test_pytest.py

-# copyright 2003-2011 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
-# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr
-#
-# This file is part of logilab-common.
-#
-# logilab-common is free software: you can redistribute it and/or modify it under
-# the terms of the GNU Lesser General Public License as published by the Free
-# Software Foundation, either version 2.1 of the License, or (at your option) any
-# later version.
-#
-# logilab-common 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 Lesser General Public License for more
-# details.
-#
-# You should have received a copy of the GNU Lesser General Public License along
-# with logilab-common.  If not, see <http://www.gnu.org/licenses/>.
-import sys
-
-from os.path import join
-from logilab.common.testlib import TestCase, unittest_main
-from logilab.common.pytest import (
-    this_is_a_testdir,
-    this_is_a_testfile,
-    replace_trace,
-    pause_trace,
-    nocoverage,
-)
-
-
-class ModuleFunctionTC(TestCase):
-    def test_this_is_testdir(self):
-        self.assertTrue(this_is_a_testdir("test"))
-        self.assertTrue(this_is_a_testdir("tests"))
-        self.assertTrue(this_is_a_testdir("unittests"))
-        self.assertTrue(this_is_a_testdir("unittest"))
-        self.assertFalse(this_is_a_testdir("unit"))
-        self.assertFalse(this_is_a_testdir("units"))
-        self.assertFalse(this_is_a_testdir("undksjhqfl"))
-        self.assertFalse(this_is_a_testdir("this_is_not_a_dir_test"))
-        self.assertFalse(this_is_a_testdir("this_is_not_a_testdir"))
-        self.assertFalse(this_is_a_testdir("unittestsarenothere"))
-        self.assertTrue(this_is_a_testdir(join("coincoin", "unittests")))
-        self.assertFalse(this_is_a_testdir(join("unittests", "spongebob")))
-
-    def test_this_is_testfile(self):
-        self.assertTrue(this_is_a_testfile("test.py"))
-        self.assertTrue(this_is_a_testfile("testbabar.py"))
-        self.assertTrue(this_is_a_testfile("unittest_celestine.py"))
-        self.assertTrue(this_is_a_testfile("smoketest.py"))
-        self.assertFalse(this_is_a_testfile("test.pyc"))
-        self.assertFalse(this_is_a_testfile("zephir_test.py"))
-        self.assertFalse(this_is_a_testfile("smoketest.pl"))
-        self.assertFalse(this_is_a_testfile("unittest"))
-        self.assertTrue(this_is_a_testfile(join("coincoin", "unittest_bibi.py")))
-        self.assertFalse(this_is_a_testfile(join("unittest", "spongebob.py")))
-
-    def test_replace_trace(self):
-        def tracefn(frame, event, arg):
-            pass
-
-        oldtrace = sys.gettrace()
-        with replace_trace(tracefn):
-            self.assertIs(sys.gettrace(), tracefn)
-
-        self.assertIs(sys.gettrace(), oldtrace)
-
-    def test_pause_trace(self):
-        def tracefn(frame, event, arg):
-            pass
-
-        oldtrace = sys.gettrace()
-        sys.settrace(tracefn)
-        try:
-            self.assertIs(sys.gettrace(), tracefn)
-            with pause_trace():
-                self.assertIsNone(sys.gettrace())
-            self.assertIs(sys.gettrace(), tracefn)
-        finally:
-            sys.settrace(oldtrace)
-
-    def test_nocoverage(self):
-        def tracefn(frame, event, arg):
-            pass
-
-        @nocoverage
-        def myfn():
-            self.assertIsNone(sys.gettrace())
-
-        with replace_trace(tracefn):
-            myfn()
-
-
-if __name__ == "__main__":
-    unittest_main()

test/test_testlib.py

@@ -42,7 +42,7 @@
     require_version,
     require_module,
 )
-from logilab.common.pytest import SkipAwareTextTestRunner, NonStrictTestLoader
+from logilab.common.testlib import SkipAwareTextTestRunner, NonStrictTestLoader
 
 
 class MockTestCase(TestCase):