learningcircuit--local-deep-research/tests/web/auth/test_connection_cleanup_fallback.py

"""
Branch/defensive-path tests for connection_cleanup that exercise fallback
and error-handling paths not reached by test_connection_cleanup.py.

Targets uncovered lines in src/local_deep_research/web/auth/connection_cleanup.py:
- 56-68: _count_open_fds fallback via resource.getrlimit + os.fstat loop
- 107->113: scheduler.is_running == False branch
- 157-167: periodic pool dispose block (_last_dispose_time gate, engine.dispose
  loop with exception swallowed, disposed count log)
- 179-182: engine.pool.checkedout() raising inside FD monitor
- 190-195: high-FD (>800) warning and outer FD-monitor exception handler
"""

import datetime
from unittest.mock import MagicMock, patch

import pytest

from local_deep_research.web.auth import connection_cleanup as cc_module
from local_deep_research.web.auth.connection_cleanup import (
    cleanup_idle_connections,
)
from local_deep_research.web.auth.session_manager import SessionManager


@pytest.fixture
def sm():
    with patch(
        "local_deep_research.web.auth.session_manager.get_security_default",
        return_value=1,
    ):
        mgr = SessionManager()
    mgr.session_timeout = datetime.timedelta(seconds=1)
    mgr.remember_me_timeout = datetime.timedelta(seconds=2)
    return mgr


@pytest.fixture
def db():
    mock = MagicMock()
    mock.get_connected_usernames.return_value = set()
    mock.connections = {}
    # _connections_lock must be a context manager
    mock._connections_lock = MagicMock()
    mock._connections_lock.__enter__ = MagicMock(return_value=None)
    mock._connections_lock.__exit__ = MagicMock(return_value=None)
    return mock


@pytest.fixture(autouse=True)
def reset_dispose_timer():
    """Ensure _last_dispose_time doesn't leak across tests in this file.

    Set it far below any realistic time.monotonic() return so the
    elapsed-interval gate (>= _DISPOSE_INTERVAL_SECONDS, 1800s) is always
    satisfied by default. On freshly-booted systems time.monotonic() can
    still be below 1800s, so initializing to 0.0 isn't sufficient.
    """
    original = cc_module._last_dispose_time
    cc_module._last_dispose_time = -1e9
    yield
    cc_module._last_dispose_time = original


class TestCountOpenFdsFallback:
    """When /proc/self/fd is unavailable, _count_open_fds falls back to
    resource.getrlimit + os.fstat loop."""

    def test_falls_back_when_proc_fd_not_a_directory(self):
        """is_dir() returning False triggers the resource.getrlimit path."""
        from local_deep_research.web.auth.connection_cleanup import (
            _count_open_fds,
        )

        fake_limit = (5, 1024)
        valid_fds = {0, 1, 2, 4}

        def fake_fstat(fd):
            if fd in valid_fds:
                return MagicMock()
            raise OSError("bad fd")

        with (
            patch(
                "local_deep_research.web.auth.connection_cleanup.Path"
            ) as MockPath,
            patch(
                "local_deep_research.web.auth.connection_cleanup.os.fstat",
                side_effect=fake_fstat,
            ),
            patch("resource.getrlimit", return_value=fake_limit),
        ):
            mock_proc = MagicMock()
            mock_proc.is_dir.return_value = False
            MockPath.return_value = mock_proc

            result = _count_open_fds()

        assert result == len(valid_fds)

    def test_falls_back_when_proc_fd_iterdir_raises(self):
        """is_dir() True but iterdir() raising OSError falls through to fstat."""
        from local_deep_research.web.auth.connection_cleanup import (
            _count_open_fds,
        )

        fake_limit = (3, 1024)

        def fake_fstat(fd):
            if fd in (0, 1):
                return MagicMock()
            raise OSError("bad fd")

        with (
            patch(
                "local_deep_research.web.auth.connection_cleanup.Path"
            ) as MockPath,
            patch(
                "local_deep_research.web.auth.connection_cleanup.os.fstat",
                side_effect=fake_fstat,
            ),
            patch("resource.getrlimit", return_value=fake_limit),
        ):
            mock_proc = MagicMock()
            mock_proc.is_dir.return_value = True
            mock_proc.iterdir.side_effect = OSError("permission denied")
            MockPath.return_value = mock_proc

            result = _count_open_fds()

        assert result == 2


