learningcircuit--local-deep-research/tests/utilities/test_db_utils.py

"""Tests for db_utils module."""

import threading
from unittest.mock import Mock, patch

import pytest


class TestGetDbSession:
    """Tests for get_db_session function."""

    def test_raises_error_from_background_thread_without_context(self):
        """Should raise error when called from background thread without app context."""
        from local_deep_research.utilities.db_utils import get_db_session

        error_raised = [False]
        error_message = [""]

        def background_task():
            try:
                # Ensure has_app_context returns False in this thread
                with patch(
                    "local_deep_research.utilities.db_utils.has_app_context",
                    return_value=False,
                ):
                    get_db_session()
            except RuntimeError as e:
                error_raised[0] = True
                error_message[0] = str(e)

        # Run in a background thread
        thread = threading.Thread(
            target=background_task, name="TestBackgroundThread"
        )
        thread.start()
        thread.join()

        assert error_raised[0], (
            "Should raise RuntimeError from background thread"
        )
        assert "background thread" in error_message[0].lower()

    def test_returns_session_when_username_provided(self):
        """Should return session when username is explicitly provided."""
        from local_deep_research.utilities.db_utils import get_db_session

        mock_session = Mock()

        with patch(
            "local_deep_research.utilities.db_utils.db_manager"
        ) as mock_manager:
            with patch(
                "local_deep_research.utilities.db_utils.has_app_context",
                return_value=True,
            ):
                mock_manager.get_session.return_value = mock_session

                result = get_db_session(username="testuser")

                assert result == mock_session
                mock_manager.get_session.assert_called_with("testuser")

    def test_raises_error_when_no_db_for_user(self):
        """Should raise error when no database found for user."""
        from local_deep_research.utilities.db_utils import get_db_session

        with patch(
            "local_deep_research.utilities.db_utils.db_manager"
        ) as mock_manager:
            with patch(
                "local_deep_research.utilities.db_utils.has_app_context",
                return_value=True,
            ):
                mock_manager.get_session.return_value = None

                with pytest.raises(RuntimeError, match="No database found"):
                    get_db_session(username="nonexistent")

    def test_returns_session_from_flask_g(self):
        """Should return session from Flask g object if present (via lazy helper)."""
        from local_deep_research.utilities.db_utils import get_db_session

        mock_session = Mock()

        with patch(
            "local_deep_research.database.session_context.get_g_db_session",
            return_value=mock_session,
        ):
            with patch(
                "local_deep_research.utilities.db_utils.has_app_context",
                return_value=True,
            ):
                with patch(
                    "local_deep_research.utilities.db_utils.flask_session",
                    {"username": None},
                ):
                    result = get_db_session()

                    assert result == mock_session

    def test_checks_flask_g_for_session(self):
        """Should check Flask g object for db_session."""
        # This tests that the function checks g.db_session
        from local_deep_research.utilities.db_utils import get_db_session

        mock_session = Mock()
        mock_g = Mock()
        mock_g.db_session = mock_session

        with patch(
            "local_deep_research.utilities.db_utils.has_app_context",
            return_value=True,
        ):
            with patch("local_deep_research.utilities.db_utils.g", mock_g):
                with patch(
                    "local_deep_research.utilities.db_utils.flask_session",
                    {"username": None},
                ):
                    result = get_db_session()
                    # Should return a session (the mock)
                    assert result is not None
                    # Verify g.db_session was accessed
                    assert hasattr(mock_g, "db_session")


