diff --git a/strix/telemetry/logging.py b/strix/telemetry/logging.py index 5c152cc9..670b2236 100644 --- a/strix/telemetry/logging.py +++ b/strix/telemetry/logging.py @@ -6,7 +6,6 @@ import contextlib import logging import os import sys -import threading import warnings from contextvars import ContextVar from pathlib import Path # noqa: TC003 used at runtime by ``setup_scan_logging`` @@ -89,34 +88,20 @@ def configure_dependency_logging() -> None: logging.getLogger("asyncio").setLevel(logging.CRITICAL) logging.getLogger("asyncio").propagate = False warnings.filterwarnings("ignore", category=RuntimeWarning, module="asyncio") - _route_hook_exceptions_to_log() + _route_unraisable_to_log() -def _route_hook_exceptions_to_log() -> None: - """Python prints finalizer and thread tracebacks to stderr; send them to strix.log instead. - - Dropped when the strix loggers have no handler (after scan teardown) - rather than left to ``logging.lastResort``, and any failure inside the - hook is swallowed: the interpreter may be shutting down. - """ - - def log(message: str, exc_info: tuple[object, object, object]) -> None: +def _route_unraisable_to_log() -> None: + def hook(u: sys.UnraisableHookArgs) -> None: with contextlib.suppress(BaseException): logger = logging.getLogger("strix.telemetry") if logger.hasHandlers(): - logger.warning(message, exc_info=exc_info) # type: ignore[arg-type] + logger.warning( + u.err_msg or f"Exception ignored in {u.object!r}", + exc_info=(u.exc_type, u.exc_value, u.exc_traceback), # type: ignore[arg-type] + ) - def unraisable_hook(u: sys.UnraisableHookArgs) -> None: - log( - u.err_msg or f"Exception ignored in {u.object!r}", - (u.exc_type, u.exc_value, u.exc_traceback), - ) - - def thread_hook(a: threading.ExceptHookArgs) -> None: - log(f"Exception in thread {a.thread}", (a.exc_type, a.exc_value, a.exc_traceback)) - - sys.unraisablehook = unraisable_hook - threading.excepthook = thread_hook + sys.unraisablehook = hook class _CurrentStderrHandler(logging.StreamHandler): # type: ignore[type-arg] diff --git a/tests/test_hook_exceptions_logged.py b/tests/test_unraisable_hook.py similarity index 56% rename from tests/test_hook_exceptions_logged.py rename to tests/test_unraisable_hook.py index 41fbcd32..29d4cae5 100644 --- a/tests/test_hook_exceptions_logged.py +++ b/tests/test_unraisable_hook.py @@ -1,18 +1,9 @@ -"""Nothing from sys.unraisablehook or threading.excepthook reaches the terminal. - -Python prints "Exception ignored in ..." reports (finalizers, ``__del__``, GC -and weakref callbacks, files closed at interpreter exit) and uncaught thread -exceptions straight to stderr. Strix routes both to its own logger instead, -so they end up in strix.log (and on stderr only under STRIX_DEBUG=1). -""" - from __future__ import annotations import logging import os import subprocess import sys -import threading from pathlib import Path from types import SimpleNamespace from typing import TYPE_CHECKING @@ -54,14 +45,11 @@ def strix_records() -> Iterator[list[logging.LogRecord]]: @pytest.fixture -def hooks(monkeypatch: pytest.MonkeyPatch) -> tuple[list[object], list[object]]: - """Install strix's hooks over recording stand-ins for Python's defaults.""" - unraisable_calls: list[object] = [] - thread_calls: list[object] = [] - monkeypatch.setattr(sys, "unraisablehook", unraisable_calls.append) - monkeypatch.setattr(threading, "excepthook", thread_calls.append) +def default_hook_calls(monkeypatch: pytest.MonkeyPatch) -> list[object]: + calls: list[object] = [] + monkeypatch.setattr(sys, "unraisablehook", calls.append) tlog.configure_dependency_logging() - return unraisable_calls, thread_calls + return calls @pytest.mark.parametrize( @@ -82,15 +70,13 @@ def hooks(monkeypatch: pytest.MonkeyPatch) -> tuple[list[object], list[object]]: ) def test_every_unraisable_is_logged_not_printed( args: object, - hooks: tuple[list[object], list[object]], + default_hook_calls: list[object], strix_records: list[logging.LogRecord], capsys: pytest.CaptureFixture[str], ) -> None: - unraisable_calls, _ = hooks - sys.unraisablehook(args) # type: ignore[arg-type] - assert unraisable_calls == [] + assert default_hook_calls == [] assert capsys.readouterr().err == "" [record] = strix_records assert record.levelno == logging.WARNING @@ -103,12 +89,10 @@ def test_every_unraisable_is_logged_not_printed( assert str(args.exc_value) in message # type: ignore[attr-defined] -@pytest.mark.usefixtures("hooks") +@pytest.mark.usefixtures("default_hook_calls") def test_unraisable_is_dropped_when_strix_has_no_handlers( capsys: pytest.CaptureFixture[str], ) -> None: - """After scan teardown the strix logger tree has no handlers; the report - must be dropped rather than handed to logging.lastResort (stderr).""" root = logging.getLogger("strix") saved, root.handlers = root.handlers, [] try: @@ -118,31 +102,8 @@ def test_unraisable_is_dropped_when_strix_has_no_handlers( assert capsys.readouterr().err == "" -def test_thread_exceptions_are_logged_not_printed( - hooks: tuple[list[object], list[object]], - strix_records: list[logging.LogRecord], - capsys: pytest.CaptureFixture[str], -) -> None: - _, thread_calls = hooks - - def _boom() -> None: - raise RuntimeError("worker failed") - - thread = threading.Thread(target=_boom, name="strix-worker") - thread.start() - thread.join() - - assert thread_calls == [] - assert capsys.readouterr().err == "" - [record] = strix_records - assert record.levelno == logging.WARNING - message = logging.Formatter().format(record) - assert "Exception in thread None: - """End to end in a fresh interpreter: finalizer errors (including one at - interpreter shutdown), a thread traceback and the closed-file response - case produce no stderr output, with the default (non-debug) handlers.""" - env = {"PYTHONPATH": str(Path(__file__).resolve().parents[1])} + env = { + k: v for k, v in os.environ.items() if k in {"PATH", "HOME", "SYSTEMROOT", "TEMP", "TMP"} + } + env |= {"PYTHONPATH": str(Path(__file__).resolve().parents[1]), "STRIX_DEBUG": ""} proc = subprocess.run( # noqa: S603 [sys.executable, "-c", _EXIT_SCRIPT], capture_output=True, text=True, timeout=60, - env={**_safe_env(), **env, "STRIX_DEBUG": ""}, + env=env, check=False, ) assert proc.returncode == 0, proc.stderr assert proc.stdout.strip() == "done" assert proc.stderr == "" - - -def _safe_env() -> dict[str, str]: - return { - k: v for k, v in os.environ.items() if k in {"PATH", "HOME", "SYSTEMROOT", "TEMP", "TMP"} - }