learningcircuit--local-deep-research/tests/database/test_session_context_rollback.py

"""Regression tests: ``get_user_db_session`` rolls back the reused thread-local
session when the caller's ``with`` block raises.

Root cause behind the family of bare-``session.commit()`` cascade bugs: the
session yielded by ``get_user_db_session`` is a *reused* thread-local session
that is never closed on exit. Before this fix, an exception escaping the
``with`` block — most commonly a failed ``commit()``/``flush()`` — left the
session in ``PendingRollbackError`` state, and the *next* operation on that
thread cascaded. The context manager now rolls the session back on exception
exit and re-raises, so a single unguarded block can no longer poison the whole
thread.
"""

from unittest.mock import Mock, patch

import pytest
from sqlalchemy import create_engine
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker

from local_deep_research.database.models import Base
from local_deep_research.database.models.library import SourceType

import local_deep_research.database.session_context as _sc_mod

SC = "local_deep_research.database.session_context"
TLS = "local_deep_research.database.thread_local_session.get_metrics_session"

# Captured at module import — before any test runs, so before a sibling test
# (e.g. the rag-route suites) can leak a patch over this attribute that its
# teardown failed to restore. The autouse fixture below pins it back so these
# tests, which exercise the real context manager, are order-independent.
_REAL_GET_USER_DB_SESSION = _sc_mod.get_user_db_session


class TestGetUserDbSessionRollback:
    """Rollback-on-exception behaviour of the get_user_db_session context manager."""

    @pytest.fixture(autouse=True)
    def _isolate_session_context(self):
        """Make these tests immune to leaked global state from sibling suites.

        Two guards:
        - Restore the real ``get_user_db_session`` (a rag-route test leaks a
          MagicMock over it that its teardown fails to clear; without this our
          ``with get_user_db_session(...)`` would yield that mock, not our
          injected session).
        - Force ``get_g_db_session`` to None so we deterministically exercise
          the thread-local branch regardless of any leaked ``g.current_user``.
        """
        with (
            patch.object(
                _sc_mod, "get_user_db_session", _REAL_GET_USER_DB_SESSION
            ),
            patch.object(_sc_mod, "get_g_db_session", return_value=None),
        ):
            yield

    def test_real_thread_local_session_recovered_after_failed_commit(self, app):
        """End-to-end proof with a real SQLite session: a failed commit in one
        ``with`` block must not poison the next block on the same thread-local
        session. Without the fix, block 2's commit raises PendingRollbackError.
        """
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        real_session = sessionmaker(bind=engine)()
        # Pre-existing row so the duplicate-PK insert below is a *real*
        # constraint failure at commit time.
        real_session.add(SourceType(id="dup", name="a", display_name="A"))
        real_session.commit()

        with app.test_request_context():
            with patch(f"{SC}.db_manager") as mock_db:
                mock_db.has_encryption = False
                with patch(TLS, return_value=real_session):
                    # Block 1: a bare commit fails on a duplicate primary key.
                    with pytest.raises(IntegrityError):
                        with get_user_db_session(
                            username="u", password="p"
                        ) as s:
                            s.add(
                                SourceType(id="dup", name="b", display_name="B")
                            )
                            s.commit()

                    # Block 2: the SAME thread-local session must be usable.
                    with get_user_db_session(username="u", password="p") as s:
                        s.add(SourceType(id="ok", name="ok", display_name="OK"))
                        s.commit()

        assert real_session.query(SourceType).filter_by(id="ok").count() == 1
        real_session.close()
        engine.dispose()

    def test_clean_exit_does_not_roll_back(self, app):
        """Normal (non-exception) exit must NOT roll back — intentional pending
        state is the caller's to commit; we only recover on error.
        """
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        with app.test_request_context():
            with patch(f"{SC}.db_manager") as mock_db:
                mock_db.has_encryption = False
                mock_session = Mock()
                with patch(TLS, return_value=mock_session):
                    with get_user_db_session(username="u", password="p"):
                        pass

        mock_session.rollback.assert_not_called()

    def test_secondary_rollback_failure_does_not_mask_original_error(self, app):
        """If the recovery rollback itself raises, the ORIGINAL error must
        still propagate (safe_rollback swallows the rollback failure).
        """
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        with app.test_request_context():
            with patch(f"{SC}.db_manager") as mock_db:
                mock_db.has_encryption = False
                mock_session = Mock()
                mock_session.rollback.side_effect = RuntimeError(
                    "rollback boom"
                )
                with patch(TLS, return_value=mock_session):
                    with pytest.raises(ValueError, match="original"):
                        with get_user_db_session(username="u", password="p"):
                            raise ValueError("original")

        mock_session.rollback.assert_called_once()