Jan Doubravský
108 lines
3.5 KiB
Python
108 lines
3.5 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from sqlmem.cache import CacheManager
|
|
|
|
|
|
@pytest.fixture
|
|
def cache(tmp_path):
|
|
c = CacheManager(db_path=tmp_path / "test_cache.db", backup_interval=9999)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def source_conn():
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE users (name TEXT, email TEXT, status TEXT)")
|
|
conn.executemany(
|
|
"INSERT INTO users VALUES (?, ?, ?)",
|
|
[("alice", "alice@example.com", "active"), ("bob", "bob@example.com", "inactive")],
|
|
)
|
|
conn.commit()
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
def test_table_not_cached_initially(cache):
|
|
assert cache.is_table_cached("users") is False
|
|
|
|
|
|
def test_load_table(cache, source_conn):
|
|
cache.load_table("users", ["name", "email"], source_conn)
|
|
assert cache.is_table_cached("users") is True
|
|
|
|
|
|
def test_loaded_data_correct(cache, source_conn):
|
|
cache.load_table("users", ["name", "email"], source_conn)
|
|
rows = cache.connection.execute("SELECT name, email FROM users").fetchall()
|
|
assert len(rows) == 2
|
|
assert ("alice", "alice@example.com") in rows
|
|
|
|
|
|
def test_mark_table_refreshed(cache, source_conn):
|
|
cache.load_table("users", ["name"], source_conn)
|
|
row = cache.connection.execute(
|
|
"SELECT row_count FROM _sqlmem_tables WHERE table_name = 'users'"
|
|
).fetchone()
|
|
assert row[0] == 2
|
|
|
|
|
|
def test_backup_and_reload(tmp_path, source_conn):
|
|
db_path = tmp_path / "cache.db"
|
|
c1 = CacheManager(db_path=db_path, backup_interval=9999)
|
|
c1.load_table("users", ["name"], source_conn)
|
|
c1.close()
|
|
|
|
c2 = CacheManager(db_path=db_path, backup_interval=9999)
|
|
assert c2.is_table_cached("users") is True
|
|
c2.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Disk-backed mode (in_memory=False)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_disk_mode_persists_without_backup(tmp_path, source_conn):
|
|
"""Disk mode writes straight to the file — no explicit backup/close needed."""
|
|
db_path = tmp_path / "cache.db"
|
|
c = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
c.load_table("users", ["name"], source_conn)
|
|
# Data is already on disk; a brand-new disk-mode manager sees it.
|
|
c2 = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
assert c2.is_table_cached("users") is True
|
|
c2.close()
|
|
c.close()
|
|
|
|
|
|
def test_disk_mode_file_created_immediately(tmp_path, source_conn):
|
|
db_path = tmp_path / "cache.db"
|
|
c = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
c.load_table("users", ["name"], source_conn)
|
|
assert db_path.exists()
|
|
c.close()
|
|
|
|
|
|
def test_disk_mode_reload_in_new_instance(tmp_path, source_conn):
|
|
db_path = tmp_path / "cache.db"
|
|
c1 = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
c1.load_table("users", ["name", "email"], source_conn)
|
|
c1.close()
|
|
|
|
c2 = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
rows = c2.connection.execute("SELECT name FROM users").fetchall()
|
|
assert {r[0] for r in rows} == {"alice", "bob"}
|
|
c2.close()
|
|
|
|
|
|
def test_disk_mode_reset_keeps_file(tmp_path, source_conn):
|
|
db_path = tmp_path / "cache.db"
|
|
c = CacheManager(db_path=db_path, backup_interval=9999, in_memory=False)
|
|
c.load_table("users", ["name"], source_conn)
|
|
c.reset()
|
|
# File stays (the connection is still open) but the table is gone.
|
|
assert db_path.exists()
|
|
assert c.is_table_cached("users") is False
|
|
c.close()
|