From e0e3e1717e3d33762ce61950504f9637a6e669ed Mon Sep 17 00:00:00 2001
From: nkozlovskiy <nmk@ydb.tech>
Date: Fri, 29 Sep 2023 12:24:06 +0300
Subject: add ydb deps
---
contrib/python/pytest/py3/_pytest/unittest.py | 421 ++++++++++++++++++++++++++
1 file changed, 421 insertions(+)
create mode 100644 contrib/python/pytest/py3/_pytest/unittest.py
(limited to 'contrib/python/pytest/py3/_pytest/unittest.py')
diff --git a/contrib/python/pytest/py3/_pytest/unittest.py b/contrib/python/pytest/py3/_pytest/unittest.py
new file mode 100644
index 0000000000..d42a12a3a9
--- /dev/null
+++ b/contrib/python/pytest/py3/_pytest/unittest.py
@@ -0,0 +1,421 @@
+"""Discover and run std-library "unittest" style tests."""
+import sys
+import traceback
+import types
+from typing import Any
+from typing import Callable
+from typing import Generator
+from typing import Iterable
+from typing import List
+from typing import Optional
+from typing import Tuple
+from typing import Type
+from typing import TYPE_CHECKING
+from typing import Union
+
+import _pytest._code
+import pytest
+from _pytest.compat import getimfunc
+from _pytest.compat import is_async_function
+from _pytest.config import hookimpl
+from _pytest.fixtures import FixtureRequest
+from _pytest.nodes import Collector
+from _pytest.nodes import Item
+from _pytest.outcomes import exit
+from _pytest.outcomes import fail
+from _pytest.outcomes import skip
+from _pytest.outcomes import xfail
+from _pytest.python import Class
+from _pytest.python import Function
+from _pytest.python import Module
+from _pytest.runner import CallInfo
+from _pytest.scope import Scope
+
+if TYPE_CHECKING:
+ import unittest
+ import twisted.trial.unittest
+
+ _SysExcInfoType = Union[
+ Tuple[Type[BaseException], BaseException, types.TracebackType],
+ Tuple[None, None, None],
+ ]
+
+
+def pytest_pycollect_makeitem(
+ collector: Union[Module, Class], name: str, obj: object
+) -> Optional["UnitTestCase"]:
+ # Has unittest been imported and is obj a subclass of its TestCase?
+ try:
+ ut = sys.modules["unittest"]
+ # Type ignored because `ut` is an opaque module.
+ if not issubclass(obj, ut.TestCase): # type: ignore
+ return None
+ except Exception:
+ return None
+ # Yes, so let's collect it.
+ item: UnitTestCase = UnitTestCase.from_parent(collector, name=name, obj=obj)
+ return item
+
+
+class UnitTestCase(Class):
+ # Marker for fixturemanger.getfixtureinfo()
+ # to declare that our children do not support funcargs.
+ nofuncargs = True
+
+ def collect(self) -> Iterable[Union[Item, Collector]]:
+ from unittest import TestLoader
+
+ cls = self.obj
+ if not getattr(cls, "__test__", True):
+ return
+
+ skipped = _is_skipped(cls)
+ if not skipped:
+ self._inject_setup_teardown_fixtures(cls)
+ self._inject_setup_class_fixture()
+
+ self.session._fixturemanager.parsefactories(self, unittest=True)
+ loader = TestLoader()
+ foundsomething = False
+ for name in loader.getTestCaseNames(self.obj):
+ x = getattr(self.obj, name)
+ if not getattr(x, "__test__", True):
+ continue
+ funcobj = getimfunc(x)
+ yield TestCaseFunction.from_parent(self, name=name, callobj=funcobj)
+ foundsomething = True
+
+ if not foundsomething:
+ runtest = getattr(self.obj, "runTest", None)
+ if runtest is not None:
+ ut = sys.modules.get("twisted.trial.unittest", None)
+ # Type ignored because `ut` is an opaque module.
+ if ut is None or runtest != ut.TestCase.runTest: # type: ignore
+ yield TestCaseFunction.from_parent(self, name="runTest")
+
+ def _inject_setup_teardown_fixtures(self, cls: type) -> None:
+ """Injects a hidden auto-use fixture to invoke setUpClass/setup_method and corresponding
+ teardown functions (#517)."""
+ class_fixture = _make_xunit_fixture(
+ cls,
+ "setUpClass",
+ "tearDownClass",
+ "doClassCleanups",
+ scope=Scope.Class,
+ pass_self=False,
+ )
+ if class_fixture:
+ cls.__pytest_class_setup = class_fixture # type: ignore[attr-defined]
+
+ method_fixture = _make_xunit_fixture(
+ cls,
+ "setup_method",
+ "teardown_method",
+ None,
+ scope=Scope.Function,
+ pass_self=True,
+ )
+ if method_fixture:
+ cls.__pytest_method_setup = method_fixture # type: ignore[attr-defined]
+
+
+def _make_xunit_fixture(
+ obj: type,
+ setup_name: str,
+ teardown_name: str,
+ cleanup_name: Optional[str],
+ scope: Scope,
+ pass_self: bool,
+):
+ setup = getattr(obj, setup_name, None)
+ teardown = getattr(obj, teardown_name, None)
+ if setup is None and teardown is None:
+ return None
+
+ if cleanup_name:
+ cleanup = getattr(obj, cleanup_name, lambda *args: None)
+ else:
+
+ def cleanup(*args):
+ pass
+
+ @pytest.fixture(
+ scope=scope.value,
+ autouse=True,
+ # Use a unique name to speed up lookup.
+ name=f"_unittest_{setup_name}_fixture_{obj.__qualname__}",
+ )
+ def fixture(self, request: FixtureRequest) -> Generator[None, None, None]:
+ if _is_skipped(self):
+ reason = self.__unittest_skip_why__
+ raise pytest.skip.Exception(reason, _use_item_location=True)
+ if setup is not None:
+ try:
+ if pass_self:
+ setup(self, request.function)
+ else:
+ setup()
+ # unittest does not call the cleanup function for every BaseException, so we
+ # follow this here.
+ except Exception:
+ if pass_self:
+ cleanup(self)
+ else:
+ cleanup()
+
+ raise
+ yield
+ try:
+ if teardown is not None:
+ if pass_self:
+ teardown(self, request.function)
+ else:
+ teardown()
+ finally:
+ if pass_self:
+ cleanup(self)
+ else:
+ cleanup()
+
+ return fixture
+
+
+class TestCaseFunction(Function):
+ nofuncargs = True
+ _excinfo: Optional[List[_pytest._code.ExceptionInfo[BaseException]]] = None
+ _testcase: Optional["unittest.TestCase"] = None
+
+ def _getobj(self):
+ assert self.parent is not None
+ # Unlike a regular Function in a Class, where `item.obj` returns
+ # a *bound* method (attached to an instance), TestCaseFunction's
+ # `obj` returns an *unbound* method (not attached to an instance).
+ # This inconsistency is probably not desirable, but needs some
+ # consideration before changing.
+ return getattr(self.parent.obj, self.originalname) # type: ignore[attr-defined]
+
+ def setup(self) -> None:
+ # A bound method to be called during teardown() if set (see 'runtest()').
+ self._explicit_tearDown: Optional[Callable[[], None]] = None
+ assert self.parent is not None
+ self._testcase = self.parent.obj(self.name) # type: ignore[attr-defined]
+ self._obj = getattr(self._testcase, self.name)
+ if hasattr(self, "_request"):
+ self._request._fillfixtures()
+
+ def teardown(self) -> None:
+ if self._explicit_tearDown is not None:
+ self._explicit_tearDown()
+ self._explicit_tearDown = None
+ self._testcase = None
+ self._obj = None
+
+ def startTest(self, testcase: "unittest.TestCase") -> None:
+ pass
+
+ def _addexcinfo(self, rawexcinfo: "_SysExcInfoType") -> None:
+ # Unwrap potential exception info (see twisted trial support below).
+ rawexcinfo = getattr(rawexcinfo, "_rawexcinfo", rawexcinfo)
+ try:
+ excinfo = _pytest._code.ExceptionInfo[BaseException].from_exc_info(rawexcinfo) # type: ignore[arg-type]
+ # Invoke the attributes to trigger storing the traceback
+ # trial causes some issue there.
+ excinfo.value
+ excinfo.traceback
+ except TypeError:
+ try:
+ try:
+ values = traceback.format_exception(*rawexcinfo)
+ values.insert(
+ 0,
+ "NOTE: Incompatible Exception Representation, "
+ "displaying natively:\n\n",
+ )
+ fail("".join(values), pytrace=False)
+ except (fail.Exception, KeyboardInterrupt):
+ raise
+ except BaseException:
+ fail(
+ "ERROR: Unknown Incompatible Exception "
+ "representation:\n%r" % (rawexcinfo,),
+ pytrace=False,
+ )
+ except KeyboardInterrupt:
+ raise
+ except fail.Exception:
+ excinfo = _pytest._code.ExceptionInfo.from_current()
+ self.__dict__.setdefault("_excinfo", []).append(excinfo)
+
+ def addError(
+ self, testcase: "unittest.TestCase", rawexcinfo: "_SysExcInfoType"
+ ) -> None:
+ try:
+ if isinstance(rawexcinfo[1], exit.Exception):
+ exit(rawexcinfo[1].msg)
+ except TypeError:
+ pass
+ self._addexcinfo(rawexcinfo)
+
+ def addFailure(
+ self, testcase: "unittest.TestCase", rawexcinfo: "_SysExcInfoType"
+ ) -> None:
+ self._addexcinfo(rawexcinfo)
+
+ def addSkip(self, testcase: "unittest.TestCase", reason: str) -> None:
+ try:
+ raise pytest.skip.Exception(reason, _use_item_location=True)
+ except skip.Exception:
+ self._addexcinfo(sys.exc_info())
+
+ def addExpectedFailure(
+ self,
+ testcase: "unittest.TestCase",
+ rawexcinfo: "_SysExcInfoType",
+ reason: str = "",
+ ) -> None:
+ try:
+ xfail(str(reason))
+ except xfail.Exception:
+ self._addexcinfo(sys.exc_info())
+
+ def addUnexpectedSuccess(
+ self,
+ testcase: "unittest.TestCase",
+ reason: Optional["twisted.trial.unittest.Todo"] = None,
+ ) -> None:
+ msg = "Unexpected success"
+ if reason:
+ msg += f": {reason.reason}"
+ # Preserve unittest behaviour - fail the test. Explicitly not an XPASS.
+ try:
+ fail(msg, pytrace=False)
+ except fail.Exception:
+ self._addexcinfo(sys.exc_info())
+
+ def addSuccess(self, testcase: "unittest.TestCase") -> None:
+ pass
+
+ def stopTest(self, testcase: "unittest.TestCase") -> None:
+ pass
+
+ def addDuration(self, testcase: "unittest.TestCase", elapsed: float) -> None:
+ pass
+
+ def runtest(self) -> None:
+ from _pytest.debugging import maybe_wrap_pytest_function_for_tracing
+
+ assert self._testcase is not None
+
+ maybe_wrap_pytest_function_for_tracing(self)
+
+ # Let the unittest framework handle async functions.
+ if is_async_function(self.obj):
+ # Type ignored because self acts as the TestResult, but is not actually one.
+ self._testcase(result=self) # type: ignore[arg-type]
+ else:
+ # When --pdb is given, we want to postpone calling tearDown() otherwise
+ # when entering the pdb prompt, tearDown() would have probably cleaned up
+ # instance variables, which makes it difficult to debug.
+ # Arguably we could always postpone tearDown(), but this changes the moment where the
+ # TestCase instance interacts with the results object, so better to only do it
+ # when absolutely needed.
+ # We need to consider if the test itself is skipped, or the whole class.
+ assert isinstance(self.parent, UnitTestCase)
+ skipped = _is_skipped(self.obj) or _is_skipped(self.parent.obj)
+ if self.config.getoption("usepdb") and not skipped:
+ self._explicit_tearDown = self._testcase.tearDown
+ setattr(self._testcase, "tearDown", lambda *args: None)
+
+ # We need to update the actual bound method with self.obj, because
+ # wrap_pytest_function_for_tracing replaces self.obj by a wrapper.
+ setattr(self._testcase, self.name, self.obj)
+ try:
+ self._testcase(result=self) # type: ignore[arg-type]
+ finally:
+ delattr(self._testcase, self.name)
+
+ def _traceback_filter(
+ self, excinfo: _pytest._code.ExceptionInfo[BaseException]
+ ) -> _pytest._code.Traceback:
+ traceback = super()._traceback_filter(excinfo)
+ ntraceback = traceback.filter(
+ lambda x: not x.frame.f_globals.get("__unittest"),
+ )
+ if not ntraceback:
+ ntraceback = traceback
+ return ntraceback
+
+
+@hookimpl(tryfirst=True)
+def pytest_runtest_makereport(item: Item, call: CallInfo[None]) -> None:
+ if isinstance(item, TestCaseFunction):
+ if item._excinfo:
+ call.excinfo = item._excinfo.pop(0)
+ try:
+ del call.result
+ except AttributeError:
+ pass
+
+ # Convert unittest.SkipTest to pytest.skip.
+ # This is actually only needed for nose, which reuses unittest.SkipTest for
+ # its own nose.SkipTest. For unittest TestCases, SkipTest is already
+ # handled internally, and doesn't reach here.
+ unittest = sys.modules.get("unittest")
+ if (
+ unittest
+ and call.excinfo
+ and isinstance(call.excinfo.value, unittest.SkipTest) # type: ignore[attr-defined]
+ ):
+ excinfo = call.excinfo
+ call2 = CallInfo[None].from_call(
+ lambda: pytest.skip(str(excinfo.value)), call.when
+ )
+ call.excinfo = call2.excinfo
+
+
+# Twisted trial support.
+
+
+@hookimpl(hookwrapper=True)
+def pytest_runtest_protocol(item: Item) -> Generator[None, None, None]:
+ if isinstance(item, TestCaseFunction) and "twisted.trial.unittest" in sys.modules:
+ ut: Any = sys.modules["twisted.python.failure"]
+ Failure__init__ = ut.Failure.__init__
+ check_testcase_implements_trial_reporter()
+
+ def excstore(
+ self, exc_value=None, exc_type=None, exc_tb=None, captureVars=None
+ ):
+ if exc_value is None:
+ self._rawexcinfo = sys.exc_info()
+ else:
+ if exc_type is None:
+ exc_type = type(exc_value)
+ self._rawexcinfo = (exc_type, exc_value, exc_tb)
+ try:
+ Failure__init__(
+ self, exc_value, exc_type, exc_tb, captureVars=captureVars
+ )
+ except TypeError:
+ Failure__init__(self, exc_value, exc_type, exc_tb)
+
+ ut.Failure.__init__ = excstore
+ yield
+ ut.Failure.__init__ = Failure__init__
+ else:
+ yield
+
+
+def check_testcase_implements_trial_reporter(done: List[int] = []) -> None:
+ if done:
+ return
+ from zope.interface import classImplements
+ from twisted.trial.itrial import IReporter
+
+ classImplements(TestCaseFunction, IReporter)
+ done.append(1)
+
+
+def _is_skipped(obj) -> bool:
+ """Return True if the given object has been marked with @unittest.skip."""
+ return bool(getattr(obj, "__unittest_skip__", False))
--
cgit v1.2.3