50 lines
1.6 KiB
Python
50 lines
1.6 KiB
Python
"""Tests for SQLite connection pragmas and runtime behavior."""
|
|
|
|
import sqlite3
|
|
|
|
import atocore.config as config
|
|
from atocore.models.database import get_connection, init_db
|
|
|
|
|
|
def test_get_connection_applies_busy_timeout_and_wal(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
|
|
monkeypatch.setenv("ATOCORE_DB_BUSY_TIMEOUT_MS", "7000")
|
|
|
|
original_settings = config.settings
|
|
try:
|
|
config.settings = config.Settings()
|
|
init_db()
|
|
with get_connection() as conn:
|
|
busy_timeout = conn.execute("PRAGMA busy_timeout").fetchone()[0]
|
|
journal_mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
foreign_keys = conn.execute("PRAGMA foreign_keys").fetchone()[0]
|
|
finally:
|
|
config.settings = original_settings
|
|
|
|
assert busy_timeout == 7000
|
|
assert str(journal_mode).lower() == "wal"
|
|
assert foreign_keys == 1
|
|
|
|
|
|
def test_get_connection_uses_configured_timeout_value(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("ATOCORE_DATA_DIR", str(tmp_path / "data"))
|
|
monkeypatch.setenv("ATOCORE_DB_BUSY_TIMEOUT_MS", "2500")
|
|
|
|
original_settings = config.settings
|
|
original_connect = sqlite3.connect
|
|
calls = []
|
|
|
|
def fake_connect(*args, **kwargs):
|
|
calls.append(kwargs.get("timeout"))
|
|
return original_connect(*args, **kwargs)
|
|
|
|
try:
|
|
config.settings = config.Settings()
|
|
monkeypatch.setattr("atocore.models.database.sqlite3.connect", fake_connect)
|
|
init_db()
|
|
finally:
|
|
config.settings = original_settings
|
|
|
|
assert calls
|
|
assert calls[0] == 2.5
|