tests/test_persistence.py

import os, sqlite3
from app.persistence import safe_copy_sqlite, restore_soul, snapshot_soul

def _make_db(path):
    c = sqlite3.connect(path); c.execute("create table t(x)"); c.execute("insert into t values (42)"); c.commit(); c.close()

def test_safe_copy_is_consistent(tmp_path):
    src = str(tmp_path / "soul.db"); _make_db(src)
    live = sqlite3.connect(src)  # keep it OPEN (like the running app)
    dst = str(tmp_path / "snap.db")
    safe_copy_sqlite(src, dst)
    live.close()
    assert os.path.exists(dst)
    c = sqlite3.connect(dst); assert c.execute("select x from t").fetchone()[0] == 42; c.close()

def test_restore_noop_without_token(tmp_path, monkeypatch):
    monkeypatch.delenv("HF_TOKEN", raising=False)
    assert restore_soul(str(tmp_path / "soul.db"), "x/y") is False

def test_snapshot_never_raises_without_token(tmp_path, monkeypatch):
    monkeypatch.delenv("HF_TOKEN", raising=False)
    src = str(tmp_path / "soul.db"); _make_db(src)
    snapshot_soul(src, "x/y")  # must not raise

def test_snapshot_failsoft_on_error(tmp_path, monkeypatch):
    monkeypatch.setenv("HF_TOKEN", "tok")
    src = str(tmp_path / "soul.db"); _make_db(src)
    import app.persistence as p
    monkeypatch.setattr(p, "_upload", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("net")))
    snapshot_soul(src, "x/y")  # swallowed, no raise

def test_snapshotter_throttles(tmp_path, monkeypatch):
    from app.persistence import Snapshotter
    import sqlite3
    src = str(tmp_path / "soul.db"); sqlite3.connect(src).close()
    calls = {"n": 0}
    s = Snapshotter(db_path=src, repo_id="x/y", min_interval=999, now=lambda: 1000.0)
    monkeypatch.setattr(s, "_do", lambda: calls.__setitem__("n", calls["n"] + 1))
    s.maybe_snapshot(); s.maybe_snapshot()  # 2nd within interval -> skipped
    s._join()
    assert calls["n"] == 1