diff --git a/offline_debug/__init__.py b/offline_debug/__init__.py index 47ace16..ca4b6bb 100644 --- a/offline_debug/__init__.py +++ b/offline_debug/__init__.py @@ -1,5 +1,6 @@ """Tool for serializing and reconstructing Python exceptions with full stack traces.""" -from .serializer import load_traceback, save_traceback +from ._inner.load_traceback import load_traceback +from ._inner.save_traceback import save_traceback __all__ = ["load_traceback", "save_traceback"] diff --git a/offline_debug/_inner/__init__.py b/offline_debug/_inner/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/offline_debug/_inner/c_api/__init__.py b/offline_debug/_inner/c_api/__init__.py new file mode 100644 index 0000000..8ea2d9d --- /dev/null +++ b/offline_debug/_inner/c_api/__init__.py @@ -0,0 +1,4 @@ +from ._create_frame import create_frame +from ._link_frame import link_frame + +__all__ = ["create_frame", "link_frame"] diff --git a/offline_debug/_inner/c_api/_create_frame.py b/offline_debug/_inner/c_api/_create_frame.py new file mode 100644 index 0000000..491dd99 --- /dev/null +++ b/offline_debug/_inner/c_api/_create_frame.py @@ -0,0 +1,93 @@ +""" +Abstraction of the CPython api for creating a traceback frame object. + +The reason we have to implement it, is that cpython doesn't expose +the api to create a FrameType using pure python. +""" + +from __future__ import annotations + +import ctypes +import sys +from functools import cache +from types import CodeType, FrameType +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + import _ctypes + + +@cache +def _get_py_frame_new() -> ctypes._NamedFuncPointer: + """Get the PyFrame_New C function configured with ctypes.""" + func: ctypes._NamedFuncPointer = ctypes.pythonapi.PyFrame_New + func.argtypes = ( + ctypes.c_void_p, # PyThreadState *tstate + ctypes.py_object, # PyCodeObject *code + ctypes.py_object, # PyObject *globals + ctypes.py_object, # PyObject *locals + ) + func.restype = ctypes.py_object + + def errcheck[T]( + result: T | None, + _func: _ctypes.CFuncPtr, + _args: tuple[Any, ...], + ) -> T: # pragma: no cover + if not result: + msg = "failed to create a new frame while calling" + raise RuntimeError(msg) + return result + + func.errcheck = errcheck + return func + + +@cache +def _get_py_thread_state_get() -> ctypes._NamedFuncPointer: + """Get the PyThreadState_Get C function configured with ctypes.""" + func = ctypes.pythonapi.PyThreadState_Get + func.argtypes = () + func.restype = ctypes.c_void_p + + def errcheck[T]( + result: T | None, _func: _ctypes.CFuncPtr, _args: tuple[Any, ...] + ) -> T: # pragma: no cover + if not result: + msg = "failed to get the current thread state" + raise RuntimeError(msg) + return result + + func.errcheck = errcheck + return func + + +def create_frame( + code: CodeType, + frame_globals: dict[str, Any], + frame_locals: dict[str, Any], + thread_state: int | None = None, +) -> FrameType: + """ + Create a FrameType object using the received parameters. + + If the thread state received is None, use the current thread state automatically. + """ + py_frame_new = _get_py_frame_new() + py_thread_state_get = _get_py_thread_state_get() + + if thread_state is None: + thread_state = py_thread_state_get() + + frame: FrameType = py_frame_new(thread_state, code, frame_globals, frame_locals) + + # In 3.13+, PEP 667 allows safe write-through access to locals. + # py_frame_new ignores the frame_locals argument, + # and must be assigned after the frame's creation in 3.13+ + if sys.version_info >= (3, 13) and frame_locals: + frame.f_locals.update(frame_locals) + + if not isinstance(frame, FrameType): + msg = f"Expected types.FrameType, but got {type(frame).__name__}" + raise TypeError(msg) + return frame diff --git a/offline_debug/_inner/c_api/_link_frame.py b/offline_debug/_inner/c_api/_link_frame.py new file mode 100644 index 0000000..86ecca4 --- /dev/null +++ b/offline_debug/_inner/c_api/_link_frame.py @@ -0,0 +1,128 @@ +""" +Link 2 frames together, so that a linked chain of frames will eventually create a traceback. + +The native cpython api does not expose a way to link frames together +""" + +from __future__ import annotations + +import ctypes +import sys +from functools import cache +from typing import TYPE_CHECKING, Any + +from ._create_frame import create_frame + +if TYPE_CHECKING: + import _ctypes + from types import FrameType + + +@cache +def _get_py_incref() -> ctypes._NamedFuncPointer: + """ + Get the Py_IncRef C function configured with ctypes. + + Py_IncRef increases the reference count of an object. + We call this function when we want to "own" an object, + meaning that the object won't be deleted while we are still using it. + + In our case, we use the function to hold the f_back reference to link between frames. + """ + func: ctypes._NamedFuncPointer = ctypes.pythonapi.Py_IncRef + func.argtypes = (ctypes.py_object,) + func.restype = None + + def errcheck[T]( + result: T, + _func: _ctypes.CFuncPtr, + _args: tuple[Any, ...], + ) -> T: # pragma: no cover + if result is not None: + msg = f"Unexpected {result=}, expected None." + raise TypeError(msg) + return result + + func.errcheck = errcheck + return func + + +def link_frame(frame: FrameType, f_back: FrameType) -> None: + """Link a frame to its parent frame using the discovered offset.""" + _f_back_offset = _get_f_back_offset() + if _f_back_offset is None: + msg = "Failed discovering the offset for the f_back property." + raise RuntimeError(msg) + + # In Python, setting f_back means the child frame now owns a reference + # to the parent frame. We must increment the reference count of the + # parent to reflect this. + py_incref = _get_py_incref() + py_incref(f_back) + + # Use ctypes to write the address of the back frame into the discovered offset. + # We find the address of the f_back property, by finding the start of the frame struct, + # Then moving to the f_back offset in the struct. + ptr = ctypes.c_void_p.from_address(id(frame) + _f_back_offset) + ptr.value = id(f_back) + + +@cache +def _get_f_back_offset() -> int | None: + """ + Dynamically discover the memory offset of f_back in PyFrameObject. + + The offset for the f_back can change between python versions and operating systems, + So we find the location of the f_back property dynamically. + The general idea of the algorithm is to create a mock frame, + And scanning its memory until we hit the f_back property field. + We know we hit the f_back property by + setting the value of each slot in the memeory to another frame, and checking if the + f_back property was populated after changing that memory location. + """ + try: + # Compile an empty code object that we can use to create a frame. + code = compile("pass", "", "exec") + + # Create 2 frames that we can attempt linking between to find the f_back property. + frame = create_frame(code=code, frame_globals={}, frame_locals={}) + f_back_frame = create_frame(code=code, frame_globals={}, frame_locals={}) + + ptr_size = ctypes.sizeof(ctypes.c_void_p) + jump_size = ptr_size + + ref_count_size = ptr_size # The amount of references to the object + type_object_size = ptr_size + frame_header_size = ref_count_size + type_object_size + start = frame_header_size # The header does not contain the f_back property + + frame_size = sys.getsizeof(frame) + + # We scan the memory ptr_size bytes forward, so we decrease it from the end. + # Since we want the last memory slot to be scanned as well, + # we add +1 because the for loop is exclusive to the end. + end = frame_size - ptr_size + 1 + + # We start scanning after the PyObject header (refcnt + type). + for offset in range(start, end, jump_size): + candidate_f_back_address = id(frame) + offset + candidate_f_back_ptr = ctypes.c_ssize_t.from_address(candidate_f_back_address) + + # f_back is initially NULL (0) in a newly created frame, + # since no other frames are linked to it. + if candidate_f_back_ptr.value != 0: + continue + + candidate_f_back_ptr.value = id(f_back_frame) + # If reading f_back via Python now returns our target, we found it. + if frame.f_back is f_back_frame: + # Success, but we must restore 0 so we don't mess up refcounts + # when 'frame' is eventually garbage collected. + candidate_f_back_ptr.value = 0 + return offset + # Restore to 0 if this wasn't the correct offset. + candidate_f_back_ptr.value = 0 + + except Exception: # noqa: BLE001 + return None + return None diff --git a/offline_debug/_inner/load_traceback.py b/offline_debug/_inner/load_traceback.py new file mode 100644 index 0000000..4fdf0e3 --- /dev/null +++ b/offline_debug/_inner/load_traceback.py @@ -0,0 +1,129 @@ +"""Load traceback object from a dump file.""" + +import marshal +import pickle +import sys +import types +from io import BytesIO +from pathlib import Path +from types import CodeType +from typing import Never + +from offline_debug._inner.c_api import ( + create_frame, + link_frame, +) +from offline_debug._inner.models import ( + _ExceptionData, + _FrameData, +) + + +def _reconstruct_exc_data(data: _ExceptionData) -> BaseException: + """ + Recursively reconstruct an exception from its serialized data. + + Note on Python Locals: + Python uses two ways to store local variables: + 1. "Slow" locals: A dictionary used for module-level code and class definitions. + 2. "Fast" locals: A fixed-size array used for functions. This is faster than + dictionary lookups because variables are accessed by index. + + During reconstruction, we must explicitly synchronize these because PyFrame_New + does not automatically populate the "fast" locals array from a dictionary. + """ + exc: BaseException = pickle.loads(data.exc_pickle) # noqa: S301 + if not isinstance(exc, BaseException): + msg = f"Expected BaseException, but got {type(exc).__name__}" + raise TypeError(msg) + + reconstructed_frames: list[tuple[types.FrameType, _FrameData]] = [] + for f_data in data.tb_frames: + code: CodeType = marshal.loads(f_data.code) # noqa: S302 + + # In Python 3.11 and 3.12, accessing f_locals on a frame created via + # PyFrame_New for optimized code (functions) causes a segmentation fault + # because the internal 'fast' locals array is not initialized. + # As a workaround, we create a 'non-optimized' version of the code object + # by compiling a dummy string. This ensures the bytecode is safe + # (no LOAD_FAST) while preserving metadata like name and filename. + if sys.version_info < (3, 13): + # A simple module-level code object never has fast locals. + # Since the source is empty, no optimized locals will be created. + # Instead, python will go to the unoptimized dictionary we set under frame_locals later. + unoptimized_code = compile("", code.co_filename, "exec") + code = unoptimized_code.replace( + co_name=code.co_name, + co_firstlineno=code.co_firstlineno, + co_qualname=code.co_qualname, + ) + + # PyFrame_New returns a new reference to a PyFrameObject. + if f_data.module_name: + f_data.globals["__name__"] = f_data.module_name + + frame: types.FrameType = create_frame( + code=code, frame_globals=f_data.globals, frame_locals=f_data.locals + ) + + if reconstructed_frames: + # link the frame back to the previously constructed frame. + link_frame(frame, reconstructed_frames[-1][0]) + + reconstructed_frames.append((frame, f_data)) + + tb_next: types.TracebackType | None = None + for frame, f_data in reversed(reconstructed_frames): + tb = types.TracebackType( + tb_next=tb_next, + tb_frame=frame, + tb_lasti=f_data.lasti, + tb_lineno=f_data.lineno, + ) + tb_next = tb + + exc = exc.with_traceback(tb_next) + + if data.cause: + exc.__cause__ = _reconstruct_exc_data(data.cause) + if data.context: + exc.__context__ = _reconstruct_exc_data(data.context) + + return exc + + +def load_traceback(file: Path | BytesIO) -> Never: + """Load an exception and its traceback from a file and raise it.""" + if isinstance(file, Path): + with file.open("rb") as f: + data = pickle.load(f) # noqa: S301 + else: + data = pickle.load(file) # noqa: S301 + + if not isinstance(data, _ExceptionData): + msg = f"Expected _ExceptionData, but got {type(data).__name__}" + raise TypeError(msg) + + exc = _reconstruct_exc_data(data) + + current_frames: list[types.FrameType] = [] + curr: types.FrameType | None = sys._getframe(1) # noqa: SLF001 + while curr: + current_frames.append(curr) + curr = curr.f_back + + if exc.__traceback__ and current_frames: + reconstructed_outer = exc.__traceback__.tb_frame + link_frame(reconstructed_outer, current_frames[0]) + + tb_chain: types.TracebackType | None = exc.__traceback__ + for frame in current_frames: + tb_chain = types.TracebackType( + tb_next=tb_chain, + tb_frame=frame, + tb_lasti=frame.f_lasti, + tb_lineno=frame.f_lineno, + ) + + exc = exc.with_traceback(tb_chain) + raise exc diff --git a/offline_debug/_inner/models.py b/offline_debug/_inner/models.py new file mode 100644 index 0000000..8999761 --- /dev/null +++ b/offline_debug/_inner/models.py @@ -0,0 +1,29 @@ +"""Functions for serializing and reconstructing exceptions with their tracebacks.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + + +@dataclass +class _FrameData: + """Serialized data for a single stack frame.""" + + code: bytes + globals: dict[str, Any] + locals: dict[str, Any] + lasti: int + lineno: int + stack_depth: int + module_name: str | None = None + + +@dataclass +class _ExceptionData: + """Serialized data for an exception and its traceback.""" + + exc_pickle: bytes + tb_frames: list[_FrameData] + cause: _ExceptionData | None = None + context: _ExceptionData | None = None diff --git a/offline_debug/_inner/save_traceback.py b/offline_debug/_inner/save_traceback.py new file mode 100644 index 0000000..f623fab --- /dev/null +++ b/offline_debug/_inner/save_traceback.py @@ -0,0 +1,98 @@ +"""Save traceback to a file.""" + +import marshal +import pickle +import types +from io import BytesIO +from pathlib import Path + +from offline_debug._inner.models import _ExceptionData, _FrameData + +# Internal attributes that are either unpicklable or redundant in a new process. +# We exclude these specifically because they are automatically recreated +# when the new frame is initialized or when the module is imported. +_INTERNAL_ATTRIBUTES_TO_SKIP = ("__builtins__", "__doc__", "__loader__", "__package__", "__spec__") + + +def _get_stack_depth(frame: types.FrameType) -> int: + """Calculate the depth of the current stack frame.""" + depth = 0 + curr: types.FrameType | None = frame + while curr: + depth += 1 + curr = curr.f_back + return depth + + +def _filter_dict(d: dict) -> dict: + """Filter dictionary to include only picklable items.""" + result = {} + for k, v in d.items(): + if k in _INTERNAL_ATTRIBUTES_TO_SKIP: + continue + try: + # We must verify if the value is picklable because many globals + # (like open file handles, database connections, or modules) + # cannot be saved to disk. + pickle.dumps(v) + result[k] = v + except Exception: # noqa: BLE001 + result[k] = f"" + return result + + +def _serialize_exc_data(exc: BaseException) -> _ExceptionData: + """Recursively serialize exception data into dataclasses.""" + tb_frames: list[_FrameData] = [] + curr_tb = exc.__traceback__ + while curr_tb: + f = curr_tb.tb_frame + + # Try to get the "real" module name. If the module was run as a script, + # __name__ will be "__main__", but __spec__.name might contain the + # actual module path if run via `python -m`. + mod_name = f.f_globals.get("__name__") + if mod_name == "__main__": + spec = f.f_globals.get("__spec__") + if spec and hasattr(spec, "name"): + mod_name = spec.name + + tb_frames.append( + _FrameData( + code=marshal.dumps(f.f_code), + globals=_filter_dict(f.f_globals), + locals=_filter_dict(f.f_locals), + lasti=curr_tb.tb_lasti, + lineno=curr_tb.tb_lineno, + stack_depth=_get_stack_depth(f), + module_name=mod_name, + ) + ) + curr_tb = curr_tb.tb_next + + try: + exc_pickle = pickle.dumps(exc) + except Exception: # noqa: BLE001 + exc_pickle = pickle.dumps( + RuntimeError(f"Unpicklable exception {type(exc).__name__}: {exc!s}") + ) + + return _ExceptionData( + exc_pickle=exc_pickle, + tb_frames=tb_frames, + cause=_serialize_exc_data(exc.__cause__) if exc.__cause__ else None, + context=_serialize_exc_data(exc.__context__) if exc.__context__ else None, + ) + + +def save_traceback(exc: BaseException, file: Path | BytesIO) -> None: + """Serialize an exception and its traceback to a file.""" + data = _serialize_exc_data(exc) + if isinstance(file, Path): + with file.open("wb") as f: + pickle.dump(data, f) + elif isinstance(file, BytesIO): + pickle.dump(data, file) + else: + msg = f"Unexpected type for file {type(file).__name__}" + raise TypeError(msg) diff --git a/offline_debug/serializer.py b/offline_debug/serializer.py deleted file mode 100644 index 7f0ad23..0000000 --- a/offline_debug/serializer.py +++ /dev/null @@ -1,297 +0,0 @@ -"""Functions for serializing and reconstructing exceptions with their tracebacks.""" - -from __future__ import annotations - -import ctypes -import marshal -import pickle -import sys -import types -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Never - -# Define C API for frame creation -_py_frame_new = ctypes.pythonapi.PyFrame_New -_py_frame_new.argtypes = ( - ctypes.c_void_p, # PyThreadState *tstate - ctypes.py_object, # PyCodeObject *code - ctypes.py_object, # PyObject *globals - ctypes.py_object, # PyObject *locals -) -_py_frame_new.restype = ctypes.py_object - -_py_thread_state_get = ctypes.pythonapi.PyThreadState_Get -_py_thread_state_get.restype = ctypes.c_void_p - -_py_incref = ctypes.pythonapi.Py_IncRef -_py_incref.argtypes = (ctypes.py_object,) - -_py_decref = ctypes.pythonapi.Py_DecRef -_py_decref.argtypes = (ctypes.py_object,) - - -def _get_f_back_offset() -> int | None: - """Dynamically discover the memory offset of f_back in PyFrameObject.""" - try: - tstate = _py_thread_state_get() - # Compile a dummy code object that we can use to create a frame. - code = compile("pass", "", "exec") - # Create a new, detached frame object using the C API. - frame = _py_frame_new(tstate, code, {}, {}) - if not isinstance(frame, types.FrameType): - return None - - # We need a target frame object to point to. - target = sys._getframe() # noqa: SLF001 - target_addr = id(target) - - # We scan the frame object's memory for the f_back pointer. - # We cap the scan at the object's actual size to avoid out-of-bounds reads. - limit = sys.getsizeof(frame) - ptr_size = ctypes.sizeof(ctypes.c_void_p) - - # We start scanning after the PyObject header (refcnt + type). - for offset in range(2 * ptr_size, limit - ptr_size + 1, ptr_size): - try: - # We use c_ssize_t to read the raw value at the offset. - current_val = ctypes.c_ssize_t.from_address(id(frame) + offset).value - # f_back is initially NULL (0) in a newly created frame. - if current_val == 0: - ctypes.c_ssize_t.from_address(id(frame) + offset).value = target_addr - # If reading f_back via Python now returns our target, we found it. - if frame.f_back is target: - # Success, but we must restore 0 so we don't mess up refcounts - # when 'frame' is eventually garbage collected. - ctypes.c_ssize_t.from_address(id(frame) + offset).value = 0 - return offset - # Restore to 0 if this wasn't the correct offset. - ctypes.c_ssize_t.from_address(id(frame) + offset).value = 0 - except (AttributeError, ValueError, TypeError, RuntimeError): - continue - except Exception: # noqa: BLE001 - return None - return None - - -_F_BACK_OFFSET = _get_f_back_offset() - - -def _link_frame(frame: types.FrameType, back: types.FrameType) -> None: - """Link a frame to its parent frame using the discovered offset.""" - if _F_BACK_OFFSET is None: - return - - # In Python, setting f_back means the child frame now owns a reference - # to the parent frame. We must increment the reference count of the - # parent to reflect this. - _py_incref(back) - - # Use ctypes to write the address of the back frame into the discovered offset. - ptr = ctypes.c_void_p.from_address(id(frame) + _F_BACK_OFFSET) - ptr.value = id(back) - - -# Internal attributes that are either unpicklable or redundant in a new process. -# We exclude these specifically because they are automatically recreated -# when the new frame is initialized or when the module is imported. -_INTERNAL_ATTRIBUTES_TO_SKIP = ("__builtins__", "__doc__", "__loader__", "__package__", "__spec__") - - -@dataclass -class _FrameData: - """Serialized data for a single stack frame.""" - - code: bytes - globals: dict[str, Any] - locals: dict[str, Any] - lasti: int - lineno: int - stack_depth: int - - -@dataclass -class _ExceptionData: - """Serialized data for an exception and its traceback.""" - - exc_pickle: bytes - tb_frames: list[_FrameData] - cause: _ExceptionData | None = None - context: _ExceptionData | None = None - - -def _get_stack_depth(frame: types.FrameType) -> int: - """Calculate the depth of the current stack frame.""" - depth = 0 - curr: types.FrameType | None = frame - while curr: - depth += 1 - curr = curr.f_back - return depth - - -def _filter_dict(d: dict) -> dict: - """Filter dictionary to include only picklable items.""" - result = {} - for k, v in d.items(): - if k in _INTERNAL_ATTRIBUTES_TO_SKIP: - continue - try: - # We must verify if the value is picklable because many globals - # (like open file handles, database connections, or modules) - # cannot be saved to disk. - pickle.dumps(v) - result[k] = v - except Exception: # noqa: BLE001 - result[k] = f"" - return result - - -def _serialize_exc_data(exc: BaseException) -> _ExceptionData: - """Recursively serialize exception data into dataclasses.""" - tb_frames: list[_FrameData] = [] - curr_tb = exc.__traceback__ - while curr_tb: - f = curr_tb.tb_frame - tb_frames.append( - _FrameData( - code=marshal.dumps(f.f_code), - globals=_filter_dict(f.f_globals), - locals=_filter_dict(f.f_locals), - lasti=curr_tb.tb_lasti, - lineno=curr_tb.tb_lineno, - stack_depth=_get_stack_depth(f), - ) - ) - curr_tb = curr_tb.tb_next - - try: - exc_pickle = pickle.dumps(exc) - except Exception: # noqa: BLE001 - exc_pickle = pickle.dumps( - RuntimeError(f"Unpicklable exception {type(exc).__name__}: {exc!s}") - ) - - return _ExceptionData( - exc_pickle=exc_pickle, - tb_frames=tb_frames, - cause=_serialize_exc_data(exc.__cause__) if exc.__cause__ else None, - context=_serialize_exc_data(exc.__context__) if exc.__context__ else None, - ) - - -def save_traceback(exc: BaseException, file_path: str | Path) -> None: - """Serialize an exception and its traceback to a file.""" - data = _serialize_exc_data(exc) - with Path(file_path).open("wb") as f: - pickle.dump(data, f) - - -def _reconstruct_exc_data(data: _ExceptionData) -> BaseException: - """ - Recursively reconstruct an exception from its serialized data. - - Note on Python Locals: - Python uses two ways to store local variables: - 1. "Slow" locals: A dictionary used for module-level code and class definitions. - 2. "Fast" locals: A fixed-size array used for functions. This is faster than - dictionary lookups because variables are accessed by index. - - During reconstruction, we must explicitly synchronize these because PyFrame_New - does not automatically populate the "fast" locals array from a dictionary. - """ - exc = pickle.loads(data.exc_pickle) # noqa: S301 - if not isinstance(exc, BaseException): - msg = f"Expected BaseException, but got {type(exc).__name__}" - raise TypeError(msg) - - tstate = _py_thread_state_get() - - reconstructed_frames: list[tuple[types.FrameType, _FrameData]] = [] - prev_frame: types.FrameType | None = None - for f_data in data.tb_frames: - code = marshal.loads(f_data.code) # noqa: S302 - - # In Python 3.11 and 3.12, accessing f_locals on a frame created via - # PyFrame_New for optimized code (functions) causes a segmentation fault - # because the internal 'fast' locals array is not initialized. - # As a workaround, we create a 'non-optimized' version of the code object - # by compiling a dummy string. This ensures the bytecode is safe - # (no LOAD_FAST) while preserving metadata like name and filename. - if sys.version_info < (3, 13): - # A simple module-level code object never has fast locals. - dummy_code = compile("", code.co_filename, "exec") - code = dummy_code.replace( - co_name=code.co_name, - co_firstlineno=code.co_firstlineno, - co_qualname=code.co_qualname, - ) - - # PyFrame_New returns a new reference to a PyFrameObject. - frame = _py_frame_new(tstate, code, f_data.globals, f_data.locals) - if not isinstance(frame, types.FrameType): - msg = f"Expected types.FrameType, but got {type(frame).__name__}" - raise TypeError(msg) - - # In 3.13+, PEP 667 allows safe write-through access to locals. - if sys.version_info >= (3, 13) and f_data.locals: - frame.f_locals.update(f_data.locals) - - if prev_frame: - _link_frame(frame, prev_frame) - - reconstructed_frames.append((frame, f_data)) - prev_frame = frame - - tb_next: types.TracebackType | None = None - for frame, f_data in reversed(reconstructed_frames): - tb = types.TracebackType( - tb_next=tb_next, - tb_frame=frame, - tb_lasti=f_data.lasti, - tb_lineno=f_data.lineno, - ) - tb_next = tb - - exc = exc.with_traceback(tb_next) - - if data.cause: - exc.__cause__ = _reconstruct_exc_data(data.cause) - if data.context: - exc.__context__ = _reconstruct_exc_data(data.context) - - return exc - - -def load_traceback(file_path: str | Path) -> Never: - """Load an exception and its traceback from a file and raise it.""" - with Path(file_path).open("rb") as f: - data = pickle.load(f) # noqa: S301 - - if not isinstance(data, _ExceptionData): - msg = f"Expected _ExceptionData, but got {type(data).__name__}" - raise TypeError(msg) - - exc = _reconstruct_exc_data(data) - - current_frames: list[types.FrameType] = [] - curr: types.FrameType | None = sys._getframe(1) # noqa: SLF001 - while curr: - current_frames.append(curr) - curr = curr.f_back - - if exc.__traceback__ and current_frames: - reconstructed_outer = exc.__traceback__.tb_frame - _link_frame(reconstructed_outer, current_frames[0]) - - tb_chain: types.TracebackType | None = exc.__traceback__ - for frame in current_frames: - tb_chain = types.TracebackType( - tb_next=tb_chain, - tb_frame=frame, - tb_lasti=frame.f_lasti, - tb_lineno=frame.f_lineno, - ) - - exc = exc.with_traceback(tb_chain) - raise exc diff --git a/pyproject.toml b/pyproject.toml index d9a67e1..d0c6bad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "offline-debug" -version = "0.1.0" +version = "0.1.1" description = "Debug exceptions offline by saving them to a dump and raising them at a later point." readme = "README.md" authors = [ @@ -60,6 +60,7 @@ dev = [ [tool.coverage.report] exclude_lines = [ "pragma: no cover", + "if TYPE_CHECKING:", "if sys.version_info < \\(3, 13\\).*", "if sys.version_info >= \\(3, 13\\).*", ] diff --git a/tests/_inner/__init__.py b/tests/_inner/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/_inner/c_api/__init__.py b/tests/_inner/c_api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_f_back_discovery.py b/tests/_inner/c_api/test__link_frame.py similarity index 52% rename from tests/test_f_back_discovery.py rename to tests/_inner/c_api/test__link_frame.py index e9232ae..a985ffd 100644 --- a/tests/test_f_back_discovery.py +++ b/tests/_inner/c_api/test__link_frame.py @@ -1,9 +1,44 @@ -"""Tests for the dynamic f_back offset discovery logic.""" +"""Tests for the link_frame and f_back discovery logic.""" + +from __future__ import annotations import ctypes -from unittest.mock import patch +import sys +from typing import TYPE_CHECKING +from unittest.mock import MagicMock, patch + +import pytest + +from offline_debug._inner.c_api._link_frame import _get_f_back_offset + +if TYPE_CHECKING: + from collections.abc import Iterator + + +@pytest.fixture(autouse=True) +def clear_f_back_offset_cache() -> Iterator[None]: + _get_f_back_offset.cache_clear() + yield + _get_f_back_offset.cache_clear() + + +def test_get_f_back_offset_logic() -> None: + """Test the dynamic f_back offset discovery logic directly.""" + offset = _get_f_back_offset() + # It should either find an offset or be None (if platform is weird) + # But on standard CPython it should find something. + assert offset is None or (offset > 0 and offset % 8 == 0) -from offline_debug.serializer import _get_f_back_offset + +def test_link_frame_no_offset(monkeypatch) -> None: + """Test that link_frame raises an exception if the f back offset wasn't found.""" + import offline_debug._inner.c_api._link_frame as _link_frame_module + + monkeypatch.setattr(_link_frame_module, "_get_f_back_offset", lambda: None) + + f = sys._getframe() + with pytest.raises(RuntimeError, match="Failed discovering"): + _link_frame_module.link_frame(f, f) def test_get_f_back_offset_success() -> None: @@ -18,16 +53,21 @@ def test_get_f_back_offset_success() -> None: def test_get_f_back_offset_not_a_frame() -> None: """Test when PyFrame_New returns something that is not a FrameType.""" - with patch("offline_debug.serializer._py_frame_new", return_value="not a frame"): + import offline_debug._inner.c_api._create_frame as _create_frame_module + + with patch.object( + _create_frame_module, "_get_py_frame_new", return_value=lambda *_: "not a frame" + ): offset = _get_f_back_offset() assert offset is None def test_get_f_back_offset_exception_in_try() -> None: """Test when an exception occurs early in the discovery process.""" - with patch( - "offline_debug.serializer._py_thread_state_get", side_effect=RuntimeError("thread error") - ): + import offline_debug._inner.c_api._create_frame as _create_frame_module + + mock_func = MagicMock(side_effect=RuntimeError("thread error")) + with patch.object(_create_frame_module, "_get_py_thread_state_get", return_value=mock_func): offset = _get_f_back_offset() assert offset is None @@ -46,6 +86,9 @@ def __init__(self, val: int) -> None: def test_get_f_back_offset_wrong_offset_restoration() -> None: """Test that it restores 0 if the offset was wrong.""" + import offline_debug._inner.c_api._create_frame as _create_frame_module + import offline_debug._inner.c_api._link_frame as _link_frame_module + tstate = ctypes.pythonapi.PyThreadState_Get() code = compile("pass", "", "exec") frame = ctypes.pythonapi.PyFrame_New(tstate, code, {}, {}) @@ -55,8 +98,8 @@ def test_get_f_back_offset_wrong_offset_restoration() -> None: # Let's try this: with ( - patch("offline_debug.serializer._py_frame_new", return_value=frame), - patch("offline_debug.serializer.range", return_value=[ptr_size * 10]), + patch.object(_create_frame_module, "_get_py_frame_new", return_value=lambda *_: frame), + patch.object(_link_frame_module, "range", return_value=[ptr_size * 10]), ): # Offset 80 # Ensure offset 80 is 0 ctypes.c_ssize_t.from_address(id(frame) + ptr_size * 10).value = 0 diff --git a/tests/_inner/test_load_traceback.py b/tests/_inner/test_load_traceback.py new file mode 100644 index 0000000..0826fc5 --- /dev/null +++ b/tests/_inner/test_load_traceback.py @@ -0,0 +1,100 @@ +"""Tests for the load_traceback module.""" + +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +import pytest + +from offline_debug import load_traceback + +if TYPE_CHECKING: + import types + + +def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: + """Extract all frames from a traceback.""" + frames = [] + curr = tb + while curr: + frames.append(curr.tb_frame) + curr = curr.tb_next + return frames + + +def test_typing_never() -> None: + """Test that load_traceback is correctly annotated with Never.""" + import typing + + from offline_debug import load_traceback + + load_traceback_annotations = typing.get_type_hints(load_traceback) + assert load_traceback_annotations["return"] is typing.Never + + +def test_load_invalid_object(tmp_path: Path) -> None: + """Test that load_traceback raises TypeError when loading an invalid object.""" + import pickle + + dump_file = tmp_path / "invalid.dump" + with dump_file.open("wb") as f: + pickle.dump("not an ExceptionData object", f) + + with pytest.raises(TypeError, match="Expected _ExceptionData, but got str"): + load_traceback(dump_file) + + +def test_load_non_existent_file() -> None: + """Test that load_traceback raises FileNotFoundError when the file does not exist.""" + with pytest.raises(FileNotFoundError): + load_traceback(Path("non_existent_file.dump")) + + +def test_reconstruct_invalid_exception_type() -> None: + """Test that _reconstruct_exc_data raises TypeError when the pickled exception is invalid.""" + import pickle + + from offline_debug._inner.load_traceback import _reconstruct_exc_data + from offline_debug._inner.models import _ExceptionData + + data = _ExceptionData( + exc_pickle=pickle.dumps("not an exception"), + tb_frames=[], + ) + + with pytest.raises(TypeError, match="Expected BaseException, but got str"): + _reconstruct_exc_data(data) + + +def test_reconstruct_invalid_frame_type(monkeypatch) -> None: + """Test that _reconstruct_exc_data raises TypeError when frame creation fails.""" + import offline_debug._inner.c_api._create_frame as _create_frame_module + from offline_debug._inner.load_traceback import _reconstruct_exc_data + from offline_debug._inner.models import _ExceptionData, _FrameData + + # Mock _get_py_frame_new to return a function that returns something that is not a FrameType + monkeypatch.setattr(_create_frame_module, "_get_py_frame_new", lambda: lambda *_: "not a frame") + + import marshal + import pickle + + def dummy() -> None: + pass + + data = _ExceptionData( + exc_pickle=pickle.dumps(ValueError("test")), + tb_frames=[ + _FrameData( + code=marshal.dumps(dummy.__code__), + globals={}, + locals={}, + lasti=0, + lineno=0, + stack_depth=0, + ) + ], + ) + + with pytest.raises(TypeError, match=r"Expected types.FrameType, but got str"): + _reconstruct_exc_data(data) diff --git a/tests/_inner/test_save_traceback.py b/tests/_inner/test_save_traceback.py new file mode 100644 index 0000000..2244729 --- /dev/null +++ b/tests/_inner/test_save_traceback.py @@ -0,0 +1,164 @@ +"""Tests for the save_traceback module.""" + +from __future__ import annotations + +import types +from typing import TYPE_CHECKING, Never + +import pytest + +from offline_debug import load_traceback, save_traceback + +if TYPE_CHECKING: + from collections.abc import Callable + from pathlib import Path + + +def func_to_module_string(func: Callable) -> str: + """Convert a function's body into a module string, removing indentation.""" + import inspect + import textwrap + + source = inspect.getsource(func) + # Get the body of the function (skip the def line) + lines = source.splitlines() + body = "\n".join(lines[1:]) + return textwrap.dedent(body) + + +def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: + """Extract all frames from a traceback.""" + frames = [] + curr = tb + while curr: + frames.append(curr.tb_frame) + curr = curr.tb_next + return frames + + +def test_unpicklable_locals_verification(tmp_path: Path) -> None: + """Test that unpicklable local variables are handled gracefully.""" + dump_file = tmp_path / "unpicklable.dump" + + class Unpicklable: + def __reduce__(self) -> Never: + msg = "Cannot pickle me" + raise TypeError(msg) + + def __repr__(self) -> str: + return "" + + def fail_with_unpicklable() -> Never: + _obj = Unpicklable() + # Use locals() to force inclusion in the locals dictionary. + _ = locals()["_obj"] + msg = "Error with unpicklable" + raise ValueError(msg) + + try: + fail_with_unpicklable() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + with pytest.raises(ValueError, match="Error with unpicklable") as exc_info: + load_traceback(dump_file) + + frames = get_frames(exc_info.tb) + + frame_names = [f.f_code.co_name for f in frames] + assert "fail_with_unpicklable" in frame_names + f = next(f for f in frames if f.f_code.co_name == "fail_with_unpicklable") + # Verify that our filtering logic caught the unpicklable item. + assert any(" None: + """Test that global variables are preserved in the stack.""" + dump_file = tmp_path / "globals.dump" + + def fail_with_globals() -> None: + global GLOBAL_TEST_VAL # noqa: PLW0602 + if GLOBAL_TEST_VAL == "I am global": + msg = "Global test" + raise ValueError(msg) + + try: + fail_with_globals() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + with pytest.raises(ValueError, match="Global test") as exc_info: + load_traceback(dump_file) + + frames = get_frames(exc_info.tb) + + frame_names = [f.f_code.co_name for f in frames] + assert "fail_with_globals" in frame_names + f = next(f for f in frames if f.f_code.co_name == "fail_with_globals") + assert f.f_globals["GLOBAL_TEST_VAL"] == "I am global" + + +def test_globals_changing_between_frames(tmp_path: Path) -> None: + """Test that globals are captured per-frame for different modules.""" + dump_file = tmp_path / "globals_changing.dump" + + # Define the second module's content inside a function + def module2_content() -> None: + GLOBAL_VAR = "initial" # noqa: N806, F841 + + def level_2() -> Never: + global GLOBAL_VAR + GLOBAL_VAR = "changed" + msg = "Error at level 2" + raise ValueError(msg) + + mod2_code = func_to_module_string(module2_content) + mod2 = types.ModuleType("mod2") + exec(mod2_code, mod2.__dict__) # noqa: S102 + + def level_1() -> None: + try: + mod2.level_2() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + level_1() + + with pytest.raises(ValueError, match="Error at level 2") as exc_info: + load_traceback(dump_file) + + frames = get_frames(exc_info.tb) + + l1_f = next(f for f in frames if f.f_code.co_name == "level_1") + l2_f = next(f for f in frames if f.f_code.co_name == "level_2") + + # mod2.level_2 changed its OWN GLOBAL_VAR. + # level_1's globals (this module) should NOT have GLOBAL_VAR (or it should be different) + assert l2_f.f_globals["GLOBAL_VAR"] == "changed" + assert "GLOBAL_VAR" not in l1_f.f_globals or l1_f.f_globals["GLOBAL_VAR"] == "initial" + + +def test_unpicklable_exception_coverage(tmp_path: Path) -> None: + """Test that unpicklable exceptions are handled by falling back to RuntimeError.""" + dump_file = tmp_path / "unpicklable_exc.dump" + + class UnpicklableError(Exception): + def __reduce__(self) -> Never: + msg = "Cannot pickle me" + raise TypeError(msg) + + def raise_unpicklable() -> Never: + msg = "Unpicklable" + raise UnpicklableError(msg) + + try: + raise_unpicklable() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + with pytest.raises(RuntimeError, match="Unpicklable exception UnpicklableError: Unpicklable"): + load_traceback(dump_file) diff --git a/tests/manual_tests/__init__.py b/tests/manual_tests/__init__.py new file mode 100644 index 0000000..ffd5734 --- /dev/null +++ b/tests/manual_tests/__init__.py @@ -0,0 +1 @@ +"""Tests meant to be run manually by a use to see how the exceptions look in a debugger.""" diff --git a/tests/debug.py b/tests/manual_tests/debug.py similarity index 93% rename from tests/debug.py rename to tests/manual_tests/debug.py index 7ec9407..f0930a9 100644 --- a/tests/debug.py +++ b/tests/manual_tests/debug.py @@ -9,7 +9,7 @@ from rich import traceback from rich.console import Console -from offline_debug.serializer import load_traceback, save_traceback +from offline_debug import load_traceback, save_traceback global_variable = 1 diff --git a/tests/manual_tests/global_debug.py b/tests/manual_tests/global_debug.py new file mode 100644 index 0000000..bdebc69 --- /dev/null +++ b/tests/manual_tests/global_debug.py @@ -0,0 +1,16 @@ +"""Intended for manual use, run the test in debug mode and view the exceptions.""" + +import tempfile +from pathlib import Path + +from offline_debug import load_traceback, save_traceback + +if __name__ == "__main__": + try: + raise ValueError("Trigger") # noqa: TRY301 + except ValueError as e: + with tempfile.TemporaryDirectory() as tmpdir: + dump_file = Path(tmpdir) / "exception.dump" + save_traceback(e, dump_file) + + load_traceback(dump_file) diff --git a/tests/test_bytesio.py b/tests/test_bytesio.py new file mode 100644 index 0000000..9c78e33 --- /dev/null +++ b/tests/test_bytesio.py @@ -0,0 +1,70 @@ +"""Tests for BytesIO support in save_traceback and load_traceback.""" + +from __future__ import annotations + +from io import BytesIO +from typing import TYPE_CHECKING, Never + +import pytest + +from offline_debug import load_traceback, save_traceback + +if TYPE_CHECKING: + import types + + +def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: + """Extract all frames from a traceback.""" + frames = [] + curr = tb + while curr: + frames.append(curr.tb_frame) + curr = curr.tb_next + return frames + + +def test_bytesio_roundtrip() -> None: + """Test that saving to and loading from a BytesIO object works.""" + buffer = BytesIO() + expected_val = 42 + + def func_with_locals() -> Never: + _var_a = "hello" + _var_b = expected_val + msg = "BytesIO error" + raise ValueError(msg) + + try: + func_with_locals() + except ValueError as e: + save_traceback(e, buffer) + + # Reset buffer for reading + buffer.seek(0) + + with pytest.raises(ValueError, match="BytesIO error") as exc_info: + load_traceback(buffer) + + frames = get_frames(exc_info.tb) + f = next(f for f in frames if f.f_code.co_name == "func_with_locals") + + assert f.f_locals.get("_var_a") == "hello" + assert f.f_locals.get("_var_b") == expected_val + + +def test_save_traceback_invalid_type() -> None: + """Test that save_traceback raises TypeError for invalid file type.""" + with pytest.raises(TypeError, match="Unexpected type for file str"): + save_traceback(ValueError("test"), "not a path or bytesio") # type: ignore[arg-type] # ty:ignore[invalid-argument-type] + + +def test_load_bytesio_invalid_object() -> None: + """Test that load_traceback raises TypeError for invalid objects in BytesIO.""" + import pickle + + buffer = BytesIO() + pickle.dump("not an ExceptionData object", buffer) + buffer.seek(0) + + with pytest.raises(TypeError, match="Expected _ExceptionData, but got str"): + load_traceback(buffer) diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..db37807 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,224 @@ +"""Integration tests for offline-debug.""" + +from __future__ import annotations + +import sys +import types +from typing import TYPE_CHECKING, Never + +import pytest + +from offline_debug import load_traceback, save_traceback + +if TYPE_CHECKING: + from pathlib import Path + + +def get_stack_depth(item: types.FrameType | types.TracebackType | None) -> int: + """Calculate the depth of the given stack frame or traceback chain.""" + depth = 0 + curr = item + while curr: + depth += 1 + curr = curr.f_back if isinstance(curr, types.FrameType) else curr.tb_next + return depth + + +def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: + """Extract all frames from a traceback.""" + frames = [] + curr = tb + while curr: + frames.append(curr.tb_frame) + curr = curr.tb_next + return frames + + +def test_stack_depth_preservation(tmp_path: Path) -> None: + """Test that the stack depth is preserved during serialization.""" + dump_file = tmp_path / "depth.dump" + + original_depths = [] + + def level_3() -> Never: + f = sys._getframe() + original_depths.append(get_stack_depth(f)) + msg = "Depth error" + raise ValueError(msg) + + def level_2() -> None: + f = sys._getframe() + original_depths.append(get_stack_depth(f)) + level_3() + + def level_1() -> None: + f = sys._getframe() + original_depths.append(get_stack_depth(f)) + try: + level_2() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + level_1() + + with pytest.raises(ValueError, match="Depth error") as exc_info: + load_traceback(dump_file) + + # Reconstruct the depth from the traceback chain + tb = exc_info.tb + # Actually, load_traceback adds current frames. + + frames = get_frames(tb) + l1_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_1") + l2_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_2") + l3_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_3") + + # In a traceback, the list is TOP to BOTTOM (outer to inner) + # So level_1 should be before level_2 + assert l1_idx < l2_idx + assert l2_idx < l3_idx + + +def test_simple_exception_full_stack(tmp_path: Path) -> None: + """Test serialization of a simple exception with a full stack.""" + dump_file = tmp_path / "simple.dump" + + def inner_raise() -> Never: + _x = 10 + _y = "hello" + msg = "Simple error" + raise ValueError(msg) + + def middle_step() -> None: + inner_raise() + + def capture_it() -> None: + try: + middle_step() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + capture_it() + assert dump_file.exists() + + # Now we call load_traceback from another stack + def second_stack_caller() -> None: + load_traceback(dump_file) + + with pytest.raises(ValueError, match="Simple error") as exc_info: + second_stack_caller() + + frames = get_frames(exc_info.tb) + + frame_names = [f.f_code.co_name for f in frames] + assert "inner_raise" in frame_names + assert "middle_step" in frame_names + assert "second_stack_caller" in frame_names + + inner_idx = frame_names.index("inner_raise") + middle_idx = frame_names.index("middle_step") + second_idx = frame_names.index("second_stack_caller") + + # In a traceback, the list is TOP to BOTTOM (outer to inner) + # second_stack_caller -> capture_it -> middle_step -> inner_raise + assert second_idx < middle_idx + assert middle_idx < inner_idx + + +def test_chained_exceptions_stack(tmp_path: Path) -> None: + """Test serialization of chained exceptions.""" + dump_file = tmp_path / "chained.dump" + + def fail_inner() -> Never: + msg = "Inner key error" + raise KeyError(msg) + + def fail_outer() -> None: + try: + fail_inner() + except KeyError as e: + msg = "Outer runtime error" + raise RuntimeError(msg) from e + + try: + fail_outer() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + with pytest.raises(RuntimeError, match="Outer runtime error") as exc_info: + load_traceback(dump_file) + + reconstructed_exc = exc_info.value + assert isinstance(reconstructed_exc.__cause__, KeyError) + + outer_frames = get_frames(reconstructed_exc.__traceback__) + outer_names = [f.f_code.co_name for f in outer_frames] + assert "fail_outer" in outer_names + + inner_exc = reconstructed_exc.__cause__ + assert isinstance(inner_exc, KeyError) + inner_frames = get_frames(inner_exc.__traceback__) + inner_names = [f.f_code.co_name for f in inner_frames] + assert "fail_inner" in inner_names + + +def test_reconstructed_frames_have_f_back(tmp_path: Path) -> None: + """ + Test that reconstructed frames have their f_back pointers correctly linked. + + This test is currently expected to FAIL because f_back linking was removed + to avoid segmentation faults, but it's required for full fidelity. + """ + dump_file = tmp_path / "f_back_fidelity.dump" + + def level_2() -> Never: + msg = "Fidelity error" + raise ValueError(msg) + + def level_1() -> None: + try: + level_2() + except Exception as e: # noqa: BLE001 + save_traceback(e, dump_file) + + level_1() + + with pytest.raises(ValueError, match="Fidelity error") as exc_info: + load_traceback(dump_file) + + frames = get_frames(exc_info.tb) + # Traceback frames are ordered TOP to BOTTOM (outer to inner) + # We want to check that level_2's frame points back to level_1 + l2_f = next(f for f in frames if f.f_code.co_name == "level_2") + l1_f = next(f for f in frames if f.f_code.co_name == "level_1") + + assert l2_f.f_back is not None, "level_2.f_back should not be None" + assert l2_f.f_back is l1_f, "level_2.f_back should point to level_1" + + +def test_locals_visibility_in_reconstructed_frames(tmp_path: Path) -> None: + """Regression test: verify local variables are visible in reconstructed frames.""" + dump_file = tmp_path / "locals_visibility.dump" + expected_val = 42 + + def func_with_locals() -> Never: + var_a = "hello" + var_b = expected_val + # Use them to ensure they aren't optimized away in some versions + _ = f"{var_a} {var_b}" + msg = "Locals error" + raise ValueError(msg) + + try: + func_with_locals() + except ValueError as e: + save_traceback(e, dump_file) + + with pytest.raises(ValueError, match="Locals error") as exc_info: + load_traceback(dump_file) + + frames = get_frames(exc_info.tb) + f = next(f for f in frames if f.f_code.co_name == "func_with_locals") + + assert f.f_locals.get("var_a") == "hello" + assert f.f_locals.get("var_b") == expected_val diff --git a/tests/test_module_name.py b/tests/test_module_name.py new file mode 100644 index 0000000..1e4b810 --- /dev/null +++ b/tests/test_module_name.py @@ -0,0 +1,95 @@ +"""Tests for module name preservation in save_traceback and load_traceback.""" + +from __future__ import annotations + +import types +from io import BytesIO + +import pytest + +from offline_debug import load_traceback, save_traceback + + +def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: + """Extract all frames from a traceback.""" + frames = [] + curr = tb + while curr: + frames.append(curr.tb_frame) + curr = curr.tb_next + return frames + + +def test_module_name_preservation_from_spec() -> None: + """Test that the real module name is preserved via __spec__.name if __name__ is __main__.""" + buffer = BytesIO() + real_module_name = "my_package.my_module" + + # We simulate a frame where __name__ is "__main__" but __spec__.name is the real module name + # (standard behavior for `python -m my_package.my_module`) + class MockSpec: + def __init__(self, name: str) -> None: + self.name = name + + # Define a function to be executed in a specific globals context + def fail_func() -> None: + msg = "Module name error" + raise ValueError(msg) + + custom_globals = { + "__name__": "__main__", + "__spec__": MockSpec(real_module_name), + "__builtins__": __builtins__, + } + + # Execute the function with our custom globals + # We use a wrapper to capture the exception + try: + # Create a new function object with the same code but different globals + # This is how we simulate it being part of another module + new_func = types.FunctionType(fail_func.__code__, custom_globals, "fail_func") + new_func() + except ValueError as e: + save_traceback(e, buffer) + + buffer.seek(0) + + with pytest.raises(ValueError, match="Module name error") as exc_info: + load_traceback(buffer) + + frames = get_frames(exc_info.tb) + f = next(f for f in frames if f.f_code.co_name == "fail_func") + + # Verify that __name__ in the reconstructed frame's globals is the real module name + assert f.f_globals.get("__name__") == real_module_name + + +def test_module_name_preservation_normal() -> None: + """Test that __name__ is preserved normally when it's not __main__.""" + buffer = BytesIO() + module_name = "some_other_module" + + def fail_func() -> None: + msg = "Normal name error" + raise ValueError(msg) + + custom_globals = { + "__name__": module_name, + "__builtins__": __builtins__, + } + + try: + new_func = types.FunctionType(fail_func.__code__, custom_globals, "fail_func") + new_func() + except ValueError as e: + save_traceback(e, buffer) + + buffer.seek(0) + + with pytest.raises(ValueError, match="Normal name error") as exc_info: + load_traceback(buffer) + + frames = get_frames(exc_info.tb) + f = next(f for f in frames if f.f_code.co_name == "fail_func") + + assert f.f_globals.get("__name__") == module_name diff --git a/tests/test_serializer.py b/tests/test_serializer.py deleted file mode 100644 index 06f79f2..0000000 --- a/tests/test_serializer.py +++ /dev/null @@ -1,469 +0,0 @@ -"""Tests for the serializer module.""" - -from __future__ import annotations - -import sys -import types -from typing import TYPE_CHECKING, Never - -import pytest - -from offline_debug import load_traceback, save_traceback - -if TYPE_CHECKING: - from collections.abc import Callable - from pathlib import Path - - -def func_to_module_string(func: Callable) -> str: - """Convert a function's body into a module string, removing indentation.""" - import inspect - import textwrap - - source = inspect.getsource(func) - # Get the body of the function (skip the def line) - lines = source.splitlines() - body = "\n".join(lines[1:]) - return textwrap.dedent(body) - - -def get_stack_depth(item: types.FrameType | types.TracebackType | None) -> int: - """Calculate the depth of the given stack frame or traceback chain.""" - depth = 0 - curr = item - while curr: - depth += 1 - curr = curr.f_back if isinstance(curr, types.FrameType) else curr.tb_next - return depth - - -def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: - """Extract all frames from a traceback.""" - frames = [] - curr = tb - while curr: - frames.append(curr.tb_frame) - curr = curr.tb_next - return frames - - -def test_stack_depth_preservation(tmp_path: Path) -> None: - """Test that the stack depth is preserved during serialization.""" - dump_file = tmp_path / "depth.dump" - - original_depths = [] - - def level_3() -> Never: - f = sys._getframe() - original_depths.append(get_stack_depth(f)) - msg = "Depth error" - raise ValueError(msg) - - def level_2() -> None: - f = sys._getframe() - original_depths.append(get_stack_depth(f)) - level_3() - - def level_1() -> None: - f = sys._getframe() - original_depths.append(get_stack_depth(f)) - try: - level_2() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - level_1() - - with pytest.raises(ValueError, match="Depth error") as exc_info: - load_traceback(str(dump_file)) - - # Reconstruct the depth from the traceback chain - tb = exc_info.tb - # Actually, load_traceback adds current frames. - - frames = get_frames(tb) - l1_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_1") - l2_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_2") - l3_idx = next(i for i, f in enumerate(frames) if f.f_code.co_name == "level_3") - - # In a traceback, the list is TOP to BOTTOM (outer to inner) - # So level_1 should be before level_2 - assert l1_idx < l2_idx - assert l2_idx < l3_idx - - -def test_simple_exception_full_stack(tmp_path: Path) -> None: - """Test serialization of a simple exception with a full stack.""" - dump_file = tmp_path / "simple.dump" - - def inner_raise() -> Never: - _x = 10 - _y = "hello" - msg = "Simple error" - raise ValueError(msg) - - def middle_step() -> None: - inner_raise() - - def capture_it() -> None: - try: - middle_step() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - capture_it() - assert dump_file.exists() - - # Now we call load_traceback from another stack - def second_stack_caller() -> None: - load_traceback(str(dump_file)) - - with pytest.raises(ValueError, match="Simple error") as exc_info: - second_stack_caller() - - frames = get_frames(exc_info.tb) - - frame_names = [f.f_code.co_name for f in frames] - assert "inner_raise" in frame_names - assert "middle_step" in frame_names - assert "second_stack_caller" in frame_names - - inner_idx = frame_names.index("inner_raise") - middle_idx = frame_names.index("middle_step") - second_idx = frame_names.index("second_stack_caller") - - # In a traceback, the list is TOP to BOTTOM (outer to inner) - # second_stack_caller -> capture_it -> middle_step -> inner_raise - assert second_idx < middle_idx - assert middle_idx < inner_idx - - -def test_chained_exceptions_stack(tmp_path: Path) -> None: - """Test serialization of chained exceptions.""" - dump_file = tmp_path / "chained.dump" - - def fail_inner() -> Never: - msg = "Inner key error" - raise KeyError(msg) - - def fail_outer() -> None: - try: - fail_inner() - except KeyError as e: - msg = "Outer runtime error" - raise RuntimeError(msg) from e - - try: - fail_outer() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - with pytest.raises(RuntimeError, match="Outer runtime error") as exc_info: - load_traceback(str(dump_file)) - - reconstructed_exc = exc_info.value - assert isinstance(reconstructed_exc.__cause__, KeyError) - - def get_frames(tb: types.TracebackType | None) -> list[types.FrameType]: - f = [] - curr = tb - while curr is not None: - f.append(curr.tb_frame) - curr = curr.tb_next - return f - - outer_frames = get_frames(reconstructed_exc.__traceback__) - outer_names = [f.f_code.co_name for f in outer_frames] - assert "fail_outer" in outer_names - - inner_exc = reconstructed_exc.__cause__ - assert isinstance(inner_exc, KeyError) - inner_frames = get_frames(inner_exc.__traceback__) - inner_names = [f.f_code.co_name for f in inner_frames] - assert "fail_inner" in inner_names - - -def test_unpicklable_locals_verification(tmp_path: Path) -> None: - """Test that unpicklable local variables are handled gracefully.""" - dump_file = tmp_path / "unpicklable.dump" - - class Unpicklable: - def __reduce__(self) -> Never: - msg = "Cannot pickle me" - raise TypeError(msg) - - def __repr__(self) -> str: - return "" - - def fail_with_unpicklable() -> Never: - _obj = Unpicklable() - # Use locals() to force inclusion in the locals dictionary. - _ = locals()["_obj"] - msg = "Error with unpicklable" - raise ValueError(msg) - - try: - fail_with_unpicklable() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - with pytest.raises(ValueError, match="Error with unpicklable") as exc_info: - load_traceback(str(dump_file)) - - frames = get_frames(exc_info.tb) - - frame_names = [f.f_code.co_name for f in frames] - assert "fail_with_unpicklable" in frame_names - f = next(f for f in frames if f.f_code.co_name == "fail_with_unpicklable") - # Verify that our filtering logic caught the unpicklable item. - assert any(" None: - """Test that global variables are preserved in the stack.""" - dump_file = tmp_path / "globals.dump" - - def fail_with_globals() -> None: - global GLOBAL_TEST_VAL # noqa: PLW0602 - if GLOBAL_TEST_VAL == "I am global": - msg = "Global test" - raise ValueError(msg) - - try: - fail_with_globals() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - with pytest.raises(ValueError, match="Global test") as exc_info: - load_traceback(str(dump_file)) - - frames = get_frames(exc_info.tb) - - frame_names = [f.f_code.co_name for f in frames] - assert "fail_with_globals" in frame_names - f = next(f for f in frames if f.f_code.co_name == "fail_with_globals") - assert f.f_globals["GLOBAL_TEST_VAL"] == "I am global" - - -def test_globals_changing_between_frames(tmp_path: Path) -> None: - """Test that globals are captured per-frame for different modules.""" - dump_file = tmp_path / "globals_changing.dump" - - # Define the second module's content inside a function - def module2_content() -> None: - GLOBAL_VAR = "initial" # noqa: N806, F841 - - def level_2() -> Never: - global GLOBAL_VAR - GLOBAL_VAR = "changed" - msg = "Error at level 2" - raise ValueError(msg) - - mod2_code = func_to_module_string(module2_content) - mod2 = types.ModuleType("mod2") - exec(mod2_code, mod2.__dict__) # noqa: S102 - - def level_1() -> None: - try: - mod2.level_2() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - level_1() - - with pytest.raises(ValueError, match="Error at level 2") as exc_info: - load_traceback(str(dump_file)) - - frames = get_frames(exc_info.tb) - - l1_f = next(f for f in frames if f.f_code.co_name == "level_1") - l2_f = next(f for f in frames if f.f_code.co_name == "level_2") - - # mod2.level_2 changed its OWN GLOBAL_VAR. - # level_1's globals (this module) should NOT have GLOBAL_VAR (or it should be different) - assert l2_f.f_globals["GLOBAL_VAR"] == "changed" - assert "GLOBAL_VAR" not in l1_f.f_globals or l1_f.f_globals["GLOBAL_VAR"] == "initial" - - -def test_unpicklable_exception_coverage(tmp_path: Path) -> None: - """Test that unpicklable exceptions are handled by falling back to RuntimeError.""" - dump_file = tmp_path / "unpicklable_exc.dump" - - class UnpicklableError(Exception): - def __reduce__(self) -> Never: - msg = "Cannot pickle me" - raise TypeError(msg) - - def raise_unpicklable() -> Never: - msg = "Unpicklable" - raise UnpicklableError(msg) - - try: - raise_unpicklable() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - with pytest.raises(RuntimeError, match="Unpicklable exception UnpicklableError: Unpicklable"): - load_traceback(str(dump_file)) - - -def test_typing_never() -> None: - """Test that load_traceback is correctly annotated with Never.""" - import typing - - from offline_debug import load_traceback - - load_traceback_annotations = typing.get_type_hints(load_traceback) - assert load_traceback_annotations["return"] is typing.Never - - -def test_load_invalid_object(tmp_path: Path) -> None: - """Test that load_traceback raises TypeError when loading an invalid object.""" - import pickle - - dump_file = tmp_path / "invalid.dump" - with dump_file.open("wb") as f: - pickle.dump("not an ExceptionData object", f) - - with pytest.raises(TypeError, match="Expected _ExceptionData, but got str"): - load_traceback(str(dump_file)) - - -def test_load_non_existent_file() -> None: - """Test that load_traceback raises FileNotFoundError when the file does not exist.""" - with pytest.raises(FileNotFoundError): - load_traceback("non_existent_file.dump") - - -def test_reconstruct_invalid_exception_type() -> None: - """Test that _reconstruct_exc_data raises TypeError when the pickled exception is invalid.""" - # Create dummy data with a string instead of a pickled exception - import pickle - - from offline_debug.serializer import _ExceptionData, _reconstruct_exc_data - - data = _ExceptionData( - exc_pickle=pickle.dumps("not an exception"), - tb_frames=[], - ) - - with pytest.raises(TypeError, match="Expected BaseException, but got str"): - _reconstruct_exc_data(data) - - -def test_reconstruct_invalid_frame_type(monkeypatch) -> None: - """Test that _reconstruct_exc_data raises TypeError when frame creation fails.""" - from offline_debug.serializer import _ExceptionData, _FrameData, _reconstruct_exc_data - - # Mock _py_frame_new to return something that is not a FrameType - monkeypatch.setattr("offline_debug.serializer._py_frame_new", lambda *_: "not a frame") - - import marshal - import pickle - - def dummy() -> None: - pass - - data = _ExceptionData( - exc_pickle=pickle.dumps(ValueError("test")), - tb_frames=[ - _FrameData( - code=marshal.dumps(dummy.__code__), - globals={}, - locals={}, - lasti=0, - lineno=0, - stack_depth=0, - ) - ], - ) - - with pytest.raises(TypeError, match=r"Expected types.FrameType, but got str"): - _reconstruct_exc_data(data) - - -def test_get_f_back_offset_logic() -> None: - """Test the dynamic f_back offset discovery logic directly.""" - from offline_debug.serializer import _get_f_back_offset - - offset = _get_f_back_offset() - # It should either find an offset or be None (if platform is weird) - # But on standard CPython it should find something. - assert offset is None or (offset > 0 and offset % 8 == 0) - - -def test_link_frame_no_offset(monkeypatch) -> None: - """Test that _link_frame does nothing when offset is None.""" - import offline_debug.serializer as ser - - monkeypatch.setattr(ser, "_F_BACK_OFFSET", None) - - f = sys._getframe() - # Should not raise - ser._link_frame(f, f) - - -def test_reconstructed_frames_have_f_back(tmp_path: Path) -> None: - """ - Test that reconstructed frames have their f_back pointers correctly linked. - - This test is currently expected to FAIL because f_back linking was removed - to avoid segmentation faults, but it's required for full fidelity. - """ - dump_file = tmp_path / "f_back_fidelity.dump" - - def level_2() -> Never: - msg = "Fidelity error" - raise ValueError(msg) - - def level_1() -> None: - try: - level_2() - except Exception as e: # noqa: BLE001 - save_traceback(e, str(dump_file)) - - level_1() - - with pytest.raises(ValueError, match="Fidelity error") as exc_info: - load_traceback(str(dump_file)) - - frames = get_frames(exc_info.tb) - # Traceback frames are ordered TOP to BOTTOM (outer to inner) - # We want to check that level_2's frame points back to level_1 - l2_f = next(f for f in frames if f.f_code.co_name == "level_2") - l1_f = next(f for f in frames if f.f_code.co_name == "level_1") - - assert l2_f.f_back is not None, "level_2.f_back should not be None" - assert l2_f.f_back is l1_f, "level_2.f_back should point to level_1" - - -def test_locals_visibility_in_reconstructed_frames(tmp_path: Path) -> None: - """Regression test: verify local variables are visible in reconstructed frames.""" - dump_file = tmp_path / "locals_visibility.dump" - expected_val = 42 - - def func_with_locals() -> Never: - var_a = "hello" - var_b = expected_val - # Use them to ensure they aren't optimized away in some versions - _ = f"{var_a} {var_b}" - msg = "Locals error" - raise ValueError(msg) - - try: - func_with_locals() - except ValueError as e: - save_traceback(e, str(dump_file)) - - with pytest.raises(ValueError, match="Locals error") as exc_info: - load_traceback(str(dump_file)) - - frames = get_frames(exc_info.tb) - f = next(f for f in frames if f.f_code.co_name == "func_with_locals") - - assert f.f_locals.get("var_a") == "hello" - assert f.f_locals.get("var_b") == expected_val