fix(webui): crash visibility — faulthandler + thread excepthook + exit audit (#4633)
server.py exited SILENTLY after 9-16h: no traceback, no shutdown-audit line,
no core dump — the log just stopped mid-request. It runs a ThreadingHTTPServer
with daemon_threads=True, so an unhandled exception in a request/SSE/long-poll
handler thread could terminate work with nothing recorded, and faulthandler was
not enabled so a native crash left nothing at all. The WebUI also configures no
logging handlers, so INFO/ERROR records are dropped by logging's lastResort
filter (WARNING+ only) — meaning even the existing shutdown audit never reached
the log.
Add api/crash_visibility.py (stdlib-only, hooks never raise) and wire
install_crash_visibility() into server.main() before any heavy startup:
* faulthandler.enable(all_threads=True) — native crash dumps a C-level
traceback; SIGUSR1 registered for on-demand hang diagnosis.
* threading.excepthook — logs uncaught daemon/handler-thread exceptions
(thread name, ident, traceback) instead of losing them silently.
* sys.excepthook — logs uncaught main-thread exceptions (KeyboardInterrupt
preserved).
* atexit exit-audit breadcrumb — a clean/unwound exit is recorded; its
absence narrows a silent death to an un-unwound kill (OOM/SIGKILL/abort).
All diagnostics are written directly to the fault stream (stderr, which the
bootstrap redirects into the WebUI log) AND mirrored through logging, so the
line is guaranteed to land regardless of logging config.
No request-handling behavior change, no new deps. Paired memory root-cause: #4765.
Fixes #4633.
This commit is contained in:
hermes-agent
parent
412cf01ac2
commit
5bca5db900
3 changed files with 559 additions and 0 deletions
298
api/crash_visibility.py
Normal file
298
api/crash_visibility.py
Normal file
|
|
@ -0,0 +1,298 @@
|
|||
"""Crash-visibility hardening for the Hermes WebUI server (issue #4633).
|
||||
|
||||
Background
|
||||
----------
|
||||
``server.py`` runs a ``ThreadingHTTPServer`` with ``daemon_threads = True``.
|
||||
Under that model a *silent* process death is possible in ways that leave the
|
||||
log ending mid-request with no traceback, no shutdown-audit line, and no core
|
||||
dump:
|
||||
|
||||
* A **native** crash (segfault / abort in a C extension, e.g. sqlite, ssl,
|
||||
a provider SDK) terminates the interpreter with no Python-level traceback.
|
||||
Without ``faulthandler`` enabled, nothing is written.
|
||||
* An **uncaught Python exception in a daemon request/SSE/long-poll thread**
|
||||
is reported through ``threading.excepthook``. The stock hook prints to
|
||||
``sys.stderr`` with no context, and if that stream is gone the report is
|
||||
lost — so a single handler-thread exception can vanish silently.
|
||||
* An uncaught exception on the **main thread** goes through ``sys.excepthook``.
|
||||
|
||||
The paired *root cause* (a memory issue that eventually triggers the death)
|
||||
is being fixed separately in #4765. This module is purely about
|
||||
**diagnostics**: when the process dies we want a recorded reason, and one
|
||||
handler-thread exception must never disappear without a log line.
|
||||
|
||||
Design constraints
|
||||
------------------
|
||||
* Standard library only; no new dependencies.
|
||||
* Every hook is defensive — a logging hook must NEVER raise (a raising hook
|
||||
would itself become a new silent-failure mode).
|
||||
* Diagnostics are written **directly to the stderr stream** (which the
|
||||
bootstrap redirects into the WebUI log file) *and* mirrored through the
|
||||
``logging`` module. The WebUI configures no logging handlers, so an
|
||||
``INFO``/``ERROR`` record would otherwise be swallowed by logging's
|
||||
``lastResort`` filter (WARNING+ only) and never reach the log — the exact
|
||||
failure mode of #4633. The direct write guarantees the line lands.
|
||||
* ``install_crash_visibility()`` is idempotent and safe to call at import or
|
||||
startup time.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import atexit
|
||||
import faulthandler
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import sys
|
||||
import threading
|
||||
import traceback
|
||||
from typing import Optional, TextIO
|
||||
|
||||
__all__ = [
|
||||
"install_crash_visibility",
|
||||
"is_installed",
|
||||
"thread_excepthook",
|
||||
"main_excepthook",
|
||||
]
|
||||
|
||||
_LOGGER = logging.getLogger("server")
|
||||
|
||||
# Guard so repeated calls (e.g. import + explicit startup call, or a test that
|
||||
# re-imports) don't stack duplicate hooks / atexit registrations.
|
||||
_INSTALLED = False
|
||||
_INSTALL_LOCK = threading.Lock()
|
||||
|
||||
# Strong reference to the stream faulthandler writes into. faulthandler keeps
|
||||
# the file descriptor, but holding the object prevents premature GC of a
|
||||
# wrapper and lets tests introspect what we enabled against.
|
||||
_FAULT_STREAM: Optional[TextIO] = None
|
||||
|
||||
# Remember prior hooks so we can chain (and so tests can restore them).
|
||||
_PREV_THREAD_HOOK = None
|
||||
_PREV_SYS_HOOK = None
|
||||
|
||||
|
||||
def is_installed() -> bool:
|
||||
"""Return True once :func:`install_crash_visibility` has run."""
|
||||
return _INSTALLED
|
||||
|
||||
|
||||
def _resolve_stream(stream: Optional[TextIO]) -> TextIO:
|
||||
"""Pick a real, writable stream, defaulting to stderr."""
|
||||
if stream is not None:
|
||||
return stream
|
||||
err = sys.stderr
|
||||
if err is not None:
|
||||
return err
|
||||
# As a last resort, wrap fd 2 directly so we can still emit something.
|
||||
return os.fdopen(os.dup(2), "w", closefd=True)
|
||||
|
||||
|
||||
def _direct_write(text: str) -> None:
|
||||
"""Write a diagnostic line straight to the fault stream. Never raises."""
|
||||
try:
|
||||
stream = _FAULT_STREAM if _FAULT_STREAM is not None else sys.stderr
|
||||
if stream is None:
|
||||
return
|
||||
stream.write(text if text.endswith("\n") else text + "\n")
|
||||
try:
|
||||
stream.flush()
|
||||
except Exception:
|
||||
pass
|
||||
except Exception:
|
||||
# Absolutely nothing a diagnostic writer may raise should propagate.
|
||||
pass
|
||||
|
||||
|
||||
def _emit(level: int, message: str, *, exc_info=None) -> None:
|
||||
"""Emit a diagnostic via logging AND directly to the fault stream.
|
||||
|
||||
Never raises. The direct write is what guarantees visibility in the WebUI
|
||||
log (which configures no logging handlers); the logging call is what lets
|
||||
tests/caplog and any future handler capture the same record.
|
||||
"""
|
||||
text = message
|
||||
if exc_info is not None:
|
||||
try:
|
||||
etype, evalue, etb = exc_info
|
||||
tb_text = "".join(traceback.format_exception(etype, evalue, etb))
|
||||
text = f"{message}\n{tb_text}".rstrip()
|
||||
except Exception:
|
||||
text = message
|
||||
_direct_write(text)
|
||||
try:
|
||||
_LOGGER.log(level, message, exc_info=exc_info)
|
||||
except Exception:
|
||||
# If the logging subsystem itself blows up, the direct write above
|
||||
# already captured the diagnostic — swallow so we never re-crash here.
|
||||
pass
|
||||
|
||||
|
||||
def thread_excepthook(args) -> None:
|
||||
"""threading.excepthook: log any uncaught exception in a daemon/handler thread.
|
||||
|
||||
``args`` is a ``threading.ExceptHookArgs`` (exc_type, exc_value,
|
||||
exc_traceback, thread). A normal thread shutdown may pass exc_type=None.
|
||||
"""
|
||||
try:
|
||||
exc_type = getattr(args, "exc_type", None)
|
||||
exc_value = getattr(args, "exc_value", None)
|
||||
exc_tb = getattr(args, "exc_traceback", None)
|
||||
thread = getattr(args, "thread", None)
|
||||
|
||||
# SystemExit inside a thread is an intentional stop, not a crash.
|
||||
if exc_type is None or issubclass(exc_type, SystemExit):
|
||||
return
|
||||
|
||||
thread_name = getattr(thread, "name", None) or "unknown"
|
||||
thread_ident = getattr(thread, "ident", None)
|
||||
daemon = getattr(thread, "daemon", None)
|
||||
_emit(
|
||||
logging.ERROR,
|
||||
"[crash-visibility] uncaught exception in thread "
|
||||
"name=%s ident=%s daemon=%s type=%s"
|
||||
% (thread_name, thread_ident, daemon, getattr(exc_type, "__name__", exc_type)),
|
||||
exc_info=(exc_type, exc_value, exc_tb),
|
||||
)
|
||||
except Exception:
|
||||
# A hook that raises would re-introduce the silent-death class of bug.
|
||||
try:
|
||||
_direct_write("[crash-visibility] thread_excepthook failed to log an exception")
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
# Chain to any previously-installed hook, but only if it isn't the
|
||||
# stock one (which just prints to stderr again — we already did better).
|
||||
prev = _PREV_THREAD_HOOK
|
||||
try:
|
||||
if prev is not None and prev is not threading.__excepthook__:
|
||||
prev(args)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def main_excepthook(exc_type, exc_value, exc_tb) -> None:
|
||||
"""sys.excepthook: log any uncaught exception on the main thread."""
|
||||
try:
|
||||
is_keyboard_interrupt = bool(
|
||||
isinstance(exc_type, type) and issubclass(exc_type, KeyboardInterrupt)
|
||||
)
|
||||
if is_keyboard_interrupt:
|
||||
# Preserve default Ctrl-C behaviour: no scary traceback dump.
|
||||
prev = _PREV_SYS_HOOK or sys.__excepthook__
|
||||
try:
|
||||
prev(exc_type, exc_value, exc_tb)
|
||||
except Exception:
|
||||
pass
|
||||
return
|
||||
_emit(
|
||||
logging.CRITICAL,
|
||||
"[crash-visibility] uncaught exception on main thread type=%s"
|
||||
% (getattr(exc_type, "__name__", exc_type),),
|
||||
exc_info=(exc_type, exc_value, exc_tb),
|
||||
)
|
||||
except Exception:
|
||||
try:
|
||||
_direct_write("[crash-visibility] main_excepthook failed to log an exception")
|
||||
except Exception:
|
||||
pass
|
||||
finally:
|
||||
prev = _PREV_SYS_HOOK
|
||||
try:
|
||||
if prev is not None and prev is not sys.__excepthook__:
|
||||
prev(exc_type, exc_value, exc_tb)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _exit_audit() -> None:
|
||||
"""atexit hook: record that the process is exiting.
|
||||
|
||||
A plain interpreter exit (return from main, sys.exit, or the tail of a
|
||||
fatal error that still unwinds atexit) leaves a breadcrumb so we can tell a
|
||||
*clean* shutdown apart from a truly silent death — if the log ends with no
|
||||
exit-audit line at all, the process was killed without unwinding
|
||||
(OOM kill / SIGKILL / native abort), which itself narrows the diagnosis.
|
||||
"""
|
||||
try:
|
||||
_emit(
|
||||
logging.INFO,
|
||||
"[crash-visibility] process exit pid=%s thread=%s"
|
||||
% (os.getpid(), threading.current_thread().name),
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def install_crash_visibility(
|
||||
*,
|
||||
stream: Optional[TextIO] = None,
|
||||
enable_faulthandler: bool = True,
|
||||
register_sigusr1_dump: bool = True,
|
||||
) -> bool:
|
||||
"""Install faulthandler + excepthooks + exit audit. Idempotent.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
stream:
|
||||
Where diagnostics are written. Defaults to ``sys.stderr`` (redirected
|
||||
into the WebUI log file by the bootstrap).
|
||||
enable_faulthandler:
|
||||
Enable ``faulthandler`` so a native segfault/abort dumps a C-level
|
||||
traceback of every thread instead of vanishing.
|
||||
register_sigusr1_dump:
|
||||
On POSIX, register ``SIGUSR1`` to dump all thread stacks on demand —
|
||||
invaluable for diagnosing a *hang* (send ``kill -USR1 <pid>``) as
|
||||
opposed to a crash. Ignored where SIGUSR1 is unavailable (Windows).
|
||||
|
||||
Returns True if it installed on this call, False if it was already active.
|
||||
"""
|
||||
global _INSTALLED, _FAULT_STREAM, _PREV_THREAD_HOOK, _PREV_SYS_HOOK
|
||||
|
||||
with _INSTALL_LOCK:
|
||||
if _INSTALLED:
|
||||
return False
|
||||
|
||||
_FAULT_STREAM = _resolve_stream(stream)
|
||||
|
||||
# 1) faulthandler: native crash → C-level traceback of all threads.
|
||||
if enable_faulthandler:
|
||||
try:
|
||||
faulthandler.enable(file=_FAULT_STREAM, all_threads=True)
|
||||
except Exception:
|
||||
# Some frozen/embedded builds restrict faulthandler; degrade
|
||||
# gracefully rather than blocking startup.
|
||||
_direct_write("[crash-visibility] faulthandler.enable() failed")
|
||||
|
||||
# 1b) On-demand all-thread stack dump for diagnosing hangs.
|
||||
if register_sigusr1_dump:
|
||||
sigusr1 = getattr(signal, "SIGUSR1", None)
|
||||
if sigusr1 is not None:
|
||||
try:
|
||||
faulthandler.register(
|
||||
sigusr1, file=_FAULT_STREAM, all_threads=True, chain=True
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 2) threading.excepthook: daemon/handler-thread crash → logged.
|
||||
try:
|
||||
_PREV_THREAD_HOOK = threading.excepthook
|
||||
threading.excepthook = thread_excepthook
|
||||
except Exception:
|
||||
_direct_write("[crash-visibility] failed to install threading.excepthook")
|
||||
|
||||
# 3) sys.excepthook: main-thread crash → logged.
|
||||
try:
|
||||
_PREV_SYS_HOOK = sys.excepthook
|
||||
sys.excepthook = main_excepthook
|
||||
except Exception:
|
||||
_direct_write("[crash-visibility] failed to install sys.excepthook")
|
||||
|
||||
# 4) atexit exit audit: breadcrumb on clean/unwound exit.
|
||||
try:
|
||||
atexit.register(_exit_audit)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
_INSTALLED = True
|
||||
return True
|
||||
|
|
@ -112,6 +112,7 @@ from api.profiles import set_request_profile, clear_request_profile
|
|||
from api.routes import handle_delete, handle_get, handle_patch, handle_post, handle_put
|
||||
from api.startup import auto_install_agent_deps, fix_credential_permissions
|
||||
from api.updates import WEBUI_VERSION
|
||||
from api.crash_visibility import install_crash_visibility
|
||||
|
||||
|
||||
class QuietHTTPServer(ThreadingHTTPServer):
|
||||
|
|
@ -546,6 +547,12 @@ def _abort_if_already_serving(host: str, port: int) -> None:
|
|||
def main() -> None:
|
||||
from api.config import print_startup_config, verify_hermes_imports, _HERMES_FOUND
|
||||
|
||||
# Crash visibility FIRST (issue #4633): enable faulthandler + excepthooks +
|
||||
# exit audit before any heavy startup work so a native crash or a daemon /
|
||||
# handler-thread exception during startup or serving produces a diagnostic
|
||||
# instead of a silent death. The paired memory root-cause is #4765.
|
||||
install_crash_visibility()
|
||||
|
||||
print_startup_config()
|
||||
|
||||
fd_limit = _raise_fd_soft_limit()
|
||||
|
|
|
|||
254
tests/test_issue4633_crash_visibility.py
Normal file
254
tests/test_issue4633_crash_visibility.py
Normal file
|
|
@ -0,0 +1,254 @@
|
|||
"""Regression coverage for issue #4633: crash visibility hardening.
|
||||
|
||||
server.py was exiting SILENTLY after 9-16h — no traceback, no shutdown-audit
|
||||
line, no core dump; the log just stopped mid-request. It runs a
|
||||
ThreadingHTTPServer with daemon_threads=True, so an unhandled exception in a
|
||||
request/SSE handler thread could terminate work with nothing recorded, and
|
||||
faulthandler was not enabled so a native crash left nothing at all.
|
||||
|
||||
These tests assert the diagnostic hardening in api/crash_visibility.py (wired
|
||||
into server.main()):
|
||||
|
||||
* faulthandler is enabled after install (native crash → C-level traceback);
|
||||
* threading.excepthook is installed and a real daemon-thread exception is
|
||||
logged (thread name + traceback) instead of vanishing;
|
||||
* sys.excepthook is installed and logs uncaught main-thread exceptions;
|
||||
* an atexit exit-audit breadcrumb is registered;
|
||||
* server.py wires install_crash_visibility() into main().
|
||||
|
||||
The paired MEMORY root-cause is fixed separately in #4765; this is purely
|
||||
crash VISIBILITY.
|
||||
|
||||
No live server socket is required — the helpers are exercised directly.
|
||||
"""
|
||||
import faulthandler
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
import api.crash_visibility as cv
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_hooks(tmp_path):
|
||||
"""Reset crash-visibility global state and restore process hooks after.
|
||||
|
||||
Installing crash visibility mutates the process-wide threading.excepthook,
|
||||
sys.excepthook and faulthandler state, so snapshot and restore them to keep
|
||||
the rest of the suite hermetic.
|
||||
"""
|
||||
saved_thread_hook = threading.excepthook
|
||||
saved_sys_hook = sys.excepthook
|
||||
saved_installed = cv._INSTALLED
|
||||
saved_stream = cv._FAULT_STREAM
|
||||
saved_prev_thread = cv._PREV_THREAD_HOOK
|
||||
saved_prev_sys = cv._PREV_SYS_HOOK
|
||||
was_fh_enabled = faulthandler.is_enabled()
|
||||
saved_atexit_register = cv.atexit.register
|
||||
|
||||
# Start from stock hooks so the module records the *stock* hook as "prev"
|
||||
# and never chains into pytest's own thread-exception collector during the
|
||||
# test (which would otherwise double-report and can fail the test).
|
||||
threading.excepthook = threading.__excepthook__
|
||||
sys.excepthook = sys.__excepthook__
|
||||
cv._INSTALLED = False
|
||||
cv._FAULT_STREAM = None
|
||||
cv._PREV_THREAD_HOOK = None
|
||||
cv._PREV_SYS_HOOK = None
|
||||
# Prevent tests from leaking real atexit handlers into the pytest session:
|
||||
# the fixture resets _INSTALLED each test, so every install() would register
|
||||
# a fresh (real) _exit_audit that fires at interpreter shutdown. Capture
|
||||
# them instead. The dedicated exit-audit test re-patches this itself.
|
||||
cv.atexit.register = lambda fn: fn
|
||||
|
||||
stream_path = tmp_path / "faultstream.log"
|
||||
stream = stream_path.open("w", encoding="utf-8")
|
||||
try:
|
||||
yield stream, stream_path
|
||||
finally:
|
||||
try:
|
||||
stream.flush()
|
||||
except Exception:
|
||||
pass
|
||||
# Detach faulthandler from the about-to-close temp file before closing.
|
||||
try:
|
||||
faulthandler.disable()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
stream.close()
|
||||
except Exception:
|
||||
pass
|
||||
if was_fh_enabled:
|
||||
try:
|
||||
faulthandler.enable()
|
||||
except Exception:
|
||||
pass
|
||||
threading.excepthook = saved_thread_hook
|
||||
sys.excepthook = saved_sys_hook
|
||||
cv._INSTALLED = saved_installed
|
||||
cv._FAULT_STREAM = saved_stream
|
||||
cv._PREV_THREAD_HOOK = saved_prev_thread
|
||||
cv._PREV_SYS_HOOK = saved_prev_sys
|
||||
cv.atexit.register = saved_atexit_register
|
||||
|
||||
|
||||
def _read(path):
|
||||
try:
|
||||
return path.read_text(encoding="utf-8")
|
||||
except Exception:
|
||||
return ""
|
||||
|
||||
|
||||
# ── 1. faulthandler ─────────────────────────────────────────────────────────
|
||||
def test_faulthandler_enabled_after_install(fresh_hooks):
|
||||
stream, _ = fresh_hooks
|
||||
|
||||
installed = cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
|
||||
assert installed is True
|
||||
assert cv.is_installed() is True
|
||||
# faulthandler must be active so a native segfault/abort dumps a C-level
|
||||
# traceback of every thread instead of vanishing (the #4633 failure mode).
|
||||
assert faulthandler.is_enabled() is True
|
||||
|
||||
|
||||
def test_install_is_idempotent(fresh_hooks):
|
||||
stream, _ = fresh_hooks
|
||||
assert cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False) is True
|
||||
# Second call must be a no-op (no stacked hooks / duplicate atexit handlers).
|
||||
assert cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False) is False
|
||||
|
||||
|
||||
# ── 2. threading.excepthook ─────────────────────────────────────────────────
|
||||
def test_thread_excepthook_installed(fresh_hooks):
|
||||
stream, _ = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
assert threading.excepthook is cv.thread_excepthook
|
||||
|
||||
|
||||
def test_daemon_thread_exception_is_logged(fresh_hooks, caplog):
|
||||
stream, stream_path = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
|
||||
def _boom():
|
||||
raise ValueError("simulated SSE handler crash 4633")
|
||||
|
||||
caplog.set_level(logging.ERROR, logger="server")
|
||||
|
||||
t = threading.Thread(target=_boom, name="sse-handler-42", daemon=True)
|
||||
t.start()
|
||||
t.join(timeout=5)
|
||||
assert not t.is_alive()
|
||||
|
||||
stream.flush()
|
||||
written = _read(stream_path)
|
||||
logged = "\n".join(r.getMessage() for r in caplog.records)
|
||||
|
||||
# Diagnostic must land in the fault stream (which the bootstrap redirects
|
||||
# into the WebUI log file) AND be visible via the logging subsystem.
|
||||
for haystack in (written, logged):
|
||||
assert "[crash-visibility] uncaught exception in thread" in haystack
|
||||
assert "sse-handler-42" in haystack
|
||||
# The stream copy carries the full traceback + the original exception text.
|
||||
assert "ValueError" in written
|
||||
assert "simulated SSE handler crash 4633" in written
|
||||
|
||||
|
||||
def test_thread_excepthook_ignores_clean_exit_and_never_raises(fresh_hooks):
|
||||
stream, stream_path = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
|
||||
# A normal thread teardown passes exc_type=None; must produce nothing and
|
||||
# must not raise.
|
||||
Args = type(
|
||||
"Args",
|
||||
(),
|
||||
{"exc_type": None, "exc_value": None, "exc_traceback": None, "thread": None},
|
||||
)
|
||||
cv.thread_excepthook(Args())
|
||||
|
||||
# Wholly malformed args must also be swallowed without raising.
|
||||
cv.thread_excepthook(object())
|
||||
|
||||
stream.flush()
|
||||
assert "uncaught exception in thread" not in _read(stream_path)
|
||||
|
||||
|
||||
# ── 3. sys.excepthook ───────────────────────────────────────────────────────
|
||||
def test_sys_excepthook_installed(fresh_hooks):
|
||||
stream, _ = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
assert sys.excepthook is cv.main_excepthook
|
||||
|
||||
|
||||
def test_main_thread_exception_is_logged(fresh_hooks, caplog):
|
||||
stream, stream_path = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
|
||||
caplog.set_level(logging.CRITICAL, logger="server")
|
||||
try:
|
||||
raise RuntimeError("simulated main-thread fatal 4633")
|
||||
except RuntimeError:
|
||||
cv.main_excepthook(*sys.exc_info())
|
||||
|
||||
stream.flush()
|
||||
written = _read(stream_path)
|
||||
logged = "\n".join(r.getMessage() for r in caplog.records)
|
||||
|
||||
for haystack in (written, logged):
|
||||
assert "[crash-visibility] uncaught exception on main thread" in haystack
|
||||
assert "RuntimeError" in written
|
||||
assert "simulated main-thread fatal 4633" in written
|
||||
|
||||
|
||||
def test_keyboard_interrupt_not_treated_as_crash(fresh_hooks, caplog):
|
||||
stream, stream_path = fresh_hooks
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
|
||||
caplog.set_level(logging.DEBUG, logger="server")
|
||||
try:
|
||||
raise KeyboardInterrupt()
|
||||
except KeyboardInterrupt:
|
||||
cv.main_excepthook(*sys.exc_info())
|
||||
|
||||
stream.flush()
|
||||
# Ctrl-C is intentional; it must not be logged as an uncaught crash.
|
||||
assert "uncaught exception on main thread" not in _read(stream_path)
|
||||
assert all("crash-visibility" not in r.getMessage() for r in caplog.records)
|
||||
|
||||
|
||||
# ── 4. exit audit ───────────────────────────────────────────────────────────
|
||||
def test_exit_audit_registered_and_emits(fresh_hooks, monkeypatch, caplog):
|
||||
stream, stream_path = fresh_hooks
|
||||
|
||||
registered = []
|
||||
monkeypatch.setattr(cv.atexit, "register", lambda fn: registered.append(fn) or fn)
|
||||
|
||||
cv.install_crash_visibility(stream=stream, register_sigusr1_dump=False)
|
||||
assert cv._exit_audit in registered
|
||||
|
||||
caplog.set_level(logging.INFO, logger="server")
|
||||
cv._exit_audit() # simulate interpreter exit unwinding
|
||||
|
||||
stream.flush()
|
||||
written = _read(stream_path)
|
||||
logged = "\n".join(r.getMessage() for r in caplog.records)
|
||||
for haystack in (written, logged):
|
||||
assert "[crash-visibility] process exit" in haystack
|
||||
|
||||
|
||||
# ── 5. server.py wiring ─────────────────────────────────────────────────────
|
||||
def test_server_imports_and_calls_install(fresh_hooks):
|
||||
import server
|
||||
|
||||
assert hasattr(server, "install_crash_visibility")
|
||||
assert server.install_crash_visibility is cv.install_crash_visibility
|
||||
|
||||
from tests.conftest import SERVER_SCRIPT
|
||||
|
||||
src = SERVER_SCRIPT.read_text(encoding="utf-8")
|
||||
assert "from api.crash_visibility import install_crash_visibility" in src
|
||||
assert "install_crash_visibility()" in src
|
||||
Loading…
Reference in a new issue