Skip to content
Merged
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
27 changes: 27 additions & 0 deletions exec_helpers/_helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,11 +7,38 @@
import shlex
import typing

from exec_helpers import exceptions

if typing.TYPE_CHECKING:
from collections.abc import Iterable
from collections.abc import Mapping


class _ExceptionInfo(typing.Protocol):
"""The part of ``pytest.ExceptionInfo`` that :func:`hide_traceback` reads."""

@property
def value(self) -> BaseException:
"""Get the exception being reported."""


def hide_traceback(excinfo: _ExceptionInfo | None) -> bool:
"""Tell pytest whether to leave a frame of this package out of a failure report.

A module sets its ``__tracebackhide__`` global to this function, and pytest calls it for every frame of that module.
An error this package raises on purpose - a timeout, an unexpected exit code - already names the command
and carries its output, so the report shows the caller's line and the message.
Any other exception keeps the frames, because they show where it was raised.

:param excinfo: The exception pytest is reporting, or ``None`` when it filters frames without one.
:returns: Whether the frame is left out of the report.

>>> hide_traceback(None)
False
"""
return excinfo is not None and isinstance(excinfo.value, exceptions.ExecHelperError)


def string_bytes_bytearray_as_bytes(src: str | bytes | bytearray) -> bytes:
"""Get byte string from string/bytes/bytearray union.

Expand Down
2 changes: 2 additions & 0 deletions exec_helpers/_ssh_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,8 @@

__all__ = ("SSHClientBase", "SshExecuteAsyncResult", "SupportPathT")

__tracebackhide__: typing.Final = _helpers.hide_traceback

KeepAlivePeriodT = int | bool
SupportPathT = str | pathlib.PurePath

Expand Down
2 changes: 2 additions & 0 deletions exec_helpers/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,8 @@
"OptionalTimeoutT",
)

__tracebackhide__: typing.Final = _helpers.hide_traceback

CommandT = str | Iterable[str]
LogMaskReT = str | re.Pattern[str] | None
ErrorInfoT = str | None
Expand Down
2 changes: 2 additions & 0 deletions exec_helpers/async_api/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,8 @@
"OptionalTimeoutT",
)

__tracebackhide__: typing.Final = _helpers.hide_traceback


class ExecuteContext(contextlib.AbstractAsyncContextManager[api.ExecuteAsyncResult], abc.ABC):
"""Execute context manager."""
Expand Down
2 changes: 2 additions & 0 deletions exec_helpers/async_api/subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,8 @@

__all__ = ("Subprocess", "SubprocessExecuteAsyncResult")

__tracebackhide__: typing.Final = _helpers.hide_traceback

# Seconds to wait for the streams to reach the end after the process exit.
_READERS_DRAIN_TIMEOUT: typing.Final = 1.0

Expand Down
4 changes: 4 additions & 0 deletions exec_helpers/exec_result.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,8 @@
from exec_helpers import exceptions
from exec_helpers import proc_enums

from . import _helpers

try:
# noinspection PyPackageRequirements
import yaml
Expand Down Expand Up @@ -60,6 +62,8 @@

__all__ = ("ExecResult", "OptionalStdinT")

__tracebackhide__: typing.Final = _helpers.hide_traceback

LOGGER: logging.Logger = logging.getLogger(__name__)

OptionalStdinT = bytes | str | bytearray | None
Expand Down
2 changes: 2 additions & 0 deletions exec_helpers/subprocess.py
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,8 @@

__all__ = ("CwdT", "EnvT", "Subprocess", "SubprocessExecuteAsyncResult")

__tracebackhide__: typing.Final = _helpers.hide_traceback

EnvT = Mapping[str, str] | None
CwdT = str | bytes | pathlib.Path | None

Expand Down
19 changes: 18 additions & 1 deletion test/test_subprocess_special.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@

from __future__ import annotations

import pathlib
import subprocess
import typing
from unittest import mock
Expand All @@ -32,6 +33,8 @@
from subprocess_cases import default_timeout

timeout_expired = subprocess.TimeoutExpired(command, default_timeout)
# Frames under this directory are what ``__tracebackhide__`` leaves out of a pytest report.
PACKAGE = pathlib.Path(exec_helpers.__file__).parent


@pytest.mark.parametrize(SPECIAL_ARGNAMES, SPECIAL_CASES)
Expand Down Expand Up @@ -109,8 +112,22 @@ def test_timeout(mocker, fake_popen, poll, expected_exc) -> None:
kill_tree = mocker.patch("exec_helpers._subprocess_helpers.kill_proc_tree")
proc = fake_popen(wait=(timeout_expired,), poll=(poll,))

with pytest.raises(expected_exc):
with pytest.raises(expected_exc) as exc_info:
exec_helpers.Subprocess().execute(command)

kill_tree.assert_called_once_with(proc.pid)
proc.poll.assert_called_once_with()
assert not [
entry for entry in exc_info.traceback.filter(exc_info) if pathlib.Path(entry.path).is_relative_to(PACKAGE)
]


def test_unexpected_error_keeps_traceback(mocker, fake_popen) -> None:
"""An exception the package does not raise on purpose keeps the package frames in a pytest report."""
mocker.patch("exec_helpers._subprocess_helpers.kill_proc_tree", side_effect=RuntimeError)
fake_popen(wait=(timeout_expired,))

with pytest.raises(RuntimeError) as exc_info:
exec_helpers.Subprocess().execute(command)

assert [entry for entry in exc_info.traceback.filter(exc_info) if pathlib.Path(entry.path).is_relative_to(PACKAGE)]
Loading