learningcircuit--local-deep-research/tests/web/test_thread_excepthook.py

"""Tests for the daemon-thread excepthook.

Covers:
- _install_thread_excepthook logs uncaught exceptions from daemon threads.
- _perform_post_login_tasks wraps its body so an otherwise-uncaught exception
  from inside the `with` context managers is logged instead of disappearing.
"""

import threading
from unittest.mock import MagicMock, patch


def test_install_thread_excepthook_logs_uncaught_exception():
    """After installing the hook, an uncaught exception on a daemon thread
    must be logged at ERROR level including the exception type and the
    thread name. Without this, silent crashes in the queue processor
    or APScheduler jobs are the mechanism by which the login path
    gradually starves.
    """
    from local_deep_research.web.app import _install_thread_excepthook

    previous = threading.excepthook
    _install_thread_excepthook()
    try:
        with patch("local_deep_research.web.app.logger") as mock_logger:

            def raiser():
                raise RuntimeError("boom from daemon")

            t = threading.Thread(target=raiser, name="test-daemon", daemon=True)
            t.start()
            t.join(timeout=2.0)

            assert not t.is_alive()
            assert mock_logger.error.called, "excepthook did not log"

            logged = mock_logger.error.call_args[0][0]
            assert "test-daemon" in logged
            assert "RuntimeError" in logged
            assert "boom from daemon" in logged
    finally:
        threading.excepthook = previous


def test_perform_post_login_tasks_catches_outer_exceptions():
    """If the outer structure of _perform_post_login_tasks itself raises
    (for example from inside a `with` context manager's __enter__) —
    not inside a per-step try/except — the wrapper must log the
    exception and not let the daemon thread die silently.
    """
    from local_deep_research.web.auth import routes

    with (
        patch.object(
            routes,
            "_perform_post_login_tasks_body",
            side_effect=RuntimeError("outer failure"),
        ),
        patch.object(routes, "logger") as mock_logger,
    ):
        # Must not raise.
        routes._perform_post_login_tasks("alice", "pw")

        assert mock_logger.exception.called
        msg = mock_logger.exception.call_args[0][0]
        assert "alice" in msg
        assert "crashed" in msg.lower() or "post-login" in msg.lower()


def test_perform_post_login_tasks_body_runs_when_no_outer_error():
    """Positive path: when the body succeeds, the wrapper forwards to
    it exactly once with the right args, and does not log any
    crash.
    """
    from local_deep_research.web.auth import routes

    body = MagicMock()
    with (
        patch.object(routes, "_perform_post_login_tasks_body", body),
        patch.object(routes, "logger") as mock_logger,
    ):
        routes._perform_post_login_tasks("bob", "pw")

    body.assert_called_once_with("bob", "pw")
    mock_logger.exception.assert_not_called()