"""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()