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

"""
Tests for cleanup_db_session teardown behavior in apply_middleware().

Source: app_factory.py lines 547-596.
Tests the teardown_appcontext handler that cleans up DB sessions,
thread-local sessions, and sweeps dead engines after each request.
"""

import pytest
from unittest.mock import patch, Mock

from flask import g


@pytest.fixture
def app():
    """Create a minimal Flask app with the teardown handler registered."""
    from local_deep_research.web.app_factory import create_app

    with patch("local_deep_research.web.app_factory.SocketIOService"):
        app, _ = create_app()
        app.config["TESTING"] = True
        app.config["WTF_CSRF_ENABLED"] = False
        return app


class TestCleanupDbSession:
    """Tests for the cleanup_db_session teardown handler."""

    def test_normal_cleanup_rollback_and_close(self, app):
        """Session in g.db_session gets rollback() and close() called."""
        mock_session = Mock()

        with app.test_request_context("/"):
            g.db_session = mock_session

        # After exiting context, teardown runs
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()

    def test_no_session_in_g_no_error(self, app):
        """When g has no db_session, teardown completes without error."""
        # audit: PUNCHLIST reviewed 2026-05 — KEEP (ASSERT_TRUE).
        with app.test_request_context("/"):
            # Don't set g.db_session
            pass
        # Should not raise

    def test_rollback_fails_close_still_called(self, app):
        """If rollback() raises, close() is still called."""
        mock_session = Mock()
        mock_session.rollback.side_effect = RuntimeError("rollback failed")

        with app.test_request_context("/"):
            g.db_session = mock_session

        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()

    def test_close_fails_no_propagation(self, app):
        """If close() raises, the exception does not propagate."""
        mock_session = Mock()
        mock_session.close.side_effect = RuntimeError("close failed")

        with app.test_request_context("/"):
            g.db_session = mock_session

        # Should not raise despite close() failure
        mock_session.rollback.assert_called_once()
        mock_session.close.assert_called_once()

    def test_cleanup_dead_threads_called(self, app):
        """cleanup_dead_threads is called during teardown."""
        mock_cleanup = Mock()

        with patch(
            "local_deep_research.database.thread_local_session.cleanup_dead_threads",
            mock_cleanup,
        ):
            with app.test_request_context("/"):
                pass

        mock_cleanup.assert_called_once()

    def test_cleanup_current_thread_called(self, app):
        """cleanup_current_thread is called during teardown for defense-in-depth."""
        mock_cleanup = Mock()

        with patch(
            "local_deep_research.database.thread_local_session.cleanup_current_thread",
            mock_cleanup,
        ):
            with app.test_request_context("/"):
                pass

        mock_cleanup.assert_called_once()