exception_chain.py
<<<<<<< SEARCH
# exception_chain: placeholder
=======
from __future__ import annotations

import traceback as _traceback


def _safe_str(value):
    try:
        return str(value)
    except Exception:
        try:
            return repr(value)
        except Exception:
            return "<unrepresentable %s>" % type(value).__name__


def _safe_repr(value):
    try:
        return repr(value)
    except Exception:
        return "<unrepresentable %s>" % type(value).__name__


def _sentinel(type_name="Unknown", *, cycle=False, invalid=False, truncated=False):
    return {
        "type": type_name,
        "message": "",
        "args": [],
        "suppress_context": False,
        "has_traceback": False,
        "traceback": None,
        "cycle": cycle,
        "invalid": invalid,
        "truncated": truncated,
    }


def _format_traceback(exc):
    try:
        tb = getattr(exc, "__traceback__", None)
    except Exception:
        return None
    if tb is None:
        return None
    try:
        return "".join(_traceback.format_tb(tb))
    except Exception:
        return None


def _entry(exc, *, cycle=False, invalid=False, truncated=False):
    is_exception = isinstance(exc, BaseException)
    raw_args = getattr(exc, "args", ()) if is_exception else ()
    try:
        args = [_safe_repr(arg) for arg in raw_args]
    except Exception:
        args = [_safe_repr(raw_args)]
    tb_text = _format_traceback(exc) if is_exception else None
    try:
        suppress = bool(getattr(exc, "__suppress_context__", False)) if is_exception else False
    except Exception:
        suppress = False
    return {
        "type": type(exc).__name__,
        "message": _safe_str(exc),
        "args": args,
        "suppress_context": suppress,
        "has_traceback": tb_text is not None,
        "traceback": tb_text,
        "cycle": cycle,
        "invalid": invalid,
        "truncated": truncated,
    }


def serialize_exception_chain(exc, max_depth=10):
    """Pickle-safe chain following __cause__, else __context__ unless suppressed.

    Uses only call-local state, so concurrent/xdist serialization is safe
    and the original exceptions are never mutated.
    """
    if max_depth < 1:
        raise ValueError("max_depth must be >= 1")
    if exc is None:
        return []
    result = []
    seen = set()
    current = exc
    depth = 0
    while current is not None:
        if not isinstance(current, BaseException):
            result.append(_entry(current, invalid=True))
            break
        if id(current) in seen:
            result.append(_entry(current, cycle=True))
            break
        if depth >= max_depth:
            result.append(_entry(current, truncated=True))
            break
        seen.add(id(current))
        result.append(_entry(current))
        depth += 1
        try:
            cause = getattr(current, "__cause__", None)
        except Exception:
            result.append(_sentinel(invalid=True))
            break
        if cause is not None:
            current = cause
            continue
        try:
            if bool(getattr(current, "__suppress_context__", False)):
                break
            current = getattr(current, "__context__", None)
        except Exception:
            result.append(_sentinel(invalid=True))
            break
    return result
>>>>>>> REPLACE
test_exception_chain.py
<<<<<<< SEARCH
# test_exception_chain: placeholder
=======
import pickle
import threading
from concurrent.futures import ThreadPoolExecutor

import pytest

from exception_chain import serialize_exception_chain


class _BadCause(ValueError):
    @property
    def __cause__(self):  # type: ignore[override]
        return "not-a-base-exception"  # type: ignore[return-value]

    @__cause__.setter
    def __cause__(self, value):
        self.__dict__["_stored_cause"] = value


class _BadContext(ValueError):
    @property
    def __context__(self):  # type: ignore[override]
        return 123  # type: ignore[return-value]

    @__context__.setter
    def __context__(self, value):
        self.__dict__["_stored_context"] = value


def _make_mixed_chain(length):
    excs = [ValueError("e%d" % i) for i in range(length)]
    for i in range(length - 1):
        if i % 2 == 0:
            excs[i].__cause__ = excs[i + 1]
        else:
            excs[i].__context__ = excs[i + 1]
            excs[i].__suppress_context__ = False
    return excs[0]


def test_no_cause_no_context_empty_args():
    exc = ValueError()
    assert exc.__cause__ is None
    assert exc.__context__ is None
    assert exc.args == ()
    chain = serialize_exception_chain(exc)
    assert len(chain) == 1
    assert chain[0]["type"] == "ValueError"
    assert chain[0]["message"] == ""
    assert chain[0]["args"] == []
    assert chain[0]["suppress_context"] is False
    assert chain[0]["has_traceback"] is False
    assert chain[0]["traceback"] is None
    assert chain[0]["cycle"] is False
    assert chain[0]["invalid"] is False
    assert chain[0]["truncated"] is False