class TestGetSettingsManager:
    """Tests for get_settings_manager function."""

    def test_returns_settings_manager(self):
        """Should return a SettingsManager instance."""
        from local_deep_research.utilities.db_utils import get_settings_manager

        mock_manager = Mock()

        with patch(
            "local_deep_research.utilities.db_utils.has_app_context",
            return_value=False,
        ):
            with patch(
                "local_deep_research.utilities.db_utils.get_db_session",
                side_effect=RuntimeError("No session"),
            ):
                with patch(
                    "local_deep_research.settings.SettingsManager",
                    return_value=mock_manager,
                ):
                    result = get_settings_manager()

                    # Should return a SettingsManager (mocked)
                    assert result is not None

    def test_uses_provided_db_session(self):
        """Should use provided db_session."""
        from local_deep_research.utilities.db_utils import get_settings_manager

        mock_session = Mock()
        mock_manager = Mock()

        with patch(
            "local_deep_research.settings.SettingsManager",
            return_value=mock_manager,
        ):
            result = get_settings_manager(db_session=mock_session)

            # The function should return a SettingsManager initialized with the session
            assert result is not None

    def test_handles_no_session(self):
        """Should handle case when no session available."""
        from local_deep_research.utilities.db_utils import get_settings_manager

        with patch(
            "local_deep_research.utilities.db_utils.has_app_context",
            return_value=False,
        ):
            with patch(
                "local_deep_research.utilities.db_utils.get_db_session",
                side_effect=RuntimeError("No session"),
            ):
                result = get_settings_manager()

                # Should still return a SettingsManager (with None session)
                assert result is not None


class TestNoDbSettings:
    """Tests for no_db_settings decorator."""

    def test_disables_db_during_function_execution(self):
        """Should disable DB session during function execution."""
        from local_deep_research.utilities.db_utils import no_db_settings

        mock_session = Mock()
        mock_manager = Mock()
        mock_manager.db_session = mock_session

        session_during_call = []

        @no_db_settings
        def test_func():
            session_during_call.append(mock_manager.db_session)
            return "result"

        with patch(
            "local_deep_research.utilities.db_utils.get_settings_manager",
            return_value=mock_manager,
        ):
            result = test_func()

            assert result == "result"
            assert session_during_call[0] is None
            # Session should be restored after
            assert mock_manager.db_session == mock_session

    def test_restores_db_on_exception(self):
        """Should restore DB session even if function raises."""
        from local_deep_research.utilities.db_utils import no_db_settings

        mock_session = Mock()
        mock_manager = Mock()
        mock_manager.db_session = mock_session

        @no_db_settings
        def failing_func():
            raise ValueError("Test error")

        with patch(
            "local_deep_research.utilities.db_utils.get_settings_manager",
            return_value=mock_manager,
        ):
            with pytest.raises(ValueError):
                failing_func()

            # Session should still be restored
            assert mock_manager.db_session == mock_session

    def test_preserves_function_metadata(self):
        """Should preserve function name and docstring."""
        from local_deep_research.utilities.db_utils import no_db_settings

        @no_db_settings
        def my_function():
            """My docstring."""
            pass

        assert my_function.__name__ == "my_function"
        assert my_function.__doc__ == "My docstring."


class TestThreadSafety:
    """Tests for thread safety mechanisms."""

    def test_thread_name_included_in_error(self):
        """Error message should include thread name."""
        from local_deep_research.utilities.db_utils import get_db_session

        error_message = [""]

        def background_task():
            try:
                # Ensure has_app_context returns False so the thread check triggers
                with patch(
                    "local_deep_research.utilities.db_utils.has_app_context",
                    return_value=False,
                ):
                    get_db_session()
            except RuntimeError as e:
                error_message[0] = str(e)

        thread = threading.Thread(
            target=background_task, name="CustomThreadName"
        )
        thread.start()
        thread.join()

        assert "CustomThreadName" in error_message[0]

    def test_main_thread_not_blocked_without_context(self):
        """MainThread should not be blocked by thread safety check."""
        # Verify that MainThread can proceed past the thread check
        # even without app context (unlike background threads)
        import threading

        # This test runs in MainThread
        assert threading.current_thread().name == "MainThread"

        # The function should not raise RuntimeError for MainThread
        # (it may fail later for other reasons, but not the thread check)