basicmachines-co-basic-memory/test-int/test_db_wal_mode.py

"""Integration tests for WAL mode and Windows-specific SQLite optimizations.

These tests use real filesystem databases (not in-memory) to verify WAL mode
and other SQLite configuration settings work correctly in production scenarios.
"""

import pytest
from unittest.mock import patch
from sqlalchemy import text


@pytest.mark.asyncio
async def test_wal_mode_enabled(engine_factory, db_backend):
    """Test that WAL mode is enabled on filesystem database connections."""
    if db_backend == "postgres":
        pytest.skip("SQLite-specific test - PRAGMA commands not supported in Postgres")

    engine, _ = engine_factory

    # Execute a query to verify WAL mode is enabled
    async with engine.connect() as conn:
        result = await conn.execute(text("PRAGMA journal_mode"))
        journal_mode = result.fetchone()[0]

        # WAL mode should be enabled for filesystem databases
        assert journal_mode.upper() == "WAL"


@pytest.mark.asyncio
async def test_busy_timeout_configured(engine_factory, db_backend):
    """Test that busy timeout is configured for database connections."""
    if db_backend == "postgres":
        pytest.skip("SQLite-specific test - PRAGMA commands not supported in Postgres")

    engine, _ = engine_factory

    async with engine.connect() as conn:
        result = await conn.execute(text("PRAGMA busy_timeout"))
        busy_timeout = result.fetchone()[0]

        # Busy timeout should be 10 seconds (10000 milliseconds)
        assert busy_timeout == 10000


@pytest.mark.asyncio
async def test_synchronous_mode_configured(engine_factory, db_backend):
    """Test that synchronous mode is set to NORMAL for performance."""
    if db_backend == "postgres":
        pytest.skip("SQLite-specific test - PRAGMA commands not supported in Postgres")

    engine, _ = engine_factory

    async with engine.connect() as conn:
        result = await conn.execute(text("PRAGMA synchronous"))
        synchronous = result.fetchone()[0]

        # Synchronous should be NORMAL (1)
        assert synchronous == 1


@pytest.mark.asyncio
async def test_cache_size_configured(engine_factory, db_backend):
    """Test that cache size is configured for performance."""
    if db_backend == "postgres":
        pytest.skip("SQLite-specific test - PRAGMA commands not supported in Postgres")

    engine, _ = engine_factory

    async with engine.connect() as conn:
        result = await conn.execute(text("PRAGMA cache_size"))
        cache_size = result.fetchone()[0]

        # Cache size should be -64000 (64MB)
        assert cache_size == -64000


@pytest.mark.asyncio
async def test_temp_store_configured(engine_factory, db_backend):
    """Test that temp_store is set to MEMORY."""
    if db_backend == "postgres":
        pytest.skip("SQLite-specific test - PRAGMA commands not supported in Postgres")

    engine, _ = engine_factory

    async with engine.connect() as conn:
        result = await conn.execute(text("PRAGMA temp_store"))
        temp_store = result.fetchone()[0]

        # temp_store should be MEMORY (2)
        assert temp_store == 2


@pytest.mark.asyncio
@pytest.mark.windows
@pytest.mark.skipif(
    __import__("os").name != "nt", reason="Windows-specific test - only runs on Windows platform"
)
async def test_windows_locking_mode_when_on_windows(tmp_path, monkeypatch, config_manager):
    """Test that Windows-specific locking mode is set when running on Windows."""
    from basic_memory.db import engine_session_factory, DatabaseType
    from basic_memory.config import DatabaseBackend

    # Force SQLite backend for this SQLite-specific test
    config_manager.config.database_backend = DatabaseBackend.SQLITE

    # Set HOME environment variable
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_HOME", str(tmp_path / "basic-memory"))

    db_path = tmp_path / "test_windows.db"

    async with engine_session_factory(db_path, DatabaseType.FILESYSTEM) as (
        engine,
        _,
    ):
        async with engine.connect() as conn:
            result = await conn.execute(text("PRAGMA locking_mode"))
            locking_mode = result.fetchone()[0]

            # Locking mode should be NORMAL on Windows
            assert locking_mode.upper() == "NORMAL"


@pytest.mark.asyncio
@pytest.mark.windows
@pytest.mark.skipif(
    __import__("os").name != "nt", reason="Windows-specific test - only runs on Windows platform"
)
async def test_null_pool_on_windows(tmp_path, monkeypatch):
    """Test that NullPool is used on Windows to avoid connection pooling issues."""
    from basic_memory.db import engine_session_factory, DatabaseType
    from sqlalchemy.pool import NullPool

    # Set HOME environment variable
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_HOME", str(tmp_path / "basic-memory"))

    db_path = tmp_path / "test_windows_pool.db"

    async with engine_session_factory(db_path, DatabaseType.FILESYSTEM) as (engine, _):
        # Engine should be using NullPool on Windows
        assert isinstance(engine.pool, NullPool)


@pytest.mark.asyncio
@pytest.mark.skipif(
    __import__("os").name == "nt", reason="Non-Windows test - cannot mock POSIX paths on Windows"
)
async def test_regular_pool_on_non_windows(tmp_path):
    """Test that regular pooling is used on non-Windows platforms."""
    from basic_memory.db import engine_session_factory, DatabaseType
    from sqlalchemy.pool import NullPool

    db_path = tmp_path / "test_posix_pool.db"

    with patch("basic_memory.db.os.name", "posix"):
        async with engine_session_factory(db_path, DatabaseType.FILESYSTEM) as (engine, _):
            # Engine should NOT be using NullPool on non-Windows
            assert not isinstance(engine.pool, NullPool)


@pytest.mark.asyncio
@pytest.mark.windows
@pytest.mark.skipif(
    __import__("os").name != "nt", reason="Windows-specific test - only runs on Windows platform"
)
async def test_memory_database_no_null_pool_on_windows(tmp_path, monkeypatch):
    """Test that in-memory databases do NOT use NullPool even on Windows.

    NullPool closes connections immediately, which destroys in-memory databases.
    This test ensures in-memory databases maintain connection pooling.
    """
    from basic_memory.db import engine_session_factory, DatabaseType
    from sqlalchemy.pool import NullPool

    # Set HOME environment variable
    monkeypatch.setenv("HOME", str(tmp_path))
    monkeypatch.setenv("BASIC_MEMORY_HOME", str(tmp_path / "basic-memory"))

    db_path = tmp_path / "test_memory.db"

    async with engine_session_factory(db_path, DatabaseType.MEMORY) as (engine, _):
        # In-memory databases should NOT use NullPool on Windows
        assert not isinstance(engine.pool, NullPool)