99 lines
6.4 KiB
Python
99 lines
6.4 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
from streamlit.testing.v1 import AppTest
|
|
|
|
from jarca_ledger_mcp.application.account_valuation_service import (
|
|
register_account_valuation,
|
|
update_account_snapshot_category,
|
|
)
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import create_monthly_settlement
|
|
from jarca_ledger_mcp.infrastructure.sqlite.migrations import init_db
|
|
|
|
|
|
def _ledger(tmp_path, *, account_type="bank", bucket="cash"):
|
|
path = tmp_path / "category.sqlite3"
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"INSERT INTO accounts(id,name,type,currency,liquidity_bucket,created_at,updated_at) VALUES(?,?,?,?,?,?,?)",
|
|
("acct-1", "계좌", account_type, "KRW", bucket, "2026-10-01", "2026-10-01"),
|
|
)
|
|
conn.commit()
|
|
settlement = create_monthly_settlement("2026-10", db_path=str(path))["settlement_id"]
|
|
return path, settlement
|
|
|
|
|
|
def test_registration_uses_account_default_and_rejects_invalid_explicit_override(tmp_path):
|
|
path, settlement = _ledger(tmp_path, bucket="deposit")
|
|
result = register_account_valuation("acct-1", "2026-10-31", 10, "KRW", monthly_settlement_id=settlement, db_path=str(path))
|
|
assert result["success"]
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT liquidity_bucket FROM account_snapshot").fetchone() == ("deposit",)
|
|
invalid_dir = tmp_path / "invalid"
|
|
invalid_dir.mkdir()
|
|
path2, settlement2 = _ledger(invalid_dir, bucket="cash")
|
|
result = register_account_valuation("acct-1", "2026-10-31", 10, "KRW", monthly_settlement_id=settlement2, liquidity_bucket="bogus", db_path=str(path2))
|
|
assert result["success"] is False and result["error_code"] == "validation_error"
|
|
|
|
|
|
def test_category_edit_only_changes_bucket_and_preserves_everything_else(tmp_path):
|
|
path, settlement = _ledger(tmp_path)
|
|
result = register_account_valuation("acct-1", "2026-10-31", 123, "KRW", memo="keep", monthly_settlement_id=settlement, db_path=str(path))
|
|
snapshot = result["snapshot_id"]
|
|
with sqlite3.connect(path) as conn:
|
|
before = conn.execute("SELECT * FROM account_snapshot WHERE id=?", (snapshot,)).fetchone()
|
|
changed = update_account_snapshot_category(snapshot, settlement, "loan", db_path=str(path))
|
|
assert changed["success"] and changed["liquidity_bucket"] == "loan"
|
|
with sqlite3.connect(path) as conn:
|
|
after = conn.execute("SELECT * FROM account_snapshot WHERE id=?", (snapshot,)).fetchone()
|
|
cols = [row[1] for row in conn.execute("PRAGMA table_info(account_snapshot)")]
|
|
assert dict(zip(cols, before)) | {"liquidity_bucket": "loan"} == dict(zip(cols, after))
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["draft", "in_review"])
|
|
def test_category_edit_allows_draft_and_in_review(tmp_path, status):
|
|
path, settlement = _ledger(tmp_path)
|
|
snapshot = register_account_valuation("acct-1", "2026-10-31", 1, "KRW", monthly_settlement_id=settlement, db_path=str(path))["snapshot_id"]
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE monthly_settlement SET status=? WHERE id=?", (status, settlement))
|
|
conn.commit()
|
|
assert update_account_snapshot_category(snapshot, settlement, "investment", db_path=str(path))["success"]
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["approved", "closed"])
|
|
def test_category_edit_blocks_approved_and_closed(tmp_path, status):
|
|
path, settlement = _ledger(tmp_path)
|
|
snapshot = register_account_valuation("acct-1", "2026-10-31", 1, "KRW", monthly_settlement_id=settlement, db_path=str(path))["snapshot_id"]
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE monthly_settlement SET status=?, approved_by='test', approved_at='2026-10-31', approved_version=1, approved_fingerprint='fp', closed_by=CASE WHEN ?='closed' THEN 'test' ELSE closed_by END, closed_at=CASE WHEN ?='closed' THEN '2026-10-31' ELSE closed_at END WHERE id=?", (status, status, status, settlement))
|
|
conn.commit()
|
|
result = update_account_snapshot_category(snapshot, settlement, "investment", db_path=str(path))
|
|
assert result["success"] is False and result["error_code"] == "invalid_settlement_status"
|
|
|
|
|
|
def test_category_edit_rejects_wrong_monthly(tmp_path):
|
|
path, settlement = _ledger(tmp_path)
|
|
other = create_monthly_settlement("2026-11", db_path=str(path))["settlement_id"]
|
|
snapshot = register_account_valuation("acct-1", "2026-10-31", 1, "KRW", monthly_settlement_id=settlement, db_path=str(path))["snapshot_id"]
|
|
result = update_account_snapshot_category(snapshot, other, "investment", db_path=str(path))
|
|
assert result["success"] is False and result["error_code"] == "validation_error"
|
|
|
|
|
|
def test_app_category_select_saves_clicked_snapshot_and_monthly(monkeypatch):
|
|
from jarca_ledger_mcp.adapters.streamlit import overview_gateway, settlement_gateway
|
|
calls = []
|
|
monkeypatch.setattr(overview_gateway, "get_monthly_settlement_summary", lambda *a, **k: {"success": True, "month": "2026-10", "selected_month": "2026-10", "basis_date": "2026-10-31", "reports": {"cashflow": {}, "budget": {}, "assets": {}}})
|
|
monkeypatch.setattr(settlement_gateway, "get_monthly_settlement_record", lambda *a, **k: {"success": True, "settlement": {"id": "ms-1", "month": "2026-10", "status": "draft"}})
|
|
monkeypatch.setattr(settlement_gateway, "list_account_snapshots_for_settlement", lambda *a, **k: {"success": True, "closed": False, "status": "draft", "count": 1, "registered_count": 1, "unregistered_count": 0, "snapshots": [{"account_id": "a1", "account_name": "대출", "amount": 100, "currency": "KRW", "basis_date": "2026-10-31", "snapshot_id": "snap-1", "display_status": "확인", "account_type": "loan", "account_liquidity_bucket": "loan", "snapshot_account_category": "loan"}]})
|
|
monkeypatch.setattr(settlement_gateway, "update_account_snapshot_category", lambda snapshot_id, monthly_id, value, **k: calls.append((snapshot_id, monthly_id, value)) or {"success": True})
|
|
app = AppTest.from_string("import streamlit as st\nst.query_params.update(page='monthly_settlement', id='ms-1')\nfrom dashboard.page_views.monthly_settlement import page_monthly_settlement\npage_monthly_settlement(db_path='unused.sqlite3')\n").run()
|
|
dialog = [b for b in app.button if b.label == "✎"][0].click().run()
|
|
assert dialog.selectbox[0].value == "loan"
|
|
save = [button for button in dialog.button if button.label == "저장"][0]
|
|
dialog.selectbox[0].set_value("deposit")
|
|
save.click().run()
|
|
assert save.label == "저장"
|