Adriano
263470786d
Aggiunge la persistenza SQLite, l'audit log a hash chain, il kill switch coordinato e i CLI di gestione documentati in docs/05-data-model.md e docs/07-risk-controls.md. 197 test pass, 1 skipped (sqlite3 CLI mancante), copertura totale 97%. State (`state/`): - 0001_init.sql con positions, instructions, decisions, dvol_history, manual_actions, system_state. - db.py: connect con WAL + foreign_keys + transaction ctx, runner forward-only basato su PRAGMA user_version. - models.py: record Pydantic, Decimal preservato come TEXT. - repository.py: CRUD typed con singola connessione passata, cache aware, posizioni concorrenti. Safety (`safety/`): - audit_log.py: AuditLog append-only con SHA-256 chain e fsync, verify_chain riconosce ogni manomissione (payload, prev_hash, hash, JSON, separatori). - kill_switch.py: arm/disarm transazionali, idempotenti, accoppiati all'audit chain. Config (`config/loader.py` + `strategy.yaml`): - Loader YAML con deep-merge di strategy.local.yaml. - Verifica config_hash SHA-256 (riga config_hash esclusa). - File golden strategy.yaml + esempio override. Scripts: - dead_man.sh: watchdog shell indipendente da Python. - backup.py: VACUUM INTO orario con retention 30 giorni. CLI: - audit verify (exit 2 su tampering). - kill-switch arm/disarm/status su SQLite reale. - state inspect con tabella posizioni aperte. - config hash, config validate. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
"""Migration runner + SQLite pragma sanity tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from cerbero_bite.state.db import (
|
|
connect,
|
|
current_version,
|
|
list_migrations,
|
|
run_migrations,
|
|
transaction,
|
|
)
|
|
|
|
|
|
def test_list_migrations_is_ordered_and_starts_with_0001() -> None:
|
|
migs = list_migrations()
|
|
assert migs, "expected at least one migration file"
|
|
versions = [m[0] for m in migs]
|
|
assert versions == sorted(versions)
|
|
assert versions[0] == 1
|
|
|
|
|
|
def test_run_migrations_creates_full_schema(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
conn = connect(db_path)
|
|
try:
|
|
new_version = run_migrations(conn)
|
|
assert new_version == max(m[0] for m in list_migrations())
|
|
# Sanity: every documented table is present.
|
|
existing = {
|
|
row["name"]
|
|
for row in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'"
|
|
).fetchall()
|
|
}
|
|
assert {
|
|
"positions",
|
|
"instructions",
|
|
"decisions",
|
|
"dvol_history",
|
|
"manual_actions",
|
|
"system_state",
|
|
} <= existing
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_run_migrations_is_idempotent(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
conn = connect(db_path)
|
|
try:
|
|
run_migrations(conn)
|
|
first = current_version(conn)
|
|
run_migrations(conn) # second call must be a no-op
|
|
assert current_version(conn) == first
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_pragmas_applied(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
conn = connect(db_path)
|
|
try:
|
|
run_migrations(conn)
|
|
assert conn.execute("PRAGMA foreign_keys").fetchone()[0] == 1
|
|
# WAL is sticky on the file: confirm via the journal_mode pragma.
|
|
mode = conn.execute("PRAGMA journal_mode").fetchone()[0]
|
|
assert mode.lower() == "wal"
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_transaction_rolls_back_on_exception(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
conn = connect(db_path)
|
|
try:
|
|
run_migrations(conn)
|
|
with pytest.raises(RuntimeError, match="boom"), transaction(conn):
|
|
conn.execute(
|
|
"INSERT INTO system_state(id, last_health_check, "
|
|
"config_version, started_at) VALUES (1, '2026-04-27', "
|
|
"'1.0.0', '2026-04-27')"
|
|
)
|
|
raise RuntimeError("boom")
|
|
# Row was rolled back.
|
|
rows = conn.execute("SELECT COUNT(*) FROM system_state").fetchone()[0]
|
|
assert rows == 0
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_foreign_key_enforcement(tmp_path: Path) -> None:
|
|
db_path = tmp_path / "state.sqlite"
|
|
conn = connect(db_path)
|
|
try:
|
|
run_migrations(conn)
|
|
# Inserting an instruction without a parent position must fail.
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO instructions(instruction_id, proposal_id, kind, "
|
|
"payload_json, sent_at) VALUES (?, ?, ?, ?, ?)",
|
|
("i1", "missing", "open_combo", "{}", "2026-04-27"),
|
|
)
|
|
finally:
|
|
conn.close()
|