107 lines
6.2 KiB
Python
107 lines
6.2 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def _insert_sql():
|
|
return """INSERT INTO monthly_settlement
|
|
(id, month, basis_date, status, version, created_by, created_at, updated_by, updated_at)
|
|
VALUES (?, ?, ?, 'draft', 1, 'tester', '2026-10-15T00:00:00+09:00', 'tester', '2026-10-15T00:00:00+09:00')"""
|
|
|
|
|
|
def test_fresh_db_has_month_settlement_schema_15_without_latest_run_id(tmp_path):
|
|
path = str(tmp_path / "fresh.sqlite3")
|
|
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 COUNT(*) FROM sqlite_master WHERE type='table' AND name='monthly_settlement'").fetchone()[0] == 1
|
|
columns = {row[1]: row for row in conn.execute("PRAGMA table_info(monthly_settlement)")}
|
|
assert columns["approval_flags_json"][4] == "'{}'"
|
|
assert columns["memo"][4] == "''"
|
|
columns = {row[1] for row in conn.execute("PRAGMA table_info(monthly_settlement)")}
|
|
assert "latest_run_id" not in columns
|
|
assert conn.execute("PRAGMA foreign_key_list(monthly_settlement)").fetchall() == []
|
|
conn.execute(_insert_sql(), ("m1", "2026-09", "2026-09-30"))
|
|
row = conn.execute("SELECT memo, approval_flags_json FROM monthly_settlement").fetchone()
|
|
assert row == ("", "{}")
|
|
|
|
|
|
def test_migration_is_idempotent_and_preserves_legacy_drafts(tmp_path):
|
|
path = str(tmp_path / "existing.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("INSERT INTO settlement_drafts (id, month, basis_date, status, version, created_by, created_at, updated_at, create_key) VALUES ('draft-1','2026-09','2026-09-30','active',1,'u','t','t','k')")
|
|
before = conn.execute("SELECT id, month, basis_date, status FROM settlement_drafts").fetchall()
|
|
init_db(path)
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT id, month, basis_date, status FROM settlement_drafts").fetchall() == before
|
|
assert conn.execute("SELECT COUNT(*) FROM monthly_settlement").fetchone()[0] == 0
|
|
|
|
|
|
def test_schema15_upgrade_removes_schema16_run_link_and_preserves_rows(tmp_path):
|
|
path = str(tmp_path / "schema15.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("INSERT INTO asset_snapshot_runs(id,basis_date,close_status,created_at) VALUES('run-1','2026-09-30','complete_exact','t')")
|
|
conn.execute("ALTER TABLE monthly_settlement ADD COLUMN latest_run_id TEXT REFERENCES asset_snapshot_runs(id) ON DELETE RESTRICT")
|
|
conn.execute("UPDATE schema_meta SET value='15' WHERE key='schema_version'")
|
|
conn.execute("""INSERT INTO monthly_settlement
|
|
(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at,
|
|
approved_by,approved_at,approved_version,approved_fingerprint,closed_by,closed_at,latest_run_id)
|
|
VALUES ('m1','2026-09','2026-09-30','closed',2,'u','t','u','t','r','t',1,'fp','c','t','run-1')""")
|
|
conn.execute(
|
|
"""INSERT INTO account_snapshot
|
|
(id,basis_date,source_type,source_id,name,amount,monthly_settlement_id,created_at)
|
|
VALUES ('snapshot-1','2026-09-30','account','account-1','Main',100,'m1','t')"""
|
|
)
|
|
before = conn.execute("SELECT id,month,basis_date,status,version,memo,approved_by,closed_by FROM monthly_settlement").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,basis_date,status,version,memo,approved_by,closed_by FROM monthly_settlement").fetchall() == before
|
|
assert "latest_run_id" not in {row[1] for row in conn.execute("PRAGMA table_info(monthly_settlement)")}
|
|
assert conn.execute("SELECT id FROM asset_snapshot_runs").fetchall() == [("run-1",)]
|
|
assert conn.execute("SELECT id,monthly_settlement_id FROM account_snapshot").fetchall() == [("snapshot-1", "m1")]
|
|
assert conn.execute("PRAGMA foreign_key_check").fetchall() == []
|
|
migrated_shape = [row[1:] for row in conn.execute("PRAGMA table_info(monthly_settlement)")]
|
|
canonical_path = str(tmp_path / "canonical.sqlite3")
|
|
init_db(canonical_path)
|
|
with sqlite3.connect(canonical_path) as conn:
|
|
canonical_shape = [row[1:] for row in conn.execute("PRAGMA table_info(monthly_settlement)")]
|
|
assert migrated_shape == canonical_shape
|
|
|
|
|
|
@pytest.mark.parametrize("column,value", [("month", "2026-13"), ("month", "202609"), ("basis_date", "2026-02-30"), ("version", 0)])
|
|
def test_month_settlement_rejects_invalid_values(tmp_path, column, value):
|
|
path = str(tmp_path / "invalid.sqlite3")
|
|
init_db(path)
|
|
values = {"month": "2026-09", "basis_date": "2026-09-30", "version": 1}
|
|
values[column] = value
|
|
with sqlite3.connect(path) as conn, pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(
|
|
"INSERT INTO monthly_settlement (id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) VALUES (?,?,?,?,?,?,?,?,?)",
|
|
("bad", values["month"], values["basis_date"], "draft", values["version"], "u", "t", "u", "t"),
|
|
)
|
|
|
|
|
|
def test_status_condition_checks_and_month_uniqueness(tmp_path):
|
|
path = str(tmp_path / "conditions.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(_insert_sql(), ("m1", "2026-09", "2026-09-30"))
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute(_insert_sql(), ("m2", "2026-09", "2026-09-30"))
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("UPDATE monthly_settlement SET status='approved' WHERE id='m1'")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("UPDATE monthly_settlement SET status='cancelled' WHERE id='m1'")
|
|
conn.execute("UPDATE monthly_settlement SET status='approved', approved_by='reviewer', approved_at='t', approved_version=1, approved_fingerprint='fp' WHERE id='m1'")
|
|
with pytest.raises(sqlite3.IntegrityError):
|
|
conn.execute("UPDATE monthly_settlement SET status='closed' WHERE id='m1'")
|