diff --git a/billiard/einfo.py b/billiard/einfo.py index f82c5f5..26c0e69 100644 --- a/billiard/einfo.py +++ b/billiard/einfo.py @@ -1,5 +1,6 @@ import sys import traceback +import types __all__ = ['ExceptionInfo', 'Traceback'] @@ -25,6 +26,10 @@ def __init__(self, code): if sys.version_info >= (3, 11): self._co_positions = list(code.co_positions()) + @property + def __class__(self): + return types.CodeType + if sys.version_info >= (3, 11): @property def co_positions(self): @@ -57,6 +62,10 @@ def __init__(self, frame): # don't want to hit https://bugs.python.org/issue21967 self.f_restricted = False + @property + def __class__(self): + return types.FrameType + class _Object: @@ -79,6 +88,10 @@ def __init__(self): self.tb_next = None self.tb_lasti = 0 + @property + def __class__(self): + return types.TracebackType + class Traceback: Frame = _Frame @@ -94,6 +107,10 @@ def __init__(self, tb, max_frames=DEFAULT_MAX_FRAMES, depth=0): else: self.tb_next = _Truncated() + @property + def __class__(self): + return types.TracebackType + class RemoteTraceback(Exception): def __init__(self, tb): diff --git a/t/unit/test_einfo.py b/t/unit/test_einfo.py index 5d2482d..dff7b0b 100644 --- a/t/unit/test_einfo.py +++ b/t/unit/test_einfo.py @@ -1,6 +1,9 @@ +import sys import pickle import logging -from billiard.einfo import ExceptionInfo +import inspect +import types +from billiard.einfo import ExceptionInfo, Traceback, _Code, _Frame logger = logging.getLogger(__name__) @@ -26,3 +29,36 @@ def test_exception_info_log_after_pickle(caplog): logger.exception("failed", exc_info=exception) assert ' raise RuntimeError("some message")' in caplog.text assert "RuntimeError: some message" in caplog.text + + +def make_python_tb(): + tb = None + depth = 0 + while True: + try: + frame = sys._getframe(depth) + except ValueError: + break + else: + depth += 1 + + tb = types.TracebackType(tb, frame, frame.f_lasti, frame.f_lineno) + + return tb + + +class test_inspect: + def test_istraceback(self): + assert inspect.istraceback(Traceback(tb=make_python_tb())) + + def test_isframe(self): + assert inspect.isframe(_Frame(make_python_tb().tb_frame)) + + def test_iscode(self): + assert inspect.iscode(_Code(make_python_tb().tb_frame.f_code)) + + def test_getframeinfo(self): + assert inspect.getframeinfo(Traceback(tb=make_python_tb())) + + def test_getinnerframes(self): + assert inspect.getinnerframes(Traceback(tb=make_python_tb()))