patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
102 lines
4.1 KiB
Python
102 lines
4.1 KiB
Python
import pytest
|
|
|
|
from app.db import MIGRATIONS, Database, MigrationError
|
|
from app.migrate import split_statements
|
|
|
|
CREATE_A = "CREATE TABLE a (id INTEGER PRIMARY KEY, name TEXT);"
|
|
GOOD_SECOND = "CREATE TABLE b (id INTEGER); INSERT INTO a(name) VALUES ('x');"
|
|
BROKEN_SECOND = "CREATE TABLE c (id INTEGER); INSERT INTO missing_table VALUES (1);"
|
|
|
|
|
|
def version(db):
|
|
return db.query("PRAGMA user_version")[0][0]
|
|
|
|
|
|
def tables(db):
|
|
return {row["name"] for row in db.query("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
|
|
|
|
def test_split_statements_keeps_quoted_semicolons_and_drops_empty_tail():
|
|
script = "CREATE TABLE t (v TEXT);\nINSERT INTO t VALUES ('a;b');\n \n"
|
|
assert split_statements(script) == ["CREATE TABLE t (v TEXT);", "INSERT INTO t VALUES ('a;b');"]
|
|
|
|
|
|
def test_successful_run_sets_user_version():
|
|
db = Database(":memory:", migrations=[CREATE_A, GOOD_SECOND])
|
|
assert version(db) == 2
|
|
assert {"a", "b"} <= tables(db)
|
|
assert db.query("SELECT COUNT(*) AS n FROM a")[0]["n"] == 1
|
|
|
|
|
|
def test_default_migrations_apply_completely():
|
|
db = Database(":memory:")
|
|
assert version(db) == len(MIGRATIONS)
|
|
assert {"service_states", "routes", "app_state"} <= tables(db)
|
|
columns = {row["name"] for row in db.query("PRAGMA table_info(service_states)")}
|
|
assert {"ack_state", "host_held"} <= columns
|
|
|
|
|
|
def test_broken_second_statement_rolls_back_first_and_keeps_version(tmp_path):
|
|
path = tmp_path / "m.db"
|
|
Database(path, migrations=[CREATE_A]).close()
|
|
with pytest.raises(MigrationError, match="Migration 1"):
|
|
Database(path, migrations=[CREATE_A, BROKEN_SECOND])
|
|
db = Database(path, migrations=[CREATE_A])
|
|
assert version(db) == 1
|
|
assert "c" not in tables(db)
|
|
|
|
|
|
def test_existing_tables_and_data_survive_failed_follow_up_migration(tmp_path):
|
|
path = tmp_path / "m.db"
|
|
db = Database(path, migrations=[CREATE_A])
|
|
db.insert("INSERT INTO a(name) VALUES ('keep')")
|
|
db.close()
|
|
with pytest.raises(MigrationError):
|
|
Database(path, migrations=[CREATE_A, BROKEN_SECOND])
|
|
again = Database(path, migrations=[CREATE_A])
|
|
assert [r["name"] for r in again.query("SELECT name FROM a")] == ["keep"]
|
|
|
|
|
|
def test_restart_does_not_apply_migrations_twice(tmp_path):
|
|
path = tmp_path / "m.db"
|
|
migrations = [CREATE_A, "INSERT INTO a(name) VALUES ('once');"]
|
|
Database(path, migrations=migrations).close()
|
|
db = Database(path, migrations=migrations)
|
|
assert version(db) == 2
|
|
assert db.query("SELECT COUNT(*) AS n FROM a")[0]["n"] == 1
|
|
|
|
|
|
def test_failed_migration_can_be_retried_after_fix(tmp_path):
|
|
path = tmp_path / "m.db"
|
|
Database(path, migrations=[CREATE_A]).close()
|
|
with pytest.raises(MigrationError):
|
|
Database(path, migrations=[CREATE_A, BROKEN_SECOND])
|
|
db = Database(path, migrations=[CREATE_A, GOOD_SECOND])
|
|
assert version(db) == 2
|
|
|
|
|
|
def test_connection_keeps_normal_commit_behaviour_after_migration():
|
|
db = Database(":memory:", migrations=[CREATE_A])
|
|
db.insert("INSERT INTO a(name) VALUES ('p')")
|
|
with pytest.raises(Exception):
|
|
db.execute_many([("INSERT INTO a(name) VALUES ('q')", ()), ("INSERT INTO nope VALUES (1)", ())])
|
|
assert [r["name"] for r in db.query("SELECT name FROM a")] == ["p"]
|
|
|
|
|
|
def test_phase_6c_migrations_keep_rows_set_marker_and_null_zero_reminders(tmp_path):
|
|
path = tmp_path / "m.db"
|
|
old = Database(path, migrations=MIGRATIONS[:7])
|
|
old.execute("INSERT INTO service_states(host, service, state, last_state_change) VALUES ('h','s',2,1)")
|
|
old.execute("INSERT INTO routes(name, position, reminder_hours) VALUES ('a', 1, 0)")
|
|
old.execute("INSERT INTO routes(name, position, reminder_hours) VALUES ('b', 2, 4)")
|
|
old.close()
|
|
db = Database(path)
|
|
row = db.query("SELECT ack_state, host_held FROM service_states")[0]
|
|
assert row["ack_state"] is None and row["host_held"] == 0
|
|
assert db.query("SELECT value FROM app_state WHERE key='baseline.done'")[0]["value"] == "1"
|
|
assert [r["reminder_hours"] for r in db.query("SELECT reminder_hours FROM routes ORDER BY position")] == [None, 4]
|
|
|
|
|
|
def test_fresh_database_has_no_baseline_marker():
|
|
assert Database(":memory:").query("SELECT 1 FROM app_state WHERE key='baseline.done'") == []
|