82 lines
4.8 KiB
Python
82 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
FIELD = "approved_cashflow_fingerprint"
|
|
|
|
|
|
def _column(path: str) -> tuple:
|
|
with sqlite3.connect(path) as conn:
|
|
return next(row for row in conn.execute("PRAGMA table_info(monthly_settlement)") if row[1] == FIELD)
|
|
|
|
|
|
def test_fresh_schema_has_nullable_text_field_without_default(tmp_path):
|
|
path = str(tmp_path / "fresh.sqlite3")
|
|
init_db(path)
|
|
column = _column(path)
|
|
assert column[2].upper() == "TEXT"
|
|
assert column[3] == 0
|
|
assert column[4] is None
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0] == "15"
|
|
conn.execute(
|
|
"INSERT INTO monthly_settlement "
|
|
"(id, month, basis_date, status, version, created_by, created_at, updated_by, updated_at) "
|
|
"VALUES (?, ?, ?, 'draft', 1, 'tester', 'now', 'tester', 'now')",
|
|
("m1", "2026-09", "2026-09-30"),
|
|
)
|
|
assert conn.execute(f"SELECT {FIELD} FROM monthly_settlement WHERE id='m1'").fetchone() == (None,)
|
|
|
|
|
|
def test_same_version_existing_db_upgrade_is_idempotent_and_preserves_rows(tmp_path):
|
|
path = str(tmp_path / "existing.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"INSERT INTO monthly_settlement "
|
|
"(id, month, basis_date, status, version, created_by, created_at, updated_by, updated_at) "
|
|
"VALUES ('m1', '2026-09', '2026-09-30', 'draft', 1, 'tester', 'now', 'tester', 'now')"
|
|
)
|
|
conn.execute(f"UPDATE monthly_settlement SET {FIELD}='cashflow-fp' WHERE id='m1'")
|
|
before = conn.execute("SELECT id, month, status, version, memo, approved_fingerprint, approved_cashflow_fingerprint FROM monthly_settlement").fetchall()
|
|
before_tables = conn.execute("SELECT name, sql FROM sqlite_master WHERE type IN ('table', 'index') AND name NOT LIKE 'monthly_settlement%' AND name <> 'sqlite_sequence' ORDER BY name").fetchall()
|
|
init_db(path)
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0] == "15"
|
|
assert conn.execute("SELECT id, month, status, version, memo, approved_fingerprint, approved_cashflow_fingerprint FROM monthly_settlement").fetchall() == before
|
|
assert conn.execute("SELECT name, sql FROM sqlite_master WHERE type IN ('table', 'index') AND name NOT LIKE 'monthly_settlement%' AND name <> 'sqlite_sequence' ORDER BY name").fetchall() == before_tables
|
|
|
|
|
|
def test_schema15_db_missing_field_gets_only_targeted_additive_column(tmp_path):
|
|
path = str(tmp_path / "legacy16.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("ALTER TABLE monthly_settlement RENAME TO monthly_settlement_without_fingerprint")
|
|
conn.execute(
|
|
"""CREATE TABLE monthly_settlement (
|
|
id TEXT PRIMARY KEY, month TEXT NOT NULL UNIQUE, basis_date TEXT NOT NULL,
|
|
status TEXT NOT NULL, version INTEGER NOT NULL, memo TEXT NOT NULL DEFAULT '',
|
|
created_by TEXT NOT NULL, created_at TEXT NOT NULL, updated_by TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL, reviewed_by TEXT, reviewed_at TEXT, review_fingerprint TEXT,
|
|
approved_by TEXT, approved_at TEXT, approved_version INTEGER, approved_fingerprint TEXT,
|
|
approval_reason TEXT, approval_flags_json TEXT NOT NULL DEFAULT '{}', closed_by TEXT,
|
|
closed_at TEXT, cancelled_by TEXT, cancelled_at TEXT, cancellation_reason TEXT
|
|
)"""
|
|
)
|
|
conn.execute("INSERT INTO monthly_settlement SELECT id,month,basis_date,status,version,memo,created_by,created_at,updated_by,updated_at,reviewed_by,reviewed_at,review_fingerprint,approved_by,approved_at,approved_version,approved_fingerprint,approval_reason,approval_flags_json,closed_by,closed_at,cancelled_by,cancelled_at,cancellation_reason FROM monthly_settlement_without_fingerprint")
|
|
conn.execute("DROP TABLE monthly_settlement_without_fingerprint")
|
|
conn.execute("INSERT INTO schema_meta(key,value) VALUES('schema_version','15') ON CONFLICT(key) DO UPDATE SET value='15'")
|
|
conn.execute(
|
|
"INSERT INTO monthly_settlement(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) "
|
|
"VALUES ('m1','2026-10','2026-10-31','draft',1,'u','t','u','t')"
|
|
)
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0] == "15"
|
|
assert _column(path)[2].upper() == "TEXT"
|
|
assert conn.execute("SELECT approved_cashflow_fingerprint FROM monthly_settlement WHERE id='m1'").fetchone() == (None,)
|