52 lines
3.4 KiB
Python
52 lines
3.4 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", "run-a", "other", "x", "y", 1, 1, "KRW"), ("d4", "run-a", "expense", "", "", 7, 1, "KRW"), ("d5", "run-a", "expense", "", "", None, 1, "KRW"), ("d6", "run-a", "expense", "", "", 1, 0, "KRW"), ("d7", "run-a", "expense", None, "", 1, 1, "KRW"), ("d8", "run-a", "expense", "", None, 1, 1, "KRW")]:
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute(insert, values)
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
db.execute(insert, ("d9", "run-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 not 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
|