Jan Doubravský
106 lines
3.3 KiB
Python
106 lines
3.3 KiB
Python
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from sqlmem.cache import CacheManager
|
|
|
|
|
|
@pytest.fixture
|
|
def source_conn():
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.execute("CREATE TABLE big (id TEXT, val TEXT)")
|
|
conn.executemany(
|
|
"INSERT INTO big VALUES (?, ?)", [(str(i), f"v{i}") for i in range(5)]
|
|
)
|
|
conn.commit()
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def cache(tmp_path):
|
|
c = CacheManager(db_path=tmp_path / "cache.db", backup_interval=9999)
|
|
yield c
|
|
c.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def small_batches(monkeypatch):
|
|
# Force multiple fetch batches over the 5 source rows.
|
|
monkeypatch.setattr("sqlmem.cache.FETCH_BATCH_SIZE", 2)
|
|
|
|
|
|
def test_batched_load_loads_all_rows(cache, source_conn, small_batches):
|
|
cache.load_table("big", ["id", "val"], source_conn)
|
|
_, rows = cache.execute_in_memory(
|
|
"SELECT id, val FROM big ORDER BY CAST(id AS INTEGER)"
|
|
)
|
|
assert len(rows) == 5
|
|
assert rows[0] == ("0", "v0")
|
|
assert rows[-1] == ("4", "v4")
|
|
|
|
|
|
def test_no_staging_table_left_behind(cache, source_conn, small_batches):
|
|
cache.load_table("big", ["id", "val"], source_conn)
|
|
names = {
|
|
r[0]
|
|
for r in cache.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
).fetchall()
|
|
}
|
|
assert "big" in names
|
|
assert not any(n.endswith("__sqlmem_load") for n in names)
|
|
|
|
|
|
def test_reload_replaces_data_atomically(cache, source_conn, small_batches):
|
|
cache.load_table("big", ["id", "val"], source_conn)
|
|
source_conn.execute("DELETE FROM big")
|
|
source_conn.execute("INSERT INTO big VALUES ('99', 'new')")
|
|
source_conn.commit()
|
|
cache.load_table("big", ["id", "val"], source_conn)
|
|
_, rows = cache.execute_in_memory("SELECT id, val FROM big")
|
|
assert rows == [("99", "new")]
|
|
|
|
|
|
def test_load_sets_ready_state(cache, source_conn):
|
|
cache.load_table("big", ["id", "val"], source_conn)
|
|
assert cache.get_states()["big"] == "ready"
|
|
|
|
|
|
def test_orphan_staging_dropped_on_startup(tmp_path, source_conn):
|
|
# Simulate a crash mid-load: a staging table persisted into cache.db.
|
|
db_path = tmp_path / "cache.db"
|
|
c1 = CacheManager(db_path=db_path, backup_interval=9999)
|
|
c1.load_table("big", ["id", "val"], source_conn)
|
|
c1.connection.execute("CREATE TABLE big__sqlmem_load (id TEXT, val TEXT)")
|
|
c1.connection.commit()
|
|
c1.close() # backup writes the staging table to disk
|
|
|
|
c2 = CacheManager(db_path=db_path, backup_interval=9999)
|
|
names = {
|
|
r[0]
|
|
for r in c2.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
).fetchall()
|
|
}
|
|
c2.close()
|
|
assert "big" in names # real table survives
|
|
assert not any(n.endswith("__sqlmem_load") for n in names) # orphan cleaned
|
|
|
|
|
|
def test_failed_load_sets_error_state_and_cleans_staging(cache):
|
|
empty_source = sqlite3.connect(":memory:") # has no 'big' table
|
|
try:
|
|
with pytest.raises(sqlite3.OperationalError):
|
|
cache.load_table("big", ["id"], empty_source)
|
|
assert cache.get_states()["big"] == "error"
|
|
names = {
|
|
r[0]
|
|
for r in cache.connection.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
).fetchall()
|
|
}
|
|
assert not any(n.endswith("__sqlmem_load") for n in names)
|
|
finally:
|
|
empty_source.close()
|