151 lines
9.6 KiB
Python
151 lines
9.6 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def test_asset_snapshot_schema_is_additive_and_idempotent(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
init_db(db_path)
|
|
init_db(db_path)
|
|
|
|
with sqlite3.connect(db_path) as conn:
|
|
manual_cols = {row[1] for row in conn.execute("PRAGMA table_info(manual_assets)")}
|
|
snapshot_cols = {row[1] for row in conn.execute("PRAGMA table_info(asset_snapshots)")}
|
|
|
|
assert {"id", "name", "valuation_amount", "valuation_date", "liquidity_bucket", "source_note", "memo", "active"} <= manual_cols
|
|
assert {"id", "basis_date", "source_type", "source_id", "name", "amount", "liquidity_bucket", "valuation_source", "source_note", "memo", "snapshot_run_id", "correction_of_run_id", "correction_reason", "is_correction", "closed_at"} <= snapshot_cols
|
|
|
|
|
|
def test_manual_assets_snapshot_report_and_net_worth_formula(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
cash = service.create_ledger_account("현금", last_balance=1_000_000, last_synced_at="2026-06-30", liquidity_bucket="cash", db_path=db_path)
|
|
stocks = service.create_ledger_account("투자", type="investment", last_balance=2_000_000, liquidity_bucket="investment", db_path=db_path)
|
|
missing = service.create_ledger_account("미평가 계좌", liquidity_bucket="deposit", db_path=db_path)
|
|
assert cash["success"] and stocks["success"] and missing["success"]
|
|
assert service.register_account_valuation(cash["account"]["id"], "2026-06-30", 1_000_000, verification_status="verified", idempotency_key="cash-v1", db_path=db_path)["success"]
|
|
assert service.register_account_valuation(stocks["account"]["id"], "2026-06-30", 2_000_000, verification_status="verified", idempotency_key="stocks-v1", db_path=db_path)["success"]
|
|
|
|
house = service.create_manual_asset("전세보증금", 300_000_000, "2026-06-30", "real_estate", source_note="synthetic appraisal", db_path=db_path)
|
|
loan = service.create_manual_asset("주담대", 120_000_000, "2026-06-30", "loan", source_note="synthetic loan balance", db_path=db_path)
|
|
unvalued = service.create_manual_asset("평가대기 차량", None, "2026-06-30", "other", source_note="pending", db_path=db_path)
|
|
assert house["success"] and loan["success"] and unvalued["success"]
|
|
|
|
manual = service.list_manual_assets(db_path=db_path)
|
|
assert manual["success"] is True
|
|
assert manual["count"] == 3
|
|
assert {item["authority_source"] for item in manual["manual_assets"]} == {"manual_assets"}
|
|
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=db_path)
|
|
assert report["success"] is True
|
|
assert report["source"] == "live_authority"
|
|
assert report["valuation_policy"]["formula"] == "asset_buckets - loan"
|
|
assert report["total_assets"] == 303_000_000
|
|
assert report["total_liabilities"] == 120_000_000
|
|
assert report["net_worth"] == 183_000_000
|
|
assert {item["source_type"] for item in report["included_items"]} == {"account", "manual_asset"}
|
|
assert any(item["source_type"] == "manual_asset" and item["name"] == "평가대기 차량" for item in report["excluded_items"])
|
|
assert any(item["source_type"] == "account" and item["name"] == "미평가 계좌" for item in report["excluded_items"])
|
|
|
|
|
|
def test_snapshot_rows_are_append_only_and_timeseries_uses_same_formula(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_asset_snapshot("2026-05-31", "account", "acct-a", "현금", 1_000_000, "cash", source_note="synthetic", db_path=db_path)
|
|
service.create_asset_snapshot("2026-05-31", "manual_asset", "manual-home", "부동산", 200_000_000, "real_estate", source_note="synthetic", db_path=db_path)
|
|
service.create_asset_snapshot("2026-05-31", "manual_asset", "manual-loan", "대출", 80_000_000, "loan", source_note="synthetic", db_path=db_path)
|
|
service.create_asset_snapshot("2026-06-30", "account", "acct-a", "현금", 1_500_000, "cash", source_note="synthetic", db_path=db_path)
|
|
service.create_asset_snapshot("2026-06-30", "manual_asset", "manual-home", "부동산", 205_000_000, "real_estate", source_note="synthetic", db_path=db_path)
|
|
service.create_asset_snapshot("2026-06-30", "manual_asset", "manual-loan", "대출", 78_000_000, "loan", source_note="synthetic", db_path=db_path)
|
|
|
|
may = service.get_asset_snapshot_report("2026-05-31", db_path=db_path)
|
|
assert may["success"] is True
|
|
assert may["source"] == "legacy_asset_snapshots"
|
|
assert may["is_closed_snapshot"] is False
|
|
assert may["close_status"] == "legacy_read_only"
|
|
assert may["snapshot_count"] == 3
|
|
assert may["net_worth"] == 121_000_000
|
|
|
|
series = service.get_net_worth_timeseries(start_date="2026-05-01", end_date="2026-06-30", db_path=db_path)
|
|
assert series["success"] is True
|
|
assert [point["basis_date"] for point in series["points"]] == ["2026-05-31", "2026-06-30"]
|
|
assert [point["net_worth"] for point in series["points"]] == [121_000_000, 128_500_000]
|
|
|
|
|
|
def test_asset_liquidity_report_empty_new_tables_preserves_req005_fallback(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
service.create_ledger_account("현금성 계좌", last_balance=1_000_000, liquidity_bucket="cash", db_path=db_path)
|
|
service.create_ledger_account("투자 계좌", last_balance=3_000_000, liquidity_bucket="investment", db_path=db_path)
|
|
service.create_ledger_account("대출", last_balance=500_000, liquidity_bucket="loan", db_path=db_path)
|
|
|
|
report = service.get_asset_liquidity_report(db_path=db_path)
|
|
assert report["success"] is True
|
|
assert report["total_assets"] == 4_000_000
|
|
assert report["total_liabilities"] == 500_000
|
|
assert report["net_worth"] == 3_500_000
|
|
assert report["manual_asset_count"] == 0
|
|
assert report["source"] == "accounts.last_balance"
|
|
assert [item["source_type"] for item in report["accounts"]] == ["account", "account", "account"]
|
|
|
|
|
|
def test_manual_asset_update_deactivate_preview_and_close_correction_policy(tmp_path):
|
|
db_path = str(tmp_path / "ledger.sqlite3")
|
|
account = service.create_ledger_account("월말 현금", last_balance=1_000_000, last_synced_at="2026-06-30", liquidity_bucket="cash", db_path=db_path)
|
|
asset = service.create_manual_asset("전세보증금", 300_000_000, "2026-06-30", "real_estate", source_note="synthetic", db_path=db_path)
|
|
loan = service.create_manual_asset("대출", 120_000_000, "2026-06-30", "loan", source_note="synthetic", db_path=db_path)
|
|
dormant = service.create_manual_asset("해지 예금", 5_000_000, "2026-06-30", "deposit", source_note="synthetic", db_path=db_path)
|
|
assert account["success"] and asset["success"] and loan["success"] and dormant["success"]
|
|
assert service.register_account_valuation(account["account"]["id"], "2026-06-30", 1_000_000, verification_status="verified", idempotency_key="cash-v1", db_path=db_path)["success"]
|
|
|
|
updated = service.update_manual_asset(asset["manual_asset"]["id"], valuation_amount=310_000_000, memo="updated synthetic", db_path=db_path)
|
|
assert updated["success"] is True
|
|
assert updated["manual_asset"]["valuation_amount"] == 310_000_000
|
|
deactivated = service.deactivate_manual_asset(dormant["manual_asset"]["id"], db_path=db_path)
|
|
assert deactivated["success"] is True
|
|
assert deactivated["manual_asset"]["active"] is False
|
|
|
|
before = service.get_ledger_db_status(db_path, include_hash=False, include_integrity=False)["table_counts"].get("asset_snapshots", 0)
|
|
preview = service.preview_asset_snapshot_close("2026-06-30", db_path=db_path)
|
|
after = service.get_ledger_db_status(db_path, include_hash=False, include_integrity=False)["table_counts"].get("asset_snapshots", 0)
|
|
assert preview["success"] is True
|
|
assert preview["source"] == "live_authority"
|
|
assert preview["is_closed_snapshot"] is False
|
|
assert before == after
|
|
assert preview["total_assets"] == 311_000_000
|
|
assert preview["total_liabilities"] == 120_000_000
|
|
assert preview["net_worth"] == 191_000_000
|
|
|
|
closed = service.close_asset_snapshot("2026-06-30", source_note="synthetic close", db_path=db_path)
|
|
assert closed["success"] is True
|
|
assert closed["source"] == "snapshot"
|
|
assert closed["is_closed_snapshot"] is True
|
|
assert closed["is_correction"] is False
|
|
assert closed["inserted_count"] == 3
|
|
first_run = closed["snapshot_run_id"]
|
|
|
|
report = service.get_asset_snapshot_report("2026-06-30", db_path=db_path)
|
|
assert report["success"] is True
|
|
assert report["is_closed_snapshot"] is True
|
|
assert report["net_worth"] == closed["net_worth"]
|
|
assert report["close_metadata"]["snapshot_run_id"] == first_run
|
|
|
|
blocked = service.close_asset_snapshot("2026-06-30", db_path=db_path)
|
|
assert blocked["success"] is False
|
|
assert blocked["error_code"] == "validation_error"
|
|
|
|
service.update_manual_asset(loan["manual_asset"]["id"], valuation_amount=110_000_000, db_path=db_path)
|
|
corrected = service.close_asset_snapshot("2026-06-30", source_note="synthetic correction", correction_reason="synthetic correction test", db_path=db_path)
|
|
assert corrected["success"] is True
|
|
assert corrected["is_correction"] is True
|
|
assert corrected["correction_of_run_id"] == first_run
|
|
assert corrected["net_worth"] == 201_000_000
|
|
|
|
corrected_report = service.get_asset_snapshot_report("2026-06-30", db_path=db_path)
|
|
assert corrected_report["net_worth"] == 201_000_000
|
|
assert corrected_report["close_metadata"]["snapshot_run_id"] == corrected["snapshot_run_id"]
|
|
|
|
with sqlite3.connect(db_path) as conn:
|
|
row_count = conn.execute("SELECT COUNT(*) FROM asset_snapshots WHERE basis_date='2026-06-30'").fetchone()[0]
|
|
assert row_count == 6
|