Make hook tracing unable to fail a hook call

changelog/424.bugfix.rst

@@ -0,0 +1,5 @@
+Tracing no longer breaks hook execution when a traced object has a broken ``__repr__``
+or ``__str__``. Such a value is now rendered as
+``<[RuntimeError(...) raised in repr()] Broken object at 0x...>``, in the same style
+pytest uses for unpresentable objects, instead of propagating the exception out of the
+hook call.

changelog/681.bugfix.rst

@@ -0,0 +1,3 @@
+Tracing no longer crashes with ``UnicodeEncodeError`` when a hook argument or return
+value contains lone surrogates; they are escaped with ``backslashreplace`` before the
+message reaches the writer. Trace output is otherwise unchanged.

docs/index.rst

@@ -997,6 +997,20 @@ undo function to disable the behaviour.
     pm.trace.root.setwriter(print)
     undo = pm.enable_tracing()
 
+Each hook call is traced with its keyword arguments, followed by a ``finish``
+line carrying the result::
+
+    he_method1 [hook]
+        arg: value
+        path: /tmp
+    finish he_method1 --> ['value'] [hook]
+
+Values are rendered with :func:`str`, and the rendering is defensive: an object
+whose ``__str__`` raises is shown as
+``<[RuntimeError(...) raised in str()] Broken object at 0x...>``, and lone
+surrogates are backslash-escaped, so enabling tracing can never turn a working
+hook call into a failing one.
+
 
 Call monitoring
 ---------------

src/pluggy/_tracing.py

@@ -13,6 +13,54 @@
 _Processor = Callable[[tuple[str, ...], tuple[Any, ...]], object]
 
 
+def _try_repr_or_str(obj: object) -> str:
+    try:
+        return repr(obj)
+    except (KeyboardInterrupt, SystemExit):
+        raise
+    except BaseException:
+        return f'{type(obj).__name__}("{obj}")'
+
+
+def _format_str_exception(exc: BaseException, obj: object) -> str:
+    try:
+        exc_info = _try_repr_or_str(exc)
+    except (KeyboardInterrupt, SystemExit):
+        raise
+    except BaseException as inner:
+        exc_info = f"unpresentable exception ({_try_repr_or_str(inner)})"
+    name = type(obj).__name__
+    return f"<[{exc_info} raised in str()] {name} object at 0x{id(obj):x}>"
+
+
+def _escape_surrogates(text: str) -> str:
+    """Escape lone surrogates so the result survives any text writer.
+
+    A lone surrogate reaching the writer raises :exc:`UnicodeEncodeError`
+    inside the trace call for any utf-8 target, such as the file behind
+    pytest's ``--debug``.
+    """
+    if text.isascii():
+        return text
+    return text.encode("utf-8", "backslashreplace").decode("utf-8")
+
+
+def _safe_str(obj: object) -> str:
+    """``str(obj)`` for tracing, guaranteed not to raise and always writable.
+
+    Tracing is a debugging aid, so it must never be the reason a hook call
+    fails, and the rendering stays ``str``-based to keep the trace output
+    readable.
+    """
+    try:
+        text = str(obj)
+    except (KeyboardInterrupt, SystemExit):
+        raise
+    except BaseException as exc:
+        text = _format_str_exception(exc, obj)
+    return _escape_surrogates(text)
+
+
 class TagTracer:
     def __init__(self) -> None:
         self._tags2proc: dict[tuple[str, ...], _Processor] = {}
@@ -29,13 +77,13 @@ def _format_message(self, tags: Sequence[str], args: Sequence[object]) -> str:
         else:
             extra = {}
 
-        content = " ".join(map(str, args))
+        content = " ".join(map(_safe_str, args))
         indent = "  " * self.indent
 
         lines = [f"{indent}{content} [{':'.join(tags)}]\n"]
 
         for name, value in extra.items():
-            lines.append(f"{indent}    {name}: {value}\n")
+            lines.append(f"{indent}    {name}: {_safe_str(value)}\n")
 
         return "".join(lines)
 

testing/test_pluginmanager.py

@@ -912,6 +912,77 @@ def he_method1(self):
         undo()
 
 
+def test_hook_tracing_escapes_surrogate_values(pm: PluginManager) -> None:
+    """Surrogates in traced arguments and results never reach the writer.
+
+    Regression test for #681 (pytest-dev/pytest#13750).
+    """
+
+    class Hooks:
+        @hookspec(firstresult=True)
+        def he_method1(self, arg: object) -> object:
+            raise NotImplementedError()
+
+    class Plugin:
+        @hookimpl
+        def he_method1(self, arg: object) -> object:
+            return arg
+
+    out: list[str] = []
+
+    def write(message: str) -> None:
+        message.encode()
+        out.append(message)
+
+    pm.add_hookspecs(Hooks)
+    pm.register(Plugin())
+    pm.trace.root.setwriter(write)
+    undo = pm.enable_tracing()
+    try:
+        result = pm.hook.he_method1(arg="\ud800")
+    finally:
+        undo()
+
+    assert result == "\ud800"
+    assert out == [
+        "  he_method1 [hook]\n      arg: \\ud800\n",
+        "  finish he_method1 --> \\ud800 [hook]\n",
+    ]
+
+
+def test_hook_tracing_with_broken_repr(he_pm: PluginManager) -> None:
+    """A broken ``__repr__`` does not break the hook call.
+
+    Regression test for #424 (kedro-org/kedro#2630).
+    """
+
+    class BrokenRepr:
+        def __repr__(self) -> str:
+            raise RuntimeError("repr is broken")
+
+    class api1:
+        @hookimpl
+        def he_method1(self, arg):
+            return arg
+
+    he_pm.register(api1())
+    out: list[str] = []
+    he_pm.trace.root.setwriter(out.append)
+    undo = he_pm.enable_tracing()
+    arg = BrokenRepr()
+    try:
+        result = he_pm.hook.he_method1(arg=arg)
+    finally:
+        undo()
+
+    assert result == [arg]
+    assert len(out) == 2
+    assert "he_method1" in out[0]
+    assert "RuntimeError('repr is broken') raised in str()" in out[0]
+    assert "BrokenRepr object at 0x" in out[0]
+    assert "finish" in out[1]
+
+
 @pytest.mark.parametrize("historic", [False, True])
 def test_register_while_calling(
     pm: PluginManager,

testing/test_tracer.py

@@ -158,3 +158,130 @@ def hello_again(self, arg):
         "  hello [hook]\n      arg: 3\n",
         "  finish hello --> [] [hook]\n",
     ]