def test_explicit_from_vs_implicit_context():
    explicit = None
    try:
        try:
            raise KeyError("inner")
        except KeyError as inner:
            raise ValueError("outer") from inner
    except ValueError as caught:
        explicit = caught
    assert isinstance(explicit.__cause__, KeyError)
    assert explicit.__suppress_context__ is True
    explicit_chain = serialize_exception_chain(explicit)
    assert [e["type"] for e in explicit_chain] == ["ValueError", "KeyError"]
    assert explicit_chain[0]["suppress_context"] is True
    assert explicit_chain[0]["has_traceback"] is True
    assert explicit_chain[0]["traceback"]

    implicit = None
    try:
        try:
            raise KeyError("inner")
        except KeyError:
            raise ValueError("outer")
    except ValueError as caught:
        implicit = caught
    assert implicit.__cause__ is None
    assert isinstance(implicit.__context__, KeyError)
    assert implicit.__suppress_context__ is False
    implicit_chain = serialize_exception_chain(implicit)
    assert [e["type"] for e in implicit_chain] == ["ValueError", "KeyError"]
    assert implicit_chain[0]["suppress_context"] is False


def test_raise_from_none_suppresses_nonempty_context():
    caught_exc = None
    try:
        try:
            raise KeyError("inner")
        except KeyError:
            raise ValueError("outer") from None
    except ValueError as caught:
        caught_exc = caught
    assert caught_exc.__cause__ is None
    assert isinstance(caught_exc.__context__, KeyError)
    assert caught_exc.__suppress_context__ is True
    chain = serialize_exception_chain(caught_exc)
    assert len(chain) == 1
    assert chain[0]["type"] == "ValueError"
    assert chain[0]["suppress_context"] is True


def test_cause_context_type_mismatch():
    assert serialize_exception_chain("not-an-exception")[0]["invalid"] is True
    assert serialize_exception_chain(123)[0]["type"] == "int"
    assert serialize_exception_chain(123)[0]["invalid"] is True
    bad_cause = serialize_exception_chain(_BadCause("x"))
    assert [e["type"] for e in bad_cause] == ["_BadCause", "str"]
    assert bad_cause[-1]["invalid"] is True
    assert bad_cause[-1]["message"] == "not-a-base-exception"
    bad_context = serialize_exception_chain(_BadContext("y"))
    assert [e["type"] for e in bad_context] == ["_BadContext", "int"]
    assert bad_context[-1]["invalid"] is True


def test_unpicklable_arg_and_missing_traceback():
    exc = ValueError(threading.Lock())
    exc.__cause__ = TypeError("cause")
    with pytest.raises(Exception):
        pickle.dumps(exc)
    chain = serialize_exception_chain(exc)
    assert pickle.loads(pickle.dumps(chain)) == chain
    assert "lock" in chain[0]["args"][0]
    assert chain[0]["has_traceback"] is False
    assert chain[0]["traceback"] is None
    assert chain[1]["type"] == "TypeError"


@pytest.mark.parametrize(
    "length,max_depth,expected_real,expect_truncated",
    [(1, 1, 1, False), (2, 1, 1, True), (2, 2, 2, False), (3, 3, 3, False), (4, 3, 3, True), (5, 3, 3, True)],
)
def test_chain_depth_off_by_one_mixed(length, max_depth, expected_real, expect_truncated):
    chain = serialize_exception_chain(_make_mixed_chain(length), max_depth=max_depth)
    real = [e for e in chain if not e["truncated"]]
    assert len(real) == expected_real
    assert [e["message"] for e in real] == ["e%d" % i for i in range(expected_real)]
    assert any(e["truncated"] for e in chain) is expect_truncated
    if expect_truncated:
        assert chain[-1]["truncated"] is True
        assert len(chain) == expected_real + 1


def test_max_depth_must_be_positive():
    with pytest.raises(ValueError):
        serialize_exception_chain(ValueError("x"), max_depth=0)


def test_unicode_empty_and_non_str_args_in_chain():
    outer = ValueError("")
    inner = ValueError("héllo 🌍")
    inner.__cause__ = ValueError(123, None, "üñï")
    outer.__cause__ = inner
    chain = serialize_exception_chain(outer)
    assert [e["message"] for e in chain] == ["", "héllo 🌍", "(123, None, 'üñï')"]
    assert chain[1]["args"] == ["'héllo 🌍'"]
    assert chain[2]["args"] == ["123", "None", "'üñï'"]
    assert pickle.loads(pickle.dumps(chain)) == chain


def test_cyclic_cause_and_concurrent_serialization():
    a = ValueError("a")
    b = TypeError("b")
    a.__cause__ = b
    b.__cause__ = a
    chain = serialize_exception_chain(a)
    assert [e["message"] for e in chain[:2]] == ["a", "b"]
    assert chain[-1]["cycle"] is True
    assert chain[-1]["type"] == "ValueError"
    assert a.__cause__ is b
    assert b.__cause__ is a
    self_cycle = ValueError("self")
    self_cycle.__cause__ = self_cycle
    self_chain = serialize_exception_chain(self_cycle)
    assert len(self_chain) == 2
    assert self_chain[-1]["cycle"] is True
    # Call-local seen-set: stateless under threads and pytest-xdist workers.
    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(lambda _: serialize_exception_chain(a), range(32)))
    assert all(result == chain for result in results)
    assert all(pickle.loads(pickle.dumps(result)) == result for result in results)
    assert serialize_exception_chain(b)[0]["message"] == "b"
>>>>>>> REPLACE