Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions changelog/424.bugfix.rst
Original file line number Diff line number Diff line change
@@ -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.
3 changes: 3 additions & 0 deletions changelog/681.bugfix.rst
Original file line number Diff line number Diff line change
@@ -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.
5 changes: 5 additions & 0 deletions changelog/729.feature.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
Traced values now gain detail where ``str()`` is ambiguous: an empty string or a string
carrying whitespace is quoted, an enum member shows its name, and a path shows its type,
so that two arguments pointing at the same place are distinguishable. Values that read
unambiguously as themselves are unchanged, and a value spanning several lines is drawn
as a block attached to its key instead of running into the surrounding trace.
24 changes: 24 additions & 0 deletions docs/index.rst
Original file line number Diff line number Diff line change
Expand Up @@ -997,6 +997,30 @@ 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]
plugin_name: example
path: PosixPath('/tmp')
reason: 'needs a network connection'
status: <ExitCode.TESTS_FAILED: 1>
explanation:
| first line
\ second line
finish he_method1 --> ['value'] [hook]

Values are rendered with :func:`str` wherever that reads unambiguously, and with
:func:`repr` where it does not: an empty string, a string carrying whitespace,
an enum member, or a path, whose type is otherwise easy to lose. A value
spanning several lines is drawn as a block so that it stays attached to its key
instead of running into the surrounding trace.

The rendering is also 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
---------------
Expand Down
106 changes: 104 additions & 2 deletions src/pluggy/_tracing.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,110 @@

from collections.abc import Callable
from collections.abc import Sequence
import enum
import os
from typing import Any


_Writer = Callable[[str], object]
_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_conversion_exception(exc: BaseException, obj: object, func: str) -> 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 {func}()] {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_conversion_exception(exc, obj, "str")
return _escape_surrogates(text)


def _is_plain_token(text: str) -> bool:
"""Whether ``text`` can be shown bare, without quotes around it."""
return bool(text) and text.isprintable() and " " not in text


def _format_block(indent: str, text: str) -> list[str]:
"""Draw a multi line value as a box, so it reads as one value.

The left edge marks every line as continuation, and the final ``\\``
closes it, which keeps a block distinguishable from the trace lines
around it.
"""
body = text.split("\n")
edges = ["|"] * (len(body) - 1) + ["\\"]
return [f"{indent} {edge} {line}\n" for edge, line in zip(edges, body)]


def _render_value(obj: object) -> str:
"""Render a traced value, adding detail only where ``str`` is ambiguous.

Most values keep their plain ``str`` rendering, which is what makes a trace
readable. ``repr`` is used only where ``str`` hides something the reader
needs: the type of a path, the name of an enum member, or the boundaries of
a string that is empty or carries whitespace.
"""
if isinstance(obj, str):
if "\n" in obj or "\r" in obj:
return _safe_str(obj)
if _is_plain_token(obj):
return _safe_str(obj)
return _safe_repr(obj)
if isinstance(obj, (enum.Enum, os.PathLike)):
return _safe_repr(obj)
return _safe_str(obj)


def _safe_repr(obj: object) -> str:
"""``repr(obj)`` for tracing, guaranteed not to raise and always writable."""
try:
text = repr(obj)
except (KeyboardInterrupt, SystemExit):
raise
except BaseException as exc:
text = _format_conversion_exception(exc, obj, "repr")
return _escape_surrogates(text)


class TagTracer:
def __init__(self) -> None:
self._tags2proc: dict[tuple[str, ...], _Processor] = {}
Expand All @@ -29,13 +126,18 @@ 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")
rendered = _render_value(value)
if "\n" in rendered:
lines.append(f"{indent} {name}:\n")
lines.extend(_format_block(indent, rendered))
else:
lines.append(f"{indent} {name}: {rendered}\n")

return "".join(lines)

Expand Down
71 changes: 71 additions & 0 deletions testing/test_pluginmanager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
Loading