test-int/test_db_wal_mode.py
Rama main · réplica de lectura del repositorio original.
"""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 sqlalchemy import text
def _first_value(row):
assert row is not None
return row[0]
@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 = _first_value(result.fetchone())
# 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 = _first_value(result.fetchone())
# 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):
"""Synchronous defaults to NORMAL — safe with WAL, and OFF showed no
measurable throughput gain in benchmarks. See config.sqlite_synchronous."""
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 = _first_value(result.fetchone())
# NORMAL (1) is the default.
assert synchronous == 1
@pytest.mark.asyncio
@pytest.mark.parametrize("value,expected", [("OFF", 0), ("NORMAL", 1), ("FULL", 2)])
async def test_sqlite_synchronous_follows_config(tmp_path, value, expected):
"""_create_sqlite_engine applies the configured PRAGMA synchronous level."""
from basic_memory import db
from basic_memory.config import BasicMemoryConfig
config = BasicMemoryConfig(
projects={"main": {"path": str(tmp_path)}},
default_project="main",
sqlite_synchronous=value,
)
engine = db._create_sqlite_engine(
f"sqlite+aiosqlite:///{tmp_path / 'sync.db'}", db.DatabaseType.FILESYSTEM, config
)
try:
async with engine.connect() as conn:
got = _first_value((await conn.execute(text("PRAGMA synchronous"))).fetchone())
assert got == expected
finally:
await engine.dispose()
@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 = _first_value(result.fetchone())
# 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 = _first_value(result.fetchone())
# 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 = _first_value(result.fetchone())
# 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_bounded_pool_on_windows(tmp_path, monkeypatch):
"""Windows queues excess callers instead of opening competing SQLite writers."""
from basic_memory.db import engine_session_factory, DatabaseType
from sqlalchemy.pool import AsyncAdaptedQueuePool
# 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, _):
assert isinstance(engine.pool, AsyncAdaptedQueuePool)
assert engine.pool.size() == 5
assert engine.pool._max_overflow == 0
@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)