101 lines
7.2 KiB
Python
101 lines
7.2 KiB
Python
"""Cashflow schema and legacy migration contract regressions."""
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def _schema(path):
|
|
with sqlite3.connect(path) as db:
|
|
tables = {r[0] for r in db.execute("SELECT name FROM sqlite_master WHERE type='table'")}
|
|
assert {"settlement_cashflow_snapshots", "monthly_settlement"} <= tables
|
|
detail = {r[1] for r in db.execute("PRAGMA table_info(settlement_cashflow_snapshots)")}
|
|
assert {"id", "monthly_settlement_id", "cashflow_kind", "category", "subcategory", "amount", "transaction_count", "currency"} <= detail
|
|
|
|
|
|
def test_cashflow_detail_constraints_preserve_blank_categories(tmp_path):
|
|
path = str(tmp_path / "detail-constraints.sqlite3")
|
|
init_db(path)
|
|
_schema(path)
|
|
with sqlite3.connect(path) as db:
|
|
db.execute("PRAGMA foreign_keys=ON")
|
|
db.execute("INSERT INTO monthly_settlement(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) VALUES('month-a','2026-09','2026-09-30','draft',1,'test','now','test','now')")
|
|
insert = "INSERT INTO settlement_cashflow_snapshots(id,monthly_settlement_id,cashflow_kind,category,subcategory,amount,transaction_count,currency) VALUES(?,?,?,?,?,?,?,?)"
|
|
good = ("d1", "month-a", "expense", "", "", -7, 1, "KRW")
|
|
db.execute(insert, good)
|
|
for values in [("d2", "missing", "expense", "x", "y", 1, 1, "KRW"), ("d3", "month-a", "other", "x", "y", 1, 1, "KRW"), ("d4", "month-a", "expense", "", "", 7, 1, "KRW"), ("d5", "month-a", "expense", "", "", None, 1, "KRW"), ("d6", "month-a", "expense", "", "", 1, 0, "KRW"), ("d7", "month-a", "expense", None, "", 1, 1, "KRW"), ("d8", "month-a", "expense", "", None, 1, 1, "KRW")]:
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute(insert, values)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute(insert, ("d9", "month-a", "expense", "", "", -7, 1, "KRW"))
|
|
db.execute(insert, ("d10", "month-a", "income", "", "", 2, 1, "KRW"))
|
|
indexes = {r[1]: r[2] for r in db.execute("PRAGMA index_list(settlement_cashflow_snapshots)")}
|
|
unique_fields = [tuple(r[2] for r in db.execute(f"PRAGMA index_info('{name}')")) for name, unique in indexes.items() if unique]
|
|
assert any(x == ("monthly_settlement_id", "cashflow_kind", "category", "subcategory", "currency") for x in unique_fields)
|
|
assert db.execute("SELECT amount,category,subcategory FROM settlement_cashflow_snapshots WHERE id='d1'").fetchone() == (-7, "", "")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute("DELETE FROM monthly_settlement WHERE id='month-a'")
|
|
assert db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots WHERE monthly_settlement_id='month-a'").fetchone()[0] == 2
|
|
|
|
|
|
def test_init_is_idempotent_and_retains_schema_15_contract(tmp_path):
|
|
path = str(tmp_path / "fresh.sqlite3")
|
|
init_db(path)
|
|
_schema(path)
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT value FROM schema_meta WHERE key='schema_version'").fetchone()[0] == '15'
|
|
assert db.execute("SELECT 1 FROM sqlite_master WHERE name='asset_snapshot_runs'").fetchone()
|
|
first = db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots").fetchone()[0]
|
|
init_db(path)
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots").fetchone()[0] == first
|
|
|
|
|
|
def test_legacy_migration_is_idempotent_and_keeps_unstored_totals_null(tmp_path):
|
|
"""Schema-13 fixture uses the frozen production header and asset DDL."""
|
|
path = str(tmp_path / "schema13.sqlite3")
|
|
with sqlite3.connect(path) as db:
|
|
db.executescript("""
|
|
CREATE TABLE schema_meta(key TEXT PRIMARY KEY,value TEXT NOT NULL);
|
|
INSERT INTO schema_meta VALUES('schema_version','13');
|
|
CREATE TABLE asset_snapshots (
|
|
id TEXT PRIMARY KEY, basis_date TEXT NOT NULL, source_type TEXT NOT NULL,
|
|
source_id TEXT NOT NULL DEFAULT '', name TEXT NOT NULL, amount INTEGER,
|
|
currency TEXT NOT NULL DEFAULT 'KRW', liquidity_bucket TEXT NOT NULL DEFAULT 'other',
|
|
valuation_source TEXT NOT NULL DEFAULT '', source_note TEXT NOT NULL DEFAULT '',
|
|
memo TEXT NOT NULL DEFAULT '', snapshot_run_id TEXT NOT NULL DEFAULT '',
|
|
correction_of_run_id TEXT NOT NULL DEFAULT '', correction_reason TEXT NOT NULL DEFAULT '',
|
|
is_correction INTEGER NOT NULL DEFAULT 0, closed_at TEXT NOT NULL DEFAULT '',
|
|
valuation_id TEXT NOT NULL DEFAULT '', selection_policy TEXT NOT NULL DEFAULT 'exact',
|
|
verification_status TEXT NOT NULL DEFAULT 'unverified', evidence_kind TEXT NOT NULL DEFAULT 'applied',
|
|
exclusion_reason TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE asset_snapshot_runs (
|
|
id TEXT PRIMARY KEY, basis_date TEXT NOT NULL, close_status TEXT NOT NULL,
|
|
selection_policy TEXT NOT NULL DEFAULT 'exact', idempotency_key TEXT NOT NULL DEFAULT '',
|
|
approved_by TEXT NOT NULL DEFAULT '', approved_at TEXT NOT NULL DEFAULT '',
|
|
approval_reason TEXT NOT NULL DEFAULT '', approval_reference TEXT NOT NULL DEFAULT '',
|
|
correction_of_run_id TEXT NOT NULL DEFAULT '', correction_reason TEXT NOT NULL DEFAULT '',
|
|
applied_evidence TEXT NOT NULL DEFAULT '[]', excluded_evidence TEXT NOT NULL DEFAULT '[]',
|
|
carried_evidence TEXT NOT NULL DEFAULT '[]', request_fingerprint TEXT NOT NULL DEFAULT '',
|
|
created_at TEXT NOT NULL, draft_id TEXT NOT NULL DEFAULT '', draft_version INTEGER NOT NULL DEFAULT 0,
|
|
review_fingerprint TEXT NOT NULL DEFAULT ''
|
|
);
|
|
INSERT INTO asset_snapshot_runs(id,basis_date,close_status,created_at) VALUES('legacy-run','2026-09-30','complete_exact','then');
|
|
INSERT INTO asset_snapshots(id,basis_date,source_type,source_id,name,amount,created_at,snapshot_run_id) VALUES('legacy-row','2026-09-30','manual_asset','synthetic','Synthetic',42,'then','legacy-run');
|
|
""")
|
|
init_db(path)
|
|
_schema(path)
|
|
with sqlite3.connect(path) as db:
|
|
columns = {r[1] for r in db.execute("PRAGMA table_info(asset_snapshot_runs)")}
|
|
assert "draft_id" in columns and "income_total" in columns
|
|
first = db.execute("SELECT id,basis_date,close_status,created_at,income_total,expense_total,net_cashflow,total_assets,total_liabilities,net_worth FROM asset_snapshot_runs WHERE id='legacy-run'").fetchone()
|
|
assert first[:4] == ("legacy-run", "2026-09-30", "complete_exact", "then")
|
|
assert first[4:] == (None, None, None, None, None, None)
|
|
old = db.execute("SELECT id,basis_date,source_type,source_id,name,amount,snapshot_run_id FROM asset_snapshots").fetchall()
|
|
init_db(path)
|
|
with sqlite3.connect(path) as db:
|
|
assert db.execute("SELECT id,basis_date,close_status,created_at,income_total,expense_total,net_cashflow,total_assets,total_liabilities,net_worth FROM asset_snapshot_runs WHERE id='legacy-run'").fetchone() == first
|
|
assert db.execute("SELECT id,basis_date,source_type,source_id,name,amount,snapshot_run_id FROM asset_snapshots").fetchall() == old
|
|
assert db.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots").fetchone()[0] == 0
|