versioning/tests/test_dbsafety.py

"""Zero-error policy for live databases: only verified snapshots are stored."""

import sqlite3

import pytest

from versiond import dbsafe
from versiond.db import Database
from versiond.filters import Filters
from versiond.ingest import Rejected, Repository
from versiond.store import BlobStore


def _repo(tmp_path):
    db = Database(tmp_path / "index.sqlite")
    return Repository(db, BlobStore(tmp_path / "spool"), Filters())


def _open_db(path, **kw):
    conn = sqlite3.connect(path, **kw)
    conn.execute("CREATE TABLE IF NOT EXISTS t(id INTEGER PRIMARY KEY, v TEXT)")
    return conn


def _read_snapshot_bytes(repo, path):
    data, _ = repo.read_file(path)
    return data


def _rows_in_snapshot(data):
    import tempfile, os
    fd, tmp = tempfile.mkstemp(suffix=".sqlite")
    try:
        with os.fdopen(fd, "wb") as fh:
            fh.write(data)
        conn = sqlite3.connect(f"file:{tmp}?mode=ro", uri=True)
        try:
            assert conn.execute("PRAGMA integrity_check").fetchone()[0] == "ok"
            return conn.execute("SELECT v FROM t ORDER BY id").fetchall()
        finally:
            conn.close()
    finally:
        os.unlink(tmp)


def test_sqlite_snapshot_roundtrip(tmp_path):
    repo = _repo(tmp_path)
    db_path = str(tmp_path / "app.db")
    conn = _open_db(db_path)
    conn.execute("INSERT INTO t(v) VALUES ('a'), ('b')")
    conn.commit()
    data = _read_snapshot_bytes(repo, db_path)
    assert _rows_in_snapshot(data) == [("a",), ("b",)]
    result = repo.commit(db_path, data, "test", "edit")
    assert result["status"] == "committed"
    conn.close()


def test_snapshot_during_uncommitted_write_is_consistent(tmp_path):
    """A writer with an open transaction must not produce a torn version."""
    repo = _repo(tmp_path)
    db_path = str(tmp_path / "live.db")
    writer = _open_db(db_path, isolation_level=None)
    writer.execute("INSERT INTO t(v) VALUES ('committed')")
    writer.execute("BEGIN IMMEDIATE")
    writer.execute("INSERT INTO t(v) VALUES ('uncommitted')")
    try:
        data = _read_snapshot_bytes(repo, db_path)
    finally:
        writer.execute("ROLLBACK")
        writer.close()
    rows = _rows_in_snapshot(data)
    assert rows == [("committed",)]  # committed state only, verified clean


def test_wal_mode_snapshot(tmp_path):
    repo = _repo(tmp_path)
    db_path = str(tmp_path / "wal.db")
    conn = _open_db(db_path)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("INSERT INTO t(v) VALUES ('w1')")
    conn.commit()
    conn.execute("INSERT INTO t(v) VALUES ('w2')")  # may sit in -wal
    data = _read_snapshot_bytes(repo, db_path)
    conn.close()
    rows = _rows_in_snapshot(data)
    assert ("w1",) in rows  # committed data present, snapshot verified


def test_corrupt_sqlite_magic_rejected(tmp_path):
    repo = _repo(tmp_path)
    bad = tmp_path / "bad.db"
    bad.write_bytes(dbsafe.SQLITE_MAGIC + b"\x00" * 100 + b"garbage")
    with pytest.raises(Rejected) as exc:
        repo.read_file(str(bad))
    assert exc.value.reason in ("database-corrupt", "database-unreadable")


def test_locked_database_rejected_not_stored(tmp_path):
    repo = _repo(tmp_path)
    db_path = str(tmp_path / "locked.db")
    holder = _open_db(db_path, isolation_level=None)
    holder.execute("INSERT INTO t(v) VALUES ('x')")
    holder.execute("BEGIN EXCLUSIVE")
    try:
        with pytest.raises(dbsafe.DatabaseUnsafe) as exc:
            dbsafe.snapshot_sqlite(db_path, 10_485_760, timeout=0.3)
        assert exc.value.reason == "database-locked"
    finally:
        holder.execute("ROLLBACK")
        holder.close()
    # After the lock clears, the same file snapshots fine.
    assert _rows_in_snapshot(_read_snapshot_bytes(repo, db_path)) == [("x",)]


def test_pushed_sqlite_bytes_verified(tmp_path):
    repo = _repo(tmp_path)
    db_path = str(tmp_path / "push.db")
    with pytest.raises(Rejected):
        repo.check_content(dbsafe.SQLITE_MAGIC + b"\x00" * 100 + b"garbage")
    # Genuine DB bytes pass.
    conn = _open_db(db_path)
    conn.execute("INSERT INTO t(v) VALUES ('ok')")
    conn.commit()
    conn.close()
    repo.check_content((tmp_path / "push.db").read_bytes())