learningcircuit--local-deep-research/tests/web/utils/test_request_timing.py

"""Tests for the CI request-timing forensics middleware (#4431)."""

from unittest.mock import MagicMock

import pytest

from local_deep_research.web.utils.request_timing import (
    FREEZE_DUMP_SECONDS,
    RequestTimingMiddleware,
    SLOW_REQUEST_SECONDS,
    _should_arm_freeze_dump as _orig_should_arm_freeze_dump,
)


@pytest.fixture(autouse=True)
def _mock_faulthandler(monkeypatch):
    """Replace faulthandler so tests never arm a real repeating thread-dump
    timer (which would fire on stderr mid-suite), and force the pytest gate
    OPEN so the arming path is actually exercised. Returns the mock so a
    test can assert the dead-man's switch was armed."""
    from local_deep_research.web.utils import request_timing as mod

    fake = MagicMock()
    monkeypatch.setattr(mod, "faulthandler", fake)
    # _should_arm_freeze_dump() returns False under pytest; override so the
    # arming path runs in these tests.
    monkeypatch.setattr(mod, "_should_arm_freeze_dump", lambda: True)
    return fake


def _make_environ(method="GET", path="/chat/", query=""):
    return {
        "REQUEST_METHOD": method,
        "PATH_INFO": path,
        "QUERY_STRING": query,
    }


def _collect_logs(monkeypatch):
    """Capture loguru records emitted by the middleware module."""
    records = []
    from local_deep_research.web.utils import request_timing as mod

    class FakeLogger:
        def info(self, msg):
            records.append(("info", msg))

        def warning(self, msg):
            records.append(("warning", msg))

        def debug(self, msg):
            records.append(("debug", msg))

    monkeypatch.setattr(mod, "logger", FakeLogger())
    return records


def test_logs_arrival_and_completion(monkeypatch):
    records = _collect_logs(monkeypatch)

    def app(environ, start_response):
        return [b"ok"]

    mw = RequestTimingMiddleware(app)
    body = mw(_make_environ(), lambda *a: None)

    assert body == [b"ok"]
    assert records[0] == ("info", "[req] > GET /chat/")
    level, msg = records[1]
    assert level == "info"
    assert msg.startswith("[req] < GET /chat/")


def test_slow_request_logged_as_warning(monkeypatch):
    records = _collect_logs(monkeypatch)
    from local_deep_research.web.utils import request_timing as mod

    # Simulate a slow WSGI call without sleeping: jump the clock by
    # patching time.monotonic to advance past the slow threshold.
    ticks = iter([100.0, 100.0 + SLOW_REQUEST_SECONDS + 1])
    monkeypatch.setattr(mod.time, "monotonic", lambda: next(ticks))

    mw = RequestTimingMiddleware(lambda e, s: [b"ok"])
    mw(_make_environ(path="/slow"), lambda *a: None)

    level, msg = records[-1]
    assert level == "warning"
    assert "SLOW" in msg


def test_socketio_paths_keep_query_string(monkeypatch):
    records = _collect_logs(monkeypatch)

    mw = RequestTimingMiddleware(lambda e, s: [b"ok"])
    mw(
        _make_environ(path="/socket.io/", query="transport=polling&sid=abc"),
        lambda *a: None,
    )

    assert records[0] == (
        "info",
        "[req] > GET /socket.io/?transport=polling&sid=abc",
    )


def test_arrival_logged_even_when_app_raises(monkeypatch):
    records = _collect_logs(monkeypatch)

    def exploding_app(environ, start_response):
        raise RuntimeError("boom")

    mw = RequestTimingMiddleware(exploding_app)
    try:
        mw(_make_environ(path="/boom"), lambda *a: None)
    except RuntimeError:
        pass

    assert records[0] == ("info", "[req] > GET /boom")
    # completion line still emitted by the finally block
    assert any("[req] < GET /boom" in msg for _, msg in records[1:])


def test_arms_freeze_dump_on_init_and_rearms_per_request(
    monkeypatch, _mock_faulthandler
):
    _collect_logs(monkeypatch)

    # Construction arms the dead-man's switch.
    mw = RequestTimingMiddleware(lambda e, s: [b"ok"])
    _mock_faulthandler.enable.assert_called_once()
    assert _mock_faulthandler.dump_traceback_later.call_count == 1
    _, kwargs = _mock_faulthandler.dump_traceback_later.call_args
    assert _mock_faulthandler.dump_traceback_later.call_args.args[0] == (
        FREEZE_DUMP_SECONDS
    )
    assert kwargs.get("repeat") is True

    # Each request re-arms it (resets the countdown).
    mw(_make_environ(), lambda *a: None)
    assert _mock_faulthandler.dump_traceback_later.call_count == 2


def test_path_crlf_is_sanitized_against_log_injection(monkeypatch):
    records = _collect_logs(monkeypatch)

    mw = RequestTimingMiddleware(lambda e, s: [b"ok"])
    mw(_make_environ(path="/evil\n[req] > GET /forged"), lambda *a: None)

    arrival = records[0][1]
    assert "\n" not in arrival
    assert "\\n" in arrival
    assert arrival == "[req] > GET /evil\\n[req] > GET /forged"


def test_freeze_dump_failure_never_breaks_requests(monkeypatch):
    """If faulthandler raises, the request must still be served."""
    from local_deep_research.web.utils import request_timing as mod

    boom = MagicMock()
    boom.enable.side_effect = RuntimeError("no faulthandler")
    boom.dump_traceback_later.side_effect = RuntimeError("nope")
    monkeypatch.setattr(mod, "faulthandler", boom)
    monkeypatch.setattr(mod, "_should_arm_freeze_dump", lambda: True)
    _collect_logs(monkeypatch)

    mw = RequestTimingMiddleware(lambda e, s: [b"served"])
    assert mw(_make_environ(), lambda *a: None) == [b"served"]


def test_freeze_dump_not_armed_under_pytest(monkeypatch):
    """The dead-man's switch must be a no-op under pytest so it doesn't spew
    dumps across the test run; only the real server arms it."""
    from local_deep_research.web.utils import request_timing as mod

    fake = MagicMock()
    monkeypatch.setattr(mod, "faulthandler", fake)
    # Undo the autouse fixture's gate override; exercise the REAL gate,
    # which sees pytest in sys.modules and must suppress arming.
    monkeypatch.setattr(
        mod, "_should_arm_freeze_dump", _orig_should_arm_freeze_dump
    )
    assert mod._should_arm_freeze_dump() is False
    mw = RequestTimingMiddleware(lambda e, s: [b"ok"])
    mw(_make_environ(), lambda *a: None)
    fake.dump_traceback_later.assert_not_called()
    fake.enable.assert_not_called()