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

"""Tests for auth_db module."""

import tempfile
import threading
from pathlib import Path
from unittest.mock import patch, Mock


class TestGetAuthDbPath:
    """Tests for get_auth_db_path function."""

    def test_returns_path_object(self):
        """get_auth_db_path returns a Path object."""
        from local_deep_research.database.auth_db import get_auth_db_path

        with patch(
            "local_deep_research.database.auth_db.get_data_directory"
        ) as mock_get_data:
            mock_get_data.return_value = Path("/fake/data/dir")

            result = get_auth_db_path()

            assert isinstance(result, Path)

    def test_returns_correct_filename(self):
        """get_auth_db_path returns path with ldr_auth.db filename."""
        from local_deep_research.database.auth_db import get_auth_db_path

        with patch(
            "local_deep_research.database.auth_db.get_data_directory"
        ) as mock_get_data:
            mock_get_data.return_value = Path("/fake/data/dir")

            result = get_auth_db_path()

            assert result.name == "ldr_auth.db"

    def test_uses_data_directory(self):
        """get_auth_db_path uses get_data_directory for parent path."""
        from local_deep_research.database.auth_db import get_auth_db_path

        with patch(
            "local_deep_research.database.auth_db.get_data_directory"
        ) as mock_get_data:
            mock_get_data.return_value = Path("/test/data/path")

            result = get_auth_db_path()

            mock_get_data.assert_called_once()
            assert result.parent == Path("/test/data/path")


class TestInitAuthDatabase:
    """Tests for init_auth_database function."""

    def test_creates_database_directory(self):
        """init_auth_database creates parent directory if needed."""
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "subdir" / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_conn = Mock()
                    mock_engine.return_value.begin.return_value.__enter__ = (
                        Mock(return_value=mock_conn)
                    )
                    mock_engine.return_value.begin.return_value.__exit__ = Mock(
                        return_value=False
                    )

                    init_auth_database()

                # Directory should be created
                assert db_path.parent.exists()

    def test_idempotent_if_database_exists(self):
        """init_auth_database is idempotent when database already exists."""
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"
            # Create the file
            db_path.touch()

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_conn = Mock()
                    mock_engine.return_value.begin.return_value.__enter__ = (
                        Mock(return_value=mock_conn)
                    )
                    mock_engine.return_value.begin.return_value.__exit__ = Mock(
                        return_value=False
                    )

                    # Should not raise even if DB already exists
                    init_auth_database()

                    # create_engine is called (uses IF NOT EXISTS)
                    mock_engine.assert_called_once()

    def test_creates_tables(self):
        """init_auth_database creates User table using CreateTable DDL."""
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_conn = Mock()
                    mock_engine.return_value.begin.return_value.__enter__ = (
                        Mock(return_value=mock_conn)
                    )
                    mock_engine.return_value.begin.return_value.__exit__ = Mock(
                        return_value=False
                    )

                    init_auth_database()

                    # Should execute DDL via conn.execute
                    assert mock_conn.execute.called


