332 lines
11 KiB
Python
332 lines
11 KiB
Python
import sqlite3
|
|
from datetime import datetime, time, timedelta, timezone
|
|
|
|
import pytest
|
|
from sqlalchemy import create_engine
|
|
|
|
import sqlmem.engine as eng_mod
|
|
from sqlmem import CachingEngine, Delta, DeltaConfig, Schedule, TableSpec
|
|
from sqlmem.cache import CacheManager
|
|
from sqlmem.executor import QueryExecutor
|
|
from sqlmem.parser import parse
|
|
from sqlmem.registry import ColumnRegistry
|
|
from sqlmem.schedule import (
|
|
is_due,
|
|
next_occurrence,
|
|
parse_schedule,
|
|
previous_occurrence,
|
|
seconds_until_next,
|
|
)
|
|
from sqlmem.stats import StatsCollector, TableState
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# schedule.py — spec parsing
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_single_time_string():
|
|
assert parse_schedule("t", "03:00") == (time(3, 0),)
|
|
|
|
|
|
def test_parse_accepts_seconds_and_time_objects():
|
|
assert parse_schedule("t", "03:30:15") == (time(3, 30, 15),)
|
|
assert parse_schedule("t", time(6, 45)) == (time(6, 45),)
|
|
|
|
|
|
def test_parse_list_is_sorted_and_deduplicated():
|
|
assert parse_schedule("t", ["15:30", "03:00", "03:00"]) == (time(3, 0), time(15, 30))
|
|
|
|
|
|
@pytest.mark.parametrize("spec", ["3pm", "25:00", "", "03:00:00.5x", None, 300])
|
|
def test_parse_rejects_invalid_time(spec):
|
|
with pytest.raises(ValueError):
|
|
parse_schedule("t", spec)
|
|
|
|
|
|
def test_parse_rejects_empty_list():
|
|
with pytest.raises(ValueError):
|
|
parse_schedule("t", [])
|
|
|
|
|
|
def test_parse_rejects_timezone_aware_time():
|
|
with pytest.raises(ValueError):
|
|
parse_schedule("t", time(3, 0, tzinfo=timezone.utc))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# schedule.py — occurrence maths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
TZ = timezone(timedelta(hours=2))
|
|
TIMES = (time(3, 0), time(15, 30))
|
|
|
|
|
|
def at(hour, minute=0, day=10):
|
|
return datetime(2026, 6, day, hour, minute, tzinfo=TZ)
|
|
|
|
|
|
def test_previous_occurrence_earlier_today():
|
|
assert previous_occurrence(TIMES, at(16)) == at(15, 30)
|
|
|
|
|
|
def test_previous_occurrence_falls_back_to_yesterday():
|
|
assert previous_occurrence(TIMES, at(2)) == at(15, 30, day=9)
|
|
|
|
|
|
def test_previous_occurrence_includes_the_exact_moment():
|
|
assert previous_occurrence(TIMES, at(3)) == at(3)
|
|
|
|
|
|
def test_next_occurrence_and_seconds_until():
|
|
assert next_occurrence(TIMES, at(4)) == at(15, 30)
|
|
assert next_occurrence(TIMES, at(16)) == at(3, 0, day=11)
|
|
assert seconds_until_next(TIMES, at(15, 0)) == 30 * 60
|
|
|
|
|
|
def test_is_due_only_when_last_refresh_predates_the_slot():
|
|
now = at(16)
|
|
assert is_due(TIMES, at(15, 0), now) is True # loaded before the 15:30 slot
|
|
assert is_due(TIMES, at(15, 45), now) is False # loaded after it
|
|
assert is_due(TIMES, None, now) is False # not cached at all
|
|
|
|
|
|
def test_is_due_compares_across_timezones():
|
|
# last_refresh is persisted in UTC; the slot is local.
|
|
assert is_due(TIMES, datetime(2026, 6, 10, 12, 0, tzinfo=timezone.utc), at(16)) is True
|
|
assert is_due(TIMES, datetime(2026, 6, 10, 14, 0, tzinfo=timezone.utc), at(16)) is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# executor-level: read-time guarantee
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def source_conn():
|
|
conn = sqlite3.connect(":memory:")
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE products (id TEXT, name TEXT, price TEXT);
|
|
INSERT INTO products VALUES ('1', 'Widget', '9.99'), ('2', 'Gadget', '19.99');
|
|
"""
|
|
)
|
|
conn.commit()
|
|
yield conn
|
|
conn.close()
|
|
|
|
|
|
def make_executor(tmp_path, source_conn, schedule):
|
|
cache = CacheManager(db_path=tmp_path / "cache.db", backup_interval=9999)
|
|
registry = ColumnRegistry(cache.connection)
|
|
stats = StatsCollector()
|
|
return QueryExecutor(cache, registry, source_conn, stats, schedule=schedule)
|
|
|
|
|
|
def run(executor, sql, params=None):
|
|
return executor.execute(parse(sql, params))
|
|
|
|
|
|
def slot(minutes_from_now):
|
|
"""A schedule time of day *minutes_from_now* relative to the real clock."""
|
|
when = datetime.now().astimezone() + timedelta(minutes=minutes_from_now)
|
|
return parse_schedule("t", when.time().replace(microsecond=0))
|
|
|
|
|
|
def backdate(cache, table, hours=2):
|
|
"""Pretend the cached copy was loaded *hours* ago, i.e. before a past slot.
|
|
|
|
Only ``last_refresh_at`` is moved (the timestamp the schedule is judged
|
|
against), so the reload path is exercised against the real clock.
|
|
"""
|
|
stamp = (datetime.now(timezone.utc) - timedelta(hours=hours)).isoformat()
|
|
cache.connection.execute(
|
|
"UPDATE _sqlmem_tables SET last_refresh_at = ? WHERE table_name = ?", (stamp, table)
|
|
)
|
|
cache.connection.commit()
|
|
|
|
|
|
def test_query_reloads_when_slot_has_passed(tmp_path, source_conn):
|
|
executor = make_executor(tmp_path, source_conn, {"products": slot(-1)})
|
|
run(executor, "SELECT id, price FROM products") # miss → load
|
|
source_conn.execute("UPDATE products SET price = '1.11' WHERE id = '1'")
|
|
source_conn.commit()
|
|
backdate(executor._cache, "products")
|
|
|
|
rows = {r["id"]: r for r in run(executor, "SELECT id, price FROM products")}
|
|
assert rows["1"]["price"] == "1.11" # slot passed → reloaded before answering
|
|
assert executor._stats.refetches == 1
|
|
assert executor._stats.misses == 1
|
|
|
|
|
|
def test_query_before_slot_is_cache_hit(tmp_path, source_conn):
|
|
# The slot is a minute away, so it cannot have passed since the load.
|
|
executor = make_executor(tmp_path, source_conn, {"products": slot(1)})
|
|
run(executor, "SELECT id, price FROM products")
|
|
source_conn.execute("UPDATE products SET price = '1.11' WHERE id = '1'")
|
|
source_conn.commit()
|
|
|
|
rows = {r["id"]: r for r in run(executor, "SELECT id, price FROM products")}
|
|
assert rows["1"]["price"] == "9.99" # still fresh → cached value served
|
|
assert executor._stats.hits == 1
|
|
assert executor._stats.refetches == 0
|
|
|
|
|
|
def test_scheduled_reload_preserves_full_status(tmp_path, source_conn):
|
|
executor = make_executor(tmp_path, source_conn, {"products": slot(-1)})
|
|
run(executor, "SELECT * FROM products") # full load
|
|
backdate(executor._cache, "products")
|
|
run(executor, "SELECT * FROM products") # slot passed → full reload
|
|
assert executor._cache.is_table_full("products") is True
|
|
|
|
|
|
def test_reload_happens_once_per_slot(tmp_path, source_conn):
|
|
executor = make_executor(tmp_path, source_conn, {"products": slot(-1)})
|
|
run(executor, "SELECT id, price FROM products")
|
|
backdate(executor._cache, "products")
|
|
run(executor, "SELECT id, price FROM products") # reload for that slot
|
|
run(executor, "SELECT id, price FROM products") # same slot → cache hit
|
|
assert executor._stats.refetches == 1
|
|
assert executor._stats.hits == 1
|
|
|
|
|
|
def test_untracked_table_never_expires_by_schedule(tmp_path, source_conn):
|
|
executor = make_executor(tmp_path, source_conn, {"other": slot(-1)})
|
|
run(executor, "SELECT id, name FROM products")
|
|
backdate(executor._cache, "products")
|
|
rows = {r["id"]: r for r in run(executor, "SELECT id, name FROM products")}
|
|
assert rows["1"]["name"] == "Widget"
|
|
assert executor._stats.hits == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# engine level: legacy schedule= kwarg
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def source_db(tmp_path):
|
|
db_path = tmp_path / "source.db"
|
|
conn = sqlite3.connect(db_path)
|
|
conn.executescript(
|
|
"""
|
|
CREATE TABLE products (id TEXT PRIMARY KEY, name TEXT, changed TEXT);
|
|
INSERT INTO products VALUES ('1', 'Widget', '2026-06-01 10:00:00');
|
|
"""
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
return db_path
|
|
|
|
|
|
@pytest.fixture
|
|
def source_engine(source_db):
|
|
engine = create_engine(f"sqlite:///{source_db}")
|
|
yield engine
|
|
engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
def patched_cache(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(eng_mod, "CACHE_DB_PATH", tmp_path / "cache.db")
|
|
monkeypatch.setattr(eng_mod, "BACKUP_INTERVAL_SECONDS", 9999)
|
|
|
|
|
|
def test_background_scheduled_refresh(source_engine, source_db, patched_cache):
|
|
engine = CachingEngine(source_engine, schedule={"products": slot(-1)[0]})
|
|
engine.execute("SELECT id, name FROM products")
|
|
|
|
conn = sqlite3.connect(source_db)
|
|
conn.execute("UPDATE products SET name = 'Widget2' WHERE id = '1'")
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
backdate(engine._cache, "products")
|
|
engine.refresh() # background-style reload of the table whose slot passed
|
|
rows = engine.execute("SELECT id, name FROM products")
|
|
assert rows[0]["name"] == "Widget2"
|
|
engine.close()
|
|
|
|
|
|
def test_stats_report_schedule_tracking_and_staleness(source_engine, patched_cache):
|
|
engine = CachingEngine(source_engine, schedule={"products": ["03:00", "15:30"]})
|
|
engine.execute("SELECT id, name FROM products")
|
|
|
|
stats = engine.stats
|
|
assert stats.tables["products"].tracking == "schedule"
|
|
assert stats.tables["products"].state == TableState.READY
|
|
|
|
backdate(engine._cache, "products", hours=48) # both slots have since passed
|
|
assert engine.stats.tables["products"].state == TableState.STALE
|
|
engine.close()
|
|
|
|
|
|
def test_next_tick_shortens_to_the_next_slot(source_engine, patched_cache):
|
|
engine = CachingEngine(source_engine) # no schedule → plain interval
|
|
engine._refresh_interval = 300
|
|
assert engine._next_tick() == 300
|
|
|
|
engine._schedule = {"products": slot(0.5)}
|
|
assert 1.0 < engine._next_tick() <= 32.0
|
|
engine.close()
|
|
|
|
|
|
def test_invalid_schedule_rejected_at_construction(source_engine, patched_cache):
|
|
with pytest.raises(ValueError):
|
|
CachingEngine(source_engine, schedule={"products": "half past three"})
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"kwargs",
|
|
[
|
|
{"ttl": {"products": 300}, "schedule": {"products": "03:00"}},
|
|
{
|
|
"delta": {"products": DeltaConfig(change_column="changed", key_columns=["id"])},
|
|
"schedule": {"products": "03:00"},
|
|
},
|
|
],
|
|
)
|
|
def test_schedule_overlapping_another_method_raises(source_engine, patched_cache, kwargs):
|
|
with pytest.raises(ValueError):
|
|
CachingEngine(source_engine, **kwargs)
|
|
|
|
|
|
def test_tables_and_schedule_kwarg_are_mutually_exclusive(source_engine, patched_cache):
|
|
with pytest.raises(ValueError):
|
|
CachingEngine(
|
|
source_engine,
|
|
tables=[TableSpec("products", ["id"])],
|
|
schedule={"products": "03:00"},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# declarative mode: Schedule refresh strategy on TableSpec
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_tablespec_schedule_tracking(source_engine, patched_cache):
|
|
engine = CachingEngine(
|
|
source_engine,
|
|
tables=[
|
|
TableSpec(
|
|
"products",
|
|
["id", "name"],
|
|
refresh=Schedule(["03:00", "15:30"]),
|
|
preload=True,
|
|
)
|
|
],
|
|
blocking_startup_refresh=True,
|
|
)
|
|
assert engine.stats.tables["products"].tracking == "schedule"
|
|
engine.close()
|
|
|
|
|
|
def test_schedule_object_validates_eagerly():
|
|
with pytest.raises(ValueError):
|
|
Schedule("not a time")
|
|
|
|
|
|
def test_schedule_and_delta_are_distinct_strategies():
|
|
# Sanity check that Schedule isn't accidentally aliased to Delta/TTL.
|
|
assert Schedule is not Delta
|