26 lines
1.1 KiB
Python
26 lines
1.1 KiB
Python
import sqlite3
|
|
|
|
from ._source import load_symbols
|
|
|
|
|
|
def test_operation_states_include_recovery_and_terminal_states():
|
|
ns = load_symbols("OPERATION_STATES", "TERMINAL_OPERATION_STATES")
|
|
assert {"requested", "running", "awaiting_reboot", "recovery_required"} <= ns["OPERATION_STATES"]
|
|
assert ns["TERMINAL_OPERATION_STATES"] <= ns["OPERATION_STATES"]
|
|
|
|
|
|
def test_operation_schema_prevents_duplicate_kind_and_idempotency_key():
|
|
ns = load_symbols("SCHEMA_VERSION", "_column_names", "run_migrations")
|
|
db = sqlite3.connect(":memory:")
|
|
db.execute("CREATE TABLE audit_log (id INTEGER PRIMARY KEY, detail TEXT)")
|
|
ns["run_migrations"](db)
|
|
values = ("1", "c1", "install", "requested", "now", "now", "same")
|
|
sql = "INSERT INTO operations (id,correlation_id,kind,state,requested_at,updated_at,idempotency_key) VALUES (?,?,?,?,?,?,?)"
|
|
db.execute(sql, values)
|
|
try:
|
|
db.execute(sql, ("2", "c2", "install", "requested", "now", "now", "same"))
|
|
except sqlite3.IntegrityError:
|
|
pass
|
|
else:
|
|
raise AssertionError("duplicate idempotent operation accepted")
|