142 lines
8.0 KiB
Python
142 lines
8.0 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import (
|
|
approve_monthly_settlement,
|
|
close_monthly_settlement,
|
|
create_monthly_settlement,
|
|
review_monthly_settlement,
|
|
)
|
|
from jarca_ledger_mcp.adapters.streamlit import settlement_gateway
|
|
from jarca_ledger_mcp.application.transaction_report_services import get_settlement_cashflow_snapshot
|
|
from jarca_ledger_mcp.infrastructure.sqlite.migrations import init_db
|
|
|
|
|
|
def _db(tmp_path, rows=()):
|
|
path = tmp_path / "monthly-cashflow-actual.sqlite3"
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as c:
|
|
for row in rows:
|
|
c.execute("INSERT INTO transactions(id,transaction_date,type,amount,currency,category,subcategory,account_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)", (*row, "2026-10-01T00:00:00Z", "2026-10-01T00:00:00Z"))
|
|
c.commit()
|
|
return path
|
|
|
|
|
|
def _row(i, typ="expense", amount=100, currency="KRW", category="Food"):
|
|
return (i, "2026-10-10", typ, amount, currency, category, "Main", "acct")
|
|
|
|
|
|
def _review_approve(path):
|
|
s = create_monthly_settlement("2026-10", db_path=str(path)); sid = s["settlement_id"]
|
|
reviewed = review_monthly_settlement(sid, db_path=str(path))
|
|
assert reviewed["success"] and reviewed["version"] == 1 and reviewed["review_fingerprint"]
|
|
approved = approve_monthly_settlement(sid, 1, reviewed["review_fingerprint"], db_path=str(path))
|
|
assert approved["success"] and approved["status"] == "approved" and approved["version"] == 1
|
|
return sid, reviewed, approved
|
|
|
|
|
|
def test_gateway_cancels_approval_against_actual_temp_db(tmp_path):
|
|
path = _db(tmp_path, [_row("t1")])
|
|
sid, _, _ = _review_approve(path)
|
|
result = settlement_gateway.cancel_monthly_settlement_approval(sid, 1, db_path=str(path))
|
|
assert result == {"success": True, "settlement_id": sid, "status": "in_review", "version": 2}
|
|
with sqlite3.connect(path) as c:
|
|
row = c.execute("SELECT status, version, approved_by, approved_at, approved_cashflow_fingerprint FROM monthly_settlement WHERE id=?", (sid,)).fetchone()
|
|
assert row == ("in_review", 2, None, None, None)
|
|
|
|
|
|
def test_review_approve_close_persists_and_freezes_original_ledger(tmp_path):
|
|
path = _db(tmp_path, [_row("t1")]); sid, reviewed, _ = _review_approve(path)
|
|
closed = close_monthly_settlement(sid, 1, db_path=str(path))
|
|
assert closed["success"] and closed["status"] == "closed" and closed["snapshot_count"] == 1
|
|
with sqlite3.connect(path) as c:
|
|
c.execute("UPDATE transactions SET amount=999 WHERE id='t1'"); c.commit()
|
|
saved = get_settlement_cashflow_snapshot(sid, db_path=str(path))
|
|
assert saved["saved"] and saved["categories"][0]["amount"] == 100
|
|
|
|
|
|
def test_review_then_ledger_edit_rejects_approval(tmp_path):
|
|
path = _db(tmp_path, [_row("t1")]); s = create_monthly_settlement("2026-10", db_path=str(path)); r = review_monthly_settlement(s["settlement_id"], db_path=str(path))
|
|
with sqlite3.connect(path) as c: c.execute("UPDATE transactions SET amount=101 WHERE id='t1'"); c.commit()
|
|
out = approve_monthly_settlement(s["settlement_id"], 1, r["review_fingerprint"], db_path=str(path))
|
|
assert not out["success"] and out["error_code"] == "stale_review"
|
|
|
|
|
|
def test_approve_then_offset_transaction_rejects_close_even_same_total(tmp_path):
|
|
path = _db(tmp_path, [_row("t1", amount=100)]); sid, _, _ = _review_approve(path)
|
|
with sqlite3.connect(path) as c:
|
|
c.execute("UPDATE transactions SET amount=50 WHERE id='t1'")
|
|
c.execute("INSERT INTO transactions(id,transaction_date,type,amount,currency,category,subcategory,account_id,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)", _row("t2", amount=50) + ("2026-10-01T00:00:00Z", "2026-10-01T00:00:00Z")); c.commit()
|
|
out = close_monthly_settlement(sid, 1, db_path=str(path))
|
|
assert not out["success"] and out["error_code"] == "stale_cashflow"
|
|
|
|
|
|
def test_null_fingerprint_close_rejects(tmp_path):
|
|
path = _db(tmp_path); sid, _, _ = _review_approve(path)
|
|
with sqlite3.connect(path) as c: c.execute("UPDATE monthly_settlement SET approved_cashflow_fingerprint=NULL WHERE id=?", (sid,)); c.commit()
|
|
out = close_monthly_settlement(sid, 1, db_path=str(path)); assert not out["success"] and out["error_code"] == "missing_fingerprint"
|
|
|
|
|
|
def test_close_snapshot_insert_failure_rolls_back_status(tmp_path, monkeypatch):
|
|
path = _db(tmp_path, [_row("t1")]); sid, _, _ = _review_approve(path)
|
|
with sqlite3.connect(path) as c: c.execute("CREATE TRIGGER fail_snapshot BEFORE INSERT ON settlement_cashflow_snapshots BEGIN SELECT RAISE(ABORT,'boom'); END;") ; c.commit()
|
|
out = close_monthly_settlement(sid, 1, db_path=str(path)); assert not out["success"]
|
|
with sqlite3.connect(path) as c: assert c.execute("SELECT status FROM monthly_settlement WHERE id=?", (sid,)).fetchone()[0] == "approved"
|
|
|
|
|
|
@pytest.mark.parametrize("currency", ["KRW", "USD"])
|
|
def test_multifx_rows_remain_separate(tmp_path, currency):
|
|
path = _db(tmp_path, [_row("a", amount=100, currency="KRW"), _row("b", amount=7, currency=currency)])
|
|
sid, _, _ = _review_approve(path); assert close_monthly_settlement(sid, 1, db_path=str(path))["success"]
|
|
saved = get_settlement_cashflow_snapshot(sid, db_path=str(path)); assert {x["currency"] for x in saved["categories"]} >= {"KRW", currency}
|
|
|
|
|
|
def test_empty_close_is_saved_without_fake_marker(tmp_path):
|
|
path = _db(tmp_path); sid, _, _ = _review_approve(path); closed = close_monthly_settlement(sid, 1, db_path=str(path)); assert closed["snapshot_count"] == 0
|
|
saved = get_settlement_cashflow_snapshot(sid, db_path=str(path)); assert saved["saved"] and saved["categories"] == []
|
|
|
|
|
|
def test_double_close_is_idempotent(tmp_path):
|
|
path = _db(tmp_path, [_row("t1")]); sid, _, _ = _review_approve(path); first = close_monthly_settlement(sid, 1, db_path=str(path)); second = close_monthly_settlement(sid, 999, db_path=str(path))
|
|
assert first["success"] and second["success"] and second["idempotent"] is True
|
|
|
|
|
|
def test_legacy_finalization_and_correction_preserve_closed_canonical_month(tmp_path):
|
|
from jarca_ledger_mcp import service
|
|
path = _db(tmp_path, [_row("t1")])
|
|
sid, _, _ = _review_approve(path)
|
|
assert close_monthly_settlement(sid, 1, db_path=str(path))["success"]
|
|
canonical = get_settlement_cashflow_snapshot(sid, db_path=str(path))
|
|
with sqlite3.connect(path) as conn:
|
|
parent = conn.execute("SELECT * FROM monthly_settlement WHERE id=?", (sid,)).fetchone()
|
|
asset = service.create_manual_asset("Legacy cash", 500, "2026-10-31", "cash", db_path=str(path))
|
|
assert asset["success"]
|
|
previous = ""
|
|
for number in range(2):
|
|
draft = service.create_settlement_draft(
|
|
"2026-10", "tester", f"legacy-draft-{number}", db_path=str(path),
|
|
correction_of_run_id=previous, correction_reason="correct" if previous else "",
|
|
)
|
|
assert draft["success"], draft
|
|
did = draft["draft"]["id"]
|
|
review = service.review_settlement_draft(did, db_path=str(path))
|
|
assert review["success"]
|
|
approved = service.approve_settlement_draft(
|
|
did, review["version"], review["review_fingerprint"], "tester", "reviewed", "fixture", db_path=str(path),
|
|
)
|
|
assert approved["success"], approved
|
|
closed = service.finalize_settlement_draft(
|
|
did, review["version"], review["review_fingerprint"], f"legacy-close-{number}", db_path=str(path),
|
|
)
|
|
assert closed["success"], closed
|
|
previous = closed["snapshot_run_id"]
|
|
assert get_settlement_cashflow_snapshot(sid, db_path=str(path)) == canonical
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT * FROM monthly_settlement WHERE id=?", (sid,)).fetchone() == parent
|
|
assert conn.execute("SELECT COUNT(DISTINCT settlement_run_id) FROM settlement_cashflow_snapshots").fetchone() == (2,)
|
|
assert conn.execute("SELECT COUNT(*) FROM settlement_cashflow_snapshots WHERE monthly_settlement_id=?", (sid,)).fetchone() == (1,)
|
|
assert conn.execute("PRAGMA foreign_key_check").fetchall() == []
|