class TestGetAuthDbSession:
    """Tests for get_auth_db_session function."""

    def test_returns_session(self):
        """get_auth_db_session returns a SQLAlchemy session."""
        from local_deep_research.database.auth_db import get_auth_db_session

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"
            # Create the file so init is skipped
            db_path.touch()

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_engine_instance = Mock()
                    mock_engine.return_value = mock_engine_instance

                    with patch("local_deep_research.database.auth_db.event"):
                        with patch(
                            "local_deep_research.database.auth_db.sessionmaker"
                        ) as mock_sessionmaker:
                            mock_session_class = Mock()
                            mock_session = Mock()
                            mock_session_class.return_value = mock_session
                            mock_sessionmaker.return_value = mock_session_class

                            result = get_auth_db_session()

                            assert result is mock_session

    def test_creates_database_if_missing(self):
        """get_auth_db_session initializes database if it doesn't exist."""
        from local_deep_research.database.auth_db import get_auth_db_session

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"
            # Don't create the file - it doesn't exist

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.init_auth_database"
                ) as mock_init:
                    with patch(
                        "local_deep_research.database.auth_db.create_engine"
                    ) as mock_engine:
                        mock_engine_instance = Mock()
                        mock_engine.return_value = mock_engine_instance

                        with patch(
                            "local_deep_research.database.auth_db.event"
                        ):
                            with patch(
                                "local_deep_research.database.auth_db.sessionmaker"
                            ) as mock_sessionmaker:
                                mock_session_class = Mock()
                                mock_session = Mock()
                                mock_session_class.return_value = mock_session
                                mock_sessionmaker.return_value = (
                                    mock_session_class
                                )

                                get_auth_db_session()

                                # init_auth_database should be called
                                mock_init.assert_called_once()

    def test_creates_engine_with_correct_url(self):
        """get_auth_db_session creates engine with correct SQLite URL."""
        from local_deep_research.database.auth_db import get_auth_db_session

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"
            db_path.touch()

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_engine_instance = Mock()
                    mock_engine.return_value = mock_engine_instance

                    with patch("local_deep_research.database.auth_db.event"):
                        with patch(
                            "local_deep_research.database.auth_db.sessionmaker"
                        ) as mock_sessionmaker:
                            mock_session_class = Mock()
                            mock_session = Mock()
                            mock_session_class.return_value = mock_session
                            mock_sessionmaker.return_value = mock_session_class

                            get_auth_db_session()

                            # Verify create_engine was called with sqlite URL
                            call_args = mock_engine.call_args[0][0]
                            assert call_args.startswith("sqlite:///")
                            assert "ldr_auth.db" in call_args


class TestAuthDbEngineCache:
    """Tests for cached engine behavior."""

    def test_get_auth_engine_returns_engine(self):
        """Test that _get_auth_engine() returns an engine."""
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                # Ensure we start with a clean state
                dispose_auth_engine()

                engine = _get_auth_engine()

                assert engine is not None

                # Clean up
                dispose_auth_engine()

    def test_engine_cached_on_subsequent_calls(self):
        """Test that _get_auth_engine() returns cached engine."""
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                # Ensure we start with a clean state
                dispose_auth_engine()

                engine1 = _get_auth_engine()
                engine2 = _get_auth_engine()

                # Should return the same cached engine
                assert engine1 is engine2

                # Clean up
                dispose_auth_engine()

    def test_engine_recreated_on_path_change(self):
        """Test that engine is recreated when data directory changes."""
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir1:
            with tempfile.TemporaryDirectory() as temp_dir2:
                db_path1 = Path(temp_dir1) / "ldr_auth.db"
                db_path2 = Path(temp_dir2) / "ldr_auth.db"

                # Ensure we start with a clean state
                dispose_auth_engine()

                with patch(
                    "local_deep_research.database.auth_db.get_auth_db_path"
                ) as mock_path:
                    # First call with path1
                    mock_path.return_value = db_path1
                    engine1 = _get_auth_engine()

                    # Change to path2
                    mock_path.return_value = db_path2
                    engine2 = _get_auth_engine()

                    # Should be different engines
                    assert engine1 is not engine2

                    # Clean up
                    dispose_auth_engine()

    def test_dispose_clears_cache(self):
        """Test that dispose_auth_engine() clears the cached engine."""
        from local_deep_research.database import auth_db
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                # Ensure we start with a clean state
                dispose_auth_engine()

                # Create an engine
                _get_auth_engine()

                # Dispose it
                dispose_auth_engine()

                # Module-level _auth_engine should be None
                assert auth_db._auth_engine is None

    def test_dispose_clears_path(self):
        """Test that dispose_auth_engine() clears the cached path."""
        from local_deep_research.database import auth_db
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                # Ensure we start with a clean state
                dispose_auth_engine()

                # Create an engine
                _get_auth_engine()

                # Dispose it
                dispose_auth_engine()

                # Module-level _auth_engine_path should be None
                assert auth_db._auth_engine_path is None

    def test_dispose_handles_no_engine(self):
        """Test that dispose_auth_engine() handles no engine gracefully."""
        from local_deep_research.database.auth_db import dispose_auth_engine

        # Dispose when there's no engine should not raise
        dispose_auth_engine()
        dispose_auth_engine()  # Calling twice should be fine


