Add initial SQLmem package structure with SQL parser, cache manager, column registry, and tests

This commit is contained in:
Jan Doubravský
2026-06-01 16:44:25 +02:00
parent 54879ef9d0
commit 74772cee4a
18 changed files with 835 additions and 0 deletions
+61
View File
@@ -0,0 +1,61 @@
import sqlite3
from pathlib import Path
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()