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

"""Deep coverage tests for database/session_context.py.

Focuses on uncovered branches:
- get_user_db_session: g.db_session reuse path
- get_user_db_session: g.user_password path
- get_user_db_session: session_password_store path
- get_user_db_session: thread context password path
- get_user_db_session: unencrypted DB placeholder
- get_user_db_session: failed session raises DatabaseSessionError
- ensure_db_session: user not in session (unauthenticated)
- ensure_db_session: db not connected + encrypted -> redirect
- ensure_db_session: db not connected + unencrypted -> reopen
- UNENCRYPTED_DB_PLACEHOLDER constant
- DatabaseAccessMixin.execute_with_db
"""

from unittest.mock import MagicMock, patch

import pytest

MODULE = "local_deep_research.database.session_context"


# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------


class TestConstants:
    def test_unencrypted_placeholder_is_string(self):
        from local_deep_research.database.session_context import (
            UNENCRYPTED_DB_PLACEHOLDER,
        )

        assert isinstance(UNENCRYPTED_DB_PLACEHOLDER, str)
        assert len(UNENCRYPTED_DB_PLACEHOLDER) > 0

    def test_database_session_error_is_exception(self):
        from local_deep_research.database.session_context import (
            DatabaseSessionError,
        )

        assert issubclass(DatabaseSessionError, Exception)


# ---------------------------------------------------------------------------
# get_user_db_session – happy paths
# ---------------------------------------------------------------------------


class TestGetUserDbSessionHappyPaths:
    def test_uses_g_db_session_when_available(self):
        """When g.db_session exists, it is yielded without opening a new session."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        mock_sess = MagicMock()
        mock_g = MagicMock()
        mock_g.db_session = mock_sess

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=True),
            patch(f"{MODULE}.g", mock_g),
        ):
            with get_user_db_session(username="alice") as sess:
                assert sess is mock_sess

    def test_unencrypted_db_uses_placeholder_password(  # DevSkim: ignore DS101155
        self,
    ):
        """When no password and db is unencrypted, placeholder password is used."""  # DevSkim: ignore DS101155
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        mock_sess = MagicMock()
        mock_g = MagicMock(spec=[])  # No db_session attribute

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=False),
            patch(f"{MODULE}.get_search_context", return_value=None),
            patch(f"{MODULE}.db_manager") as mock_db,
            patch(
                "local_deep_research.database.thread_local_session.thread_session_manager.get_session",
                return_value=mock_sess,
            ),
            patch(f"{MODULE}.g", mock_g),
        ):
            mock_db.has_encryption = False
            with get_user_db_session(username="alice") as sess:
                assert sess is mock_sess

    def test_password_from_thread_context(self):
        """Password is retrieved from thread search context."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        mock_sess = MagicMock()
        mock_g = MagicMock(spec=[])

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=False),
            patch(
                f"{MODULE}.get_search_context",
                return_value={"user_password": "thread_pw"},
            ),
            patch(f"{MODULE}.db_manager") as mock_db,
            patch(
                "local_deep_research.database.thread_local_session.thread_session_manager.get_session",
                return_value=mock_sess,
            ),
            patch(f"{MODULE}.g", mock_g),
        ):
            mock_db.has_encryption = True
            with get_user_db_session(username="alice") as sess:
                assert sess is mock_sess


# ---------------------------------------------------------------------------
# get_user_db_session – error paths
# ---------------------------------------------------------------------------


class TestGetUserDbSessionErrors:
    def test_no_username_raises(self):
        from local_deep_research.database.session_context import (
            get_user_db_session,
            DatabaseSessionError,
        )

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=False),
        ):
            with pytest.raises(
                DatabaseSessionError, match="No authenticated user"
            ):
                with get_user_db_session():
                    pass

    def test_encrypted_db_no_password_raises(self):
        from local_deep_research.database.session_context import (
            get_user_db_session,
            DatabaseSessionError,
        )

        mock_g = MagicMock(spec=[])

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=False),
            patch(f"{MODULE}.get_search_context", return_value=None),
            patch(f"{MODULE}.db_manager") as mock_db,
            patch(f"{MODULE}.g", mock_g),
        ):
            mock_db.has_encryption = True
            with pytest.raises(DatabaseSessionError, match="requires password"):
                with get_user_db_session(username="alice"):
                    pass

    def test_failed_session_raises(self):
        """When get_metrics_session returns None, DatabaseSessionError is raised."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
            DatabaseSessionError,
        )

        mock_g = MagicMock(spec=[])

        with (
            patch(f"{MODULE}.has_request_context", return_value=False),
            patch(f"{MODULE}.has_app_context", return_value=False),
            patch(f"{MODULE}.get_search_context", return_value=None),
            patch(f"{MODULE}.db_manager") as mock_db,
            patch(
                "local_deep_research.database.thread_local_session.thread_session_manager.get_session",
                return_value=None,
            ),
            patch(f"{MODULE}.g", mock_g),
        ):
            mock_db.has_encryption = False
            with pytest.raises(
                DatabaseSessionError, match="Could not establish session"
            ):
                with get_user_db_session(username="alice"):
                    pass


# ---------------------------------------------------------------------------
# with_user_database decorator
# ---------------------------------------------------------------------------


class TestWithUserDatabase:
    def test_injects_session_as_first_arg(self):
        from local_deep_research.database.session_context import (
            with_user_database,
        )

        mock_sess = MagicMock()

        @with_user_database
        def my_func(db_session, x, y=0):
            return (db_session, x + y)

        with patch(f"{MODULE}.get_user_db_session") as mock_ctx:
            mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_sess)
            mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
            result = my_func(10, y=5)

        assert result == (mock_sess, 15)

    def test_passes_username_and_password_from_kwargs(self):
        from local_deep_research.database.session_context import (
            with_user_database,
        )

        @with_user_database
        def my_func(db_session):
            return db_session

        mock_sess = MagicMock()
        with patch(f"{MODULE}.get_user_db_session") as mock_ctx:
            mock_ctx.return_value.__enter__ = MagicMock(return_value=mock_sess)
            mock_ctx.return_value.__exit__ = MagicMock(return_value=False)
            result = my_func(_username="alice", _password="pw")

        mock_ctx.assert_called_once_with("alice", "pw")
        assert result is mock_sess


# ---------------------------------------------------------------------------
# DatabaseAccessMixin
# ---------------------------------------------------------------------------


class TestDatabaseAccessMixin:
    def test_get_db_session_raises_deprecation_warning(self):
        from local_deep_research.database.session_context import (
            DatabaseAccessMixin,
        )

        class MyService(DatabaseAccessMixin):
            pass

        svc = MyService()
        with pytest.raises(DeprecationWarning):
            svc.get_db_session()

    def test_get_db_session_message_content(self):
        from local_deep_research.database.session_context import (
            DatabaseAccessMixin,
        )

        svc = DatabaseAccessMixin()
        try:
            svc.get_db_session()
        except DeprecationWarning as e:
            assert "deprecated" in str(
                e
            ).lower() or "get_user_db_session" in str(e)