class TestInitAuthDatabaseAtomicDDL:
    """Tests for atomic DDL in init_auth_database (PR #2146)."""

    def test_uses_create_table_with_if_not_exists(self):
        """init_auth_database executes CreateTable with if_not_exists=True."""
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_conn = Mock()
                    mock_engine.return_value.begin.return_value.__enter__ = (
                        Mock(return_value=mock_conn)
                    )
                    mock_engine.return_value.begin.return_value.__exit__ = Mock(
                        return_value=False
                    )

                    init_auth_database()

                    # Verify conn.execute was called (CreateTable + CreateIndex)
                    assert mock_conn.execute.call_count >= 1

                    # Check that CreateTable DDL was passed
                    from sqlalchemy.schema import CreateTable

                    first_call = mock_conn.execute.call_args_list[0]
                    ddl_arg = first_call[0][0]
                    assert isinstance(ddl_arg, CreateTable)

    def test_does_not_use_base_metadata_create_all(self):
        """init_auth_database no longer uses Base.metadata.create_all."""
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                with patch(
                    "local_deep_research.database.auth_db.create_engine"
                ) as mock_engine:
                    mock_conn = Mock()
                    mock_engine.return_value.begin.return_value.__enter__ = (
                        Mock(return_value=mock_conn)
                    )
                    mock_engine.return_value.begin.return_value.__exit__ = Mock(
                        return_value=False
                    )

                    init_auth_database()

                    # Verify engine.begin() was used (not metadata.create_all)
                    mock_engine.return_value.begin.assert_called_once()

    def test_concurrent_init_does_not_fail(self):
        """Concurrent calls to init_auth_database complete without errors (PR #2146).

        The fix replaced Python-level TOCTOU check (file exists → skip)
        with SQL-level IF NOT EXISTS, which is atomic in SQLite.
        """
        from local_deep_research.database.auth_db import init_auth_database

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"
            errors = []

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path

                def init_worker():
                    try:
                        init_auth_database()
                    except Exception as e:
                        errors.append(e)

                # Run 10 concurrent init calls
                threads = [
                    threading.Thread(target=init_worker) for _ in range(10)
                ]
                for t in threads:
                    t.start()
                for t in threads:
                    t.join()

            assert len(errors) == 0, (
                f"Concurrent init_auth_database failed: {errors}"
            )
            # Database file should exist
            assert db_path.exists()

    def test_no_init_on_import(self):
        """init_auth_database is no longer called on module import (PR #2146)."""
        import importlib
        import local_deep_research.database.auth_db as auth_module

        # The reload discards the module's cached engine state and lock;
        # restore the pre-test module dict so later tests keep working
        # against the original objects.
        snapshot = dict(auth_module.__dict__)
        try:
            with patch(
                "local_deep_research.database.auth_db.init_auth_database"
            ) as mock_init:
                importlib.reload(auth_module)
                mock_init.assert_not_called()
        finally:
            auth_module.__dict__.clear()
            auth_module.__dict__.update(snapshot)


class TestAuthDbPerformancePragmas:
    """Tests for performance PRAGMA application on auth database connections."""

    def test_busy_timeout_set(self):
        """Verify busy_timeout is applied on auth database connections."""
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path
                dispose_auth_engine()

                engine = _get_auth_engine()
                with engine.connect() as conn:
                    result = conn.exec_driver_sql(
                        "PRAGMA busy_timeout"
                    ).scalar()
                    assert result == 10000

                dispose_auth_engine()

    def test_temp_store_set(self):
        """Verify temp_store=MEMORY is applied on auth database connections."""
        from local_deep_research.database.auth_db import (
            _get_auth_engine,
            dispose_auth_engine,
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            db_path = Path(temp_dir) / "ldr_auth.db"

            with patch(
                "local_deep_research.database.auth_db.get_auth_db_path"
            ) as mock_path:
                mock_path.return_value = db_path
                dispose_auth_engine()

                engine = _get_auth_engine()
                with engine.connect() as conn:
                    result = conn.exec_driver_sql("PRAGMA temp_store").scalar()
                    # temp_store=MEMORY is value 2
                    assert result == 2

                dispose_auth_engine()