class TestSchedulerNotRunningBranch:
    """When scheduler.is_running is False, unregister_user is not called."""

    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_unregister_skipped_when_scheduler_not_running(
        self, _mock_research, mock_get_sched, sm, db
    ):
        mock_scheduler = MagicMock()
        mock_scheduler.is_running = False
        mock_get_sched.return_value = mock_scheduler

        db.get_connected_usernames.return_value = {"alice"}

        cleanup_idle_connections(sm, db)

        mock_scheduler.unregister_user.assert_not_called()
        db.close_user_database.assert_called_once_with("alice")


class TestPeriodicPoolDispose:
    """When the dispose interval has elapsed, all engines are disposed and
    individual dispose() exceptions are swallowed."""

    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_disposes_all_engines_and_swallows_errors(
        self, _mock_research, _mock_sched, sm, db
    ):
        # _last_dispose_time = 0 (from autouse fixture), so the interval
        # elapsed gate is satisfied and the dispose block runs.
        bad_engine = MagicMock()
        bad_engine.dispose.side_effect = RuntimeError("dispose failed")
        good_engine = MagicMock()

        db.connections = {"alice": bad_engine, "bob": good_engine}

        cleanup_idle_connections(sm, db)

        # Both dispose calls attempted despite the first raising.
        bad_engine.dispose.assert_called_once()
        good_engine.dispose.assert_called_once()
        # The timer was updated so a rapid second call would NOT re-dispose.
        assert cc_module._last_dispose_time > 0

    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_dispose_failures_surface_as_warnings(
        self, _mock_research, _mock_sched, sm, db
    ):
        """The pool-dispose workaround for ADR-0004's WAL/SHM handle leak
        depends on this loop succeeding. Pre-fix dispose() failures were
        logged at DEBUG, which hid silent drift if checkpoint/dispose
        repeatedly failed (disk pressure, lock starvation). Failures must
        surface at WARNING so an operator sees them, but only the
        exception TYPE NAME (no value) — the value can carry sensitive
        locals.
        """
        bad_engine = MagicMock()
        bad_engine.dispose.side_effect = RuntimeError(
            "potentially-sensitive details"
        )
        db.connections = {"alice": bad_engine}

        with patch.object(cc_module.logger, "warning") as mock_warn:
            cleanup_idle_connections(sm, db)

        # The failure path emitted a warning that names the user and the
        # exception TYPE, not the exception value.
        warning_calls = [str(call) for call in mock_warn.call_args_list]
        assert any(
            "alice" in call and "RuntimeError" in call for call in warning_calls
        )
        # The exception's message text must not appear in any warning —
        # only the type name is logged.
        assert not any(
            "potentially-sensitive details" in call for call in warning_calls
        )

    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_skip_dispose_when_interval_not_elapsed(
        self, _mock_research, _mock_sched, sm, db
    ):
        """If _last_dispose_time is recent, the dispose block is skipped."""
        import time

        cc_module._last_dispose_time = time.monotonic()  # just now
        engine = MagicMock()
        db.connections = {"alice": engine}

        cleanup_idle_connections(sm, db)

        engine.dispose.assert_not_called()


class TestFdMonitoringErrorPaths:
    """Exception handling inside and around the FD monitor block."""

    @patch(
        "local_deep_research.web.auth.connection_cleanup._count_open_fds",
        return_value=801,
    )
    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_high_fd_warning_and_checkedout_exception(
        self, _mock_research, _mock_sched, _mock_fd, sm, db
    ):
        """FD count > 800 emits a warning; engine.pool.checkedout() raising
        is swallowed silently."""
        flaky_engine = MagicMock()
        flaky_engine.pool.checkedout.side_effect = RuntimeError("pool gone")
        db.connections = {"alice": flaky_engine}

        with patch.object(cc_module.logger, "warning") as mock_warn:
            cleanup_idle_connections(sm, db)

        # Verify checkedout was attempted (the inner try path ran).
        flaky_engine.pool.checkedout.assert_called_once()
        # High-FD warning fired with the count.
        assert any(
            "801" in str(call) or "High FD" in str(call)
            for call in mock_warn.call_args_list
        )

    @patch(
        "local_deep_research.web.auth.connection_cleanup._count_open_fds",
        side_effect=RuntimeError("getrlimit unavailable"),
    )
    @patch(
        "local_deep_research.scheduler.background.get_background_job_scheduler",
    )
    @patch(
        "local_deep_research.web.auth.connection_cleanup.get_usernames_with_active_research",
        return_value=set(),
    )
    def test_outer_fd_monitor_exception_swallowed(
        self, _mock_research, _mock_sched, _mock_fd, sm, db
    ):
        """If _count_open_fds itself raises, cleanup still completes."""
        db.get_connected_usernames.return_value = set()
        # Should not propagate.
        cleanup_idle_connections(sm, db)