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

"""Tests for database/session_context.py."""

import pytest
from unittest.mock import Mock, patch
from flask import Flask


@pytest.fixture
def app():
    """Create test Flask application."""
    app = Flask(__name__)
    app.config["TESTING"] = True
    app.secret_key = "test-secret-key"
    return app


class TestDatabaseSessionError:
    """Tests for DatabaseSessionError exception."""

    def test_exception_can_be_raised(self):
        """Test that DatabaseSessionError can be raised."""
        from local_deep_research.database.session_context import (
            DatabaseSessionError,
        )

        with pytest.raises(DatabaseSessionError):
            raise DatabaseSessionError("Test error")

    def test_exception_message(self):
        """Test that exception preserves message."""
        from local_deep_research.database.session_context import (
            DatabaseSessionError,
        )

        try:
            raise DatabaseSessionError("Custom error message")
        except DatabaseSessionError as e:
            assert str(e) == "Custom error message"


class TestGetUserDbSession:
    """Tests for get_user_db_session context manager."""

    def test_raises_when_no_username_provided(self, app):
        """Test that error is raised when no username available."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
            DatabaseSessionError,
        )

        with app.test_request_context():
            # No username in session
            with pytest.raises(
                DatabaseSessionError, match="No authenticated user"
            ):
                with get_user_db_session():
                    pass

    def test_uses_provided_username(self, app):
        """Test that explicitly provided username is used."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
        )

        with app.test_request_context():
            with patch(
                "local_deep_research.database.session_context.db_manager"
            ) as mock_db:
                mock_db.has_encryption = False

                with patch(
                    "local_deep_research.database.thread_local_session.get_metrics_session"
                ) as mock_get_session:
                    mock_session = Mock()
                    mock_get_session.return_value = mock_session

                    with get_user_db_session(
                        username="testuser", password="testpass"
                    ) as session:
                        assert session is mock_session

    def test_uses_flask_session_username_when_not_provided(self, app):
        """Test that Flask session username is used when not explicitly provided."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
            UNENCRYPTED_DB_PLACEHOLDER,
        )

        with app.test_request_context():
            from flask import session as flask_session

            flask_session["username"] = "flask_user"

            with patch(
                "local_deep_research.database.session_context.db_manager"
            ) as mock_db:
                mock_db.has_encryption = False

                with patch(
                    "local_deep_research.database.thread_local_session.get_metrics_session"
                ) as mock_get_session:
                    mock_session = Mock()
                    mock_get_session.return_value = mock_session

                    with get_user_db_session() as _session:
                        mock_get_session.assert_called_once_with(
                            "flask_user", UNENCRYPTED_DB_PLACEHOLDER
                        )

    def test_raises_when_encrypted_db_requires_password(self, app):
        """Test error when encrypted DB accessed without password."""
        from local_deep_research.database.session_context import (
            get_user_db_session,
            DatabaseSessionError,
        )

        with app.test_request_context():
            from flask import session as flask_session

            flask_session["username"] = "testuser"

            with patch(
                "local_deep_research.database.session_context.db_manager"
            ) as mock_db:
                mock_db.has_encryption = True
                mock_db.connections = {}

                with patch(
                    "local_deep_research.database.session_context.get_search_context"
                ) as mock_ctx:
                    mock_ctx.return_value = None

                    with patch(
                        "local_deep_research.database.session_passwords.session_password_store"
                    ) as mock_store:
                        mock_store.get_session_password.return_value = None

                        with pytest.raises(
                            DatabaseSessionError, match="requires password"
                        ):
                            with get_user_db_session():
                                pass


class TestWithUserDatabase:
    """Tests for with_user_database decorator."""

    def test_decorator_injects_db_session(self, app):
        """Test that decorator injects db_session as first argument."""
        from local_deep_research.database.session_context import (
            with_user_database,
        )

        @with_user_database
        def test_func(db_session, arg1, arg2):
            return (db_session, arg1, arg2)

        with app.test_request_context():
            from flask import session as flask_session

            flask_session["username"] = "testuser"

            with patch(
                "local_deep_research.database.session_context.get_user_db_session"
            ) as mock_ctx:
                mock_session = Mock()
                mock_ctx.return_value.__enter__ = Mock(
                    return_value=mock_session
                )
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                result = test_func("value1", "value2")
                assert result == (mock_session, "value1", "value2")

    def test_decorator_passes_kwargs(self, app):
        """Test that decorator passes keyword arguments."""
        from local_deep_research.database.session_context import (
            with_user_database,
        )

        @with_user_database
        def test_func(db_session, key1=None, key2=None):
            return {"session": db_session, "key1": key1, "key2": key2}

        with app.test_request_context():
            from flask import session as flask_session

            flask_session["username"] = "testuser"

            with patch(
                "local_deep_research.database.session_context.get_user_db_session"
            ) as mock_ctx:
                mock_session = Mock()
                mock_ctx.return_value.__enter__ = Mock(
                    return_value=mock_session
                )
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                result = test_func(key1="a", key2="b")
                assert result["key1"] == "a"
                assert result["key2"] == "b"

    def test_decorator_extracts_special_kwargs(self, app):
        """Test that _username and _password are extracted from kwargs."""
        from local_deep_research.database.session_context import (
            with_user_database,
        )

        @with_user_database
        def test_func(db_session):
            return db_session

        with app.app_context():
            with patch(
                "local_deep_research.database.session_context.get_user_db_session"
            ) as mock_ctx:
                mock_session = Mock()
                mock_ctx.return_value.__enter__ = Mock(
                    return_value=mock_session
                )
                mock_ctx.return_value.__exit__ = Mock(return_value=False)

                test_func(_username="custom_user", _password="custom_pass")

                # Verify get_user_db_session was called with the custom credentials
                mock_ctx.assert_called_once_with("custom_user", "custom_pass")


class TestDatabaseAccessMixin:
    """Tests for DatabaseAccessMixin class."""

    def test_get_db_session_raises_deprecation_warning(self, app):
        """Test that get_db_session raises DeprecationWarning.

        The method was deprecated because it returned a closed session
        (context manager exits before returning).
        """
        from local_deep_research.database.session_context import (
            DatabaseAccessMixin,
        )
        import pytest

        class TestService(DatabaseAccessMixin):
            pass

        service = TestService()

        with pytest.raises(DeprecationWarning) as exc_info:
            service.get_db_session()

        assert "deprecated" in str(exc_info.value).lower()
        assert "get_user_db_session" in str(exc_info.value)


class TestUnencryptedDbPlaceholder:
    """Tests for UNENCRYPTED_DB_PLACEHOLDER constant."""

    def test_placeholder_value(self):
        """Test the placeholder constant value."""
        from local_deep_research.database.session_context import (
            UNENCRYPTED_DB_PLACEHOLDER,
        )

        assert UNENCRYPTED_DB_PLACEHOLDER == "unencrypted-mode"
        assert isinstance(UNENCRYPTED_DB_PLACEHOLDER, str)


class TestSafeRollback:
    """Tests for the safe_rollback helper introduced for issue #3827."""

    def test_calls_session_rollback(self):
        from local_deep_research.database.session_context import (
            safe_rollback,
        )

        session = Mock()
        safe_rollback(session, "ctx")
        session.rollback.assert_called_once()

    def test_swallows_rollback_exception(self):
        """Rollback errors must not propagate — call sites are exception
        handlers themselves and a raise here would mask the original error.
        """
        from local_deep_research.database.session_context import (
            safe_rollback,
        )

        session = Mock()
        session.rollback.side_effect = RuntimeError("simulated rollback fail")
        # Must NOT raise.
        safe_rollback(session, "ctx")
        session.rollback.assert_called_once()

    def test_works_without_context_argument(self):
        from local_deep_research.database.session_context import (
            safe_rollback,
        )

        session = Mock()
        safe_rollback(session)
        session.rollback.assert_called_once()