+
+
+class BrokenRepr:
+    def __repr__(self) -> str:
+        raise RuntimeError("repr is broken")
+
+
+class BrokenStr:
+    def __str__(self) -> str:
+        raise RuntimeError("str is broken")
+
+
+class SurrogateRepr:
+    def __repr__(self) -> str:
+        return "\ud800"
+
+
+def test_dictargs_keep_str_rendering(rootlogger: TagTracer) -> None:
+    """Values keep their ``str`` rendering, the trace is a log not a repr dump."""
+    out = rootlogger._format_message(["test"], ["call", {"name": "value", "n": 1}])
+    assert out == "call [test]\n    name: value\n    n: 1\n"
+
+
+def test_dictargs_escape_surrogate_values(rootlogger: TagTracer) -> None:
+    out = rootlogger._format_message(["test"], ["test", {"arg": "\ud800"}])
+    assert out == "test [test]\n    arg: \\ud800\n"
+    out.encode()
+
+
+def test_escape_surrogates_from_repr(rootlogger: TagTracer) -> None:
+    """A surrogate coming out of the object's own repr is escaped too."""
+    out = rootlogger._format_message(["test"], ["test", {"arg": SurrogateRepr()}])
+    assert out == "test [test]\n    arg: \\ud800\n"
+    out.encode()
+
+
+def test_escape_surrogates_in_labels(rootlogger: TagTracer) -> None:
+    out = rootlogger._format_message(["test"], ["\ud800"])
+    assert out == "\\ud800 [test]\n"
+    out.encode()
+
+
+def test_non_ascii_values_are_kept(rootlogger: TagTracer) -> None:
+    """Legible text is not mangled, only lone surrogates are escaped."""
+    out = rootlogger._format_message(["test"], ["héllo", {"arg": "wörld"}])
+    assert out == "héllo [test]\n    arg: wörld\n"
+    out.encode()
+
+
+def test_broken_repr_value_does_not_raise(rootlogger: TagTracer) -> None:
+    out = rootlogger._format_message(["test"], ["test", {"arg": BrokenRepr()}])
+    assert "RuntimeError('repr is broken') raised in str()" in out
+    assert "BrokenRepr object at 0x" in out
+    out.encode()
+
+
+def test_broken_str_label_does_not_raise(rootlogger: TagTracer) -> None:
+    out = rootlogger._format_message(["test"], [BrokenStr()])
+    assert "RuntimeError('str is broken') raised in str()" in out
+    assert "BrokenStr object at 0x" in out
+    out.encode()
+
+
+def test_keyboard_interrupt_from_str_propagates(rootlogger: TagTracer) -> None:
+    """Ctrl-C during a traced call still interrupts, it is not swallowed."""
+
+    class Interrupting:
+        def __str__(self) -> str:
+            raise KeyboardInterrupt
+
+    with pytest.raises(KeyboardInterrupt):
+        rootlogger._format_message(["test"], ["test", {"arg": Interrupting()}])
+
+
+def test_broken_exception_repr_is_handled(rootlogger: TagTracer) -> None:
+    """The exception explaining the failure may itself be unpresentable."""
+
+    class BadError(Exception):
+        def __repr__(self) -> str:
+            raise RuntimeError("exception repr is broken")
+
+        def __str__(self) -> str:
+            return "readable message"
+
+    class Broken:
+        def __str__(self) -> str:
+            raise BadError
+
+    out = rootlogger._format_message(["test"], ["test", {"arg": Broken()}])
+    assert 'BadError("readable message") raised in str()' in out
+    assert "Broken object at 0x" in out
+
+
+def test_unpresentable_exception_is_handled(rootlogger: TagTracer) -> None:
+    """Neither repr nor str of the exception works, and tracing still survives."""
+
+    class UnpresentableError(Exception):
+        def __repr__(self) -> str:
+            raise RuntimeError("exception repr is broken")
+
+        def __str__(self) -> str:
+            raise RuntimeError("exception str is broken")
+
+    class Broken:
+        def __str__(self) -> str:
+            raise UnpresentableError
+
+    out = rootlogger._format_message(["test"], ["test", {"arg": Broken()}])
+    assert "unpresentable exception (RuntimeError('exception str is broken'))" in out
+    assert "Broken object at 0x" in out
+
+
+def test_keyboard_interrupt_from_exception_repr_propagates(
+    rootlogger: TagTracer,
+) -> None:
+    """Ctrl-C while rendering the failure explanation propagates as well."""
+
+    class InterruptingError(Exception):
+        def __repr__(self) -> str:
+            raise KeyboardInterrupt
+
+    class Broken:
+        def __str__(self) -> str:
+            raise InterruptingError
+
+    with pytest.raises(KeyboardInterrupt):
+        rootlogger._format_message(["test"], ["test", {"arg": Broken()}])