68 lines
3.5 KiB
Python
68 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp.application.account_valuation_service import list_account_valuations, register_account_valuation
|
|
from jarca_ledger_mcp.application.assets_service import close_asset_snapshot, get_asset_snapshot_report
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import list_monthly_settlement_overview
|
|
from jarca_ledger_mcp.infrastructure.sqlite.migrations import init_db
|
|
|
|
|
|
def tables(path):
|
|
with sqlite3.connect(path) as c:
|
|
return {r[0] for r in c.execute("select name from sqlite_master where type='table'")}
|
|
|
|
|
|
def test_fresh_init_retains_asset_snapshot_runs(tmp_path):
|
|
path = tmp_path / "fresh.sqlite3"
|
|
init_db(str(path))
|
|
assert "asset_snapshot_runs" in tables(path)
|
|
|
|
|
|
def test_init_twice_is_idempotent_and_schema15(tmp_path):
|
|
path = tmp_path / "repeat.sqlite3"
|
|
init_db(str(path)); init_db(str(path))
|
|
with sqlite3.connect(path) as c:
|
|
assert c.execute("select value from schema_meta where key='schema_version'").fetchone()[0] == "15"
|
|
assert "asset_snapshot_runs" in tables(path)
|
|
|
|
|
|
def test_legacy_child_rows_require_deterministic_parent_mapping(tmp_path):
|
|
path = tmp_path / "legacy.sqlite3"
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as c:
|
|
c.execute("insert into asset_snapshot_runs (id,basis_date,close_status,created_at) values ('run-1','2026-09-30','complete_exact','t')")
|
|
c.execute("insert into monthly_settlement(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) values ('month-1','2026-09','2026-09-30','draft',1,'test','now','test','now')")
|
|
c.execute("insert into settlement_cashflow_snapshots (id,monthly_settlement_id,cashflow_kind,category,subcategory,amount,transaction_count,currency) values ('cf-1','month-1','income','','',10,1,'KRW')")
|
|
c.commit()
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as c:
|
|
assert c.execute("select monthly_settlement_id, amount from settlement_cashflow_snapshots").fetchone() == ('month-1', 10)
|
|
assert "asset_snapshot_runs" in tables(path)
|
|
assert "monthly_settlement_id" in (c.execute("select sql from sqlite_master where name='settlement_cashflow_snapshots'").fetchone()[0] or "")
|
|
|
|
|
|
def test_unsupported_legacy_operations_do_not_touch_missing_db(tmp_path):
|
|
path = tmp_path / "missing.sqlite3"
|
|
assert register_account_valuation("a", "invalid-date", 1, db_path=str(path))["error_code"] == "validation_error"
|
|
assert list_account_valuations("a", db_path=str(path))["error_code"] == "db_not_initialized"
|
|
assert not path.exists()
|
|
|
|
|
|
def test_invalid_close_and_read_do_not_create_db(tmp_path):
|
|
path = tmp_path / "missing.sqlite3"
|
|
assert close_asset_snapshot("invalid-date", db_path=str(path))["error_code"] == "validation_error"
|
|
assert get_asset_snapshot_report("2026-09-30", db_path=str(path))["error_code"] == "db_not_initialized"
|
|
assert not path.exists()
|
|
|
|
|
|
def test_monthly_overview_keeps_real_parent_row(tmp_path):
|
|
path = tmp_path / "monthly.sqlite3"
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as c:
|
|
c.execute("insert into monthly_settlement(id,month,basis_date,status,version,created_by,created_at,updated_by,updated_at) values (?,?,?,?,?,?,?,?,?)", ('m-1','2026-09','2026-09-30','draft',1,'test','t','test','t'))
|
|
c.commit()
|
|
result = list_monthly_settlement_overview(db_path=str(path))
|
|
assert result["success"] and result["months"][0]["settlement_id"] == "m-1"
|
|
assert result["months"][0]["income_total"] is None
|