83 lines
2.6 KiB
Python
83 lines
2.6 KiB
Python
import pytest
|
|
|
|
from app.db import check_database_integrity, checkpoint_sqlite_wal, create_sqlite_engine, ensure_sqlite_schema_compat
|
|
|
|
|
|
def test_check_database_integrity_returns_ok_for_valid_sqlite_database():
|
|
engine = create_sqlite_engine("sqlite:///:memory:")
|
|
try:
|
|
assert check_database_integrity(engine) == "ok"
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_check_database_integrity_raises_when_sqlite_reports_problem(monkeypatch):
|
|
class FakeCursor:
|
|
def execute(self, _sql):
|
|
return None
|
|
|
|
def fetchone(self):
|
|
return ("database disk image is malformed",)
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
class FakeConnection:
|
|
def cursor(self):
|
|
return FakeCursor()
|
|
|
|
class FakeRawConnection:
|
|
driver_connection = FakeConnection()
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
class FakeEngine:
|
|
def raw_connection(self):
|
|
return FakeRawConnection()
|
|
|
|
with pytest.raises(RuntimeError, match="SQLite integrity check failed"):
|
|
check_database_integrity(FakeEngine())
|
|
|
|
|
|
def test_checkpoint_sqlite_wal_truncates_wal_for_file_database(tmp_path):
|
|
db_path = tmp_path / "app.db"
|
|
engine = create_sqlite_engine(f"sqlite:///{db_path}")
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql("CREATE TABLE demo (id INTEGER PRIMARY KEY, name TEXT)")
|
|
connection.exec_driver_sql("INSERT INTO demo (name) VALUES ('ok')")
|
|
|
|
assert checkpoint_sqlite_wal(engine) is True
|
|
|
|
with engine.connect() as connection:
|
|
rows = connection.exec_driver_sql("SELECT name FROM demo").all()
|
|
assert rows == [("ok",)]
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_checkpoint_sqlite_wal_skips_in_memory_database():
|
|
engine = create_sqlite_engine("sqlite:///:memory:")
|
|
try:
|
|
assert checkpoint_sqlite_wal(engine) is False
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_ensure_sqlite_schema_compat_adds_progress_columns_to_existing_tasks_table(tmp_path):
|
|
db_path = tmp_path / "legacy.db"
|
|
engine = create_sqlite_engine(f"sqlite:///{db_path}")
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.exec_driver_sql("CREATE TABLE tasks (id VARCHAR(36) PRIMARY KEY, platform VARCHAR(32) NOT NULL)")
|
|
|
|
ensure_sqlite_schema_compat(engine)
|
|
|
|
with engine.connect() as connection:
|
|
columns = {row[1] for row in connection.exec_driver_sql("PRAGMA table_info(tasks)").all()}
|
|
assert "current_stage" in columns
|
|
assert "last_progress_at" in columns
|
|
finally:
|
|
engine.dispose()
|