32 lines
1.2 KiB
Python
32 lines
1.2 KiB
Python
import sqlite3
|
|
|
|
from ._source import load_symbols
|
|
|
|
|
|
def _migration_namespace():
|
|
return load_symbols("SCHEMA_VERSION", "_column_names", "run_migrations")
|
|
|
|
|
|
def test_migration_is_idempotent_and_creates_operation_contract():
|
|
db = sqlite3.connect(":memory:")
|
|
db.execute("CREATE TABLE audit_log (id INTEGER PRIMARY KEY, detail TEXT)")
|
|
ns = _migration_namespace()
|
|
assert ns["run_migrations"](db) == 1
|
|
assert ns["run_migrations"](db) == 1
|
|
columns = {row[1] for row in db.execute("PRAGMA table_info(operations)")}
|
|
assert {"id", "correlation_id", "kind", "state", "idempotency_key", "recovery_actions"} <= columns
|
|
assert "correlation_id" in {row[1] for row in db.execute("PRAGMA table_info(audit_log)")}
|
|
|
|
|
|
def test_newer_schema_is_rejected_without_mutation():
|
|
db = sqlite3.connect(":memory:")
|
|
db.execute("CREATE TABLE schema_metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
|
|
db.execute("INSERT INTO schema_metadata VALUES ('schema_version','999')")
|
|
ns = _migration_namespace()
|
|
try:
|
|
ns["run_migrations"](db)
|
|
except RuntimeError as exc:
|
|
assert "newer than supported" in str(exc)
|
|
else:
|
|
raise AssertionError("newer schema was accepted")
|