147 lines
6.5 KiB
Python
147 lines
6.5 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
BASIS = "2026-06-30"
|
|
|
|
|
|
def _legacy_closed_db(tmp_path):
|
|
path = str(tmp_path / "legacy.sqlite3")
|
|
init_db(path)
|
|
asset = service.create_manual_asset("legacy cash", 100, "2026-06-01", "cash", db_path=path)["manual_asset"]
|
|
with sqlite3.connect(path) as conn:
|
|
# These are schema-11 close rows after the additive schema-12 migration:
|
|
# run identity is absent, but row close/correction fields remain frozen.
|
|
conn.execute(
|
|
"""INSERT INTO asset_snapshots
|
|
(id, basis_date, source_type, source_id, name, amount, liquidity_bucket,
|
|
valuation_source, snapshot_run_id, correction_of_run_id, correction_reason,
|
|
is_correction, closed_at, created_at)
|
|
VALUES (?, ?, 'manual_asset', ?, ?, 100, 'cash', 'manual_assets', 'legacy-run-correction',
|
|
'legacy-initial', 'legacy correction', 1, '2026-06-30T12:00:00+09:00',
|
|
'2026-06-30T12:00:00+09:00')""",
|
|
("legacy-row", BASIS, asset["id"], asset["name"]),
|
|
)
|
|
conn.execute("DROP TABLE asset_snapshot_runs")
|
|
conn.execute("UPDATE schema_meta SET value='11' WHERE key='schema_version'")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
return path, asset["id"]
|
|
|
|
|
|
def test_schema11_closed_rows_remain_frozen_after_live_value_and_date_change(tmp_path):
|
|
path, asset_id = _legacy_closed_db(tmp_path)
|
|
service.update_manual_asset(asset_id, valuation_amount=900, valuation_date="2026-06-30", db_path=path)
|
|
before = hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
report = service.get_asset_snapshot_report(BASIS, db_path=path)
|
|
|
|
after = hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
assert report["success"] is True
|
|
assert report["source"] == "legacy_asset_snapshots"
|
|
assert report["is_closed_snapshot"] is True
|
|
assert report["close_status"] == "legacy_read_only"
|
|
assert report["net_worth"] == 100
|
|
assert report["included_items"][0]["amount"] == 100
|
|
assert report["correction"] == {
|
|
"is_correction": True,
|
|
"correction_of_run_id": "legacy-initial",
|
|
"correction_reason": "legacy correction",
|
|
}
|
|
assert report["snapshot_run_id"] == "legacy-run-correction"
|
|
assert report["approval"] == {}
|
|
assert "valuation_id" not in report["included_items"][0]
|
|
assert after == before
|
|
|
|
|
|
def test_schema11_closed_rows_are_frozen_in_timeseries(tmp_path):
|
|
path, asset_id = _legacy_closed_db(tmp_path)
|
|
service.update_manual_asset(asset_id, valuation_amount=900, valuation_date="2026-07-01", db_path=path)
|
|
|
|
point = service.get_net_worth_timeseries(BASIS, BASIS, db_path=path)["points"][0]
|
|
|
|
assert point["net_worth"] == 100
|
|
assert point["item_count"] == 1
|
|
assert point["close_status"] == "legacy_read_only"
|
|
assert point["snapshot_run_id"] == "legacy-run-correction"
|
|
assert point["approval"] == {}
|
|
assert point["correction"] == {
|
|
"is_correction": True,
|
|
"correction_of_run_id": "legacy-initial",
|
|
"correction_reason": "legacy correction",
|
|
}
|
|
|
|
|
|
def test_schema11_closed_rows_are_frozen_in_monthly_settlement(tmp_path):
|
|
path, asset_id = _legacy_closed_db(tmp_path)
|
|
service.update_manual_asset(asset_id, valuation_amount=900, valuation_date="2026-07-01", db_path=path)
|
|
|
|
monthly = service.get_monthly_settlement_summary("2026-06", db_path=path)
|
|
assets = monthly["reports"]["assets"]
|
|
coverage = monthly["source_coverage"]["assets"]
|
|
|
|
assert assets["net_worth"] == 100
|
|
assert assets["is_closed_snapshot"] is True
|
|
assert coverage["net_worth"] == 100
|
|
assert coverage["is_closed_snapshot"] is True
|
|
assert coverage["monthly_source_snapshot_id"] == "legacy-run-correction"
|
|
assert assets["approval"] == {}
|
|
assert coverage["approval"] == {}
|
|
assert coverage["correction"] == assets["correction"]
|
|
|
|
|
|
def test_schema11_latest_legacy_correction_run_is_selected_as_a_complete_group(tmp_path):
|
|
path = str(tmp_path / "legacy-corrections.sqlite3")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
rows = [
|
|
("old-a", "legacy-run-1", "", "", 0, "2026-06-30T10:00:00+09:00", 40),
|
|
("old-b", "legacy-run-1", "", "", 0, "2026-06-30T10:00:00+09:00", 60),
|
|
("new-a", "legacy-run-2", "legacy-run-1", "correct frozen close", 1, "2026-07-01T10:00:00+09:00", 70),
|
|
("new-b", "legacy-run-2", "legacy-run-1", "correct frozen close", 1, "2026-07-01T10:00:00+09:00", 80),
|
|
]
|
|
for row_id, run_id, prior, reason, correction, closed_at, amount in rows:
|
|
conn.execute(
|
|
"""INSERT INTO asset_snapshots
|
|
(id, basis_date, source_type, source_id, name, amount, currency,
|
|
liquidity_bucket, valuation_source, snapshot_run_id,
|
|
correction_of_run_id, correction_reason, is_correction,
|
|
closed_at, created_at)
|
|
VALUES (?, ?, 'manual_asset', ?, ?, ?, 'KRW', 'cash',
|
|
'manual_assets', ?, ?, ?, ?, ?, ?)""",
|
|
(row_id, BASIS, row_id, row_id, amount, run_id, prior, reason,
|
|
correction, closed_at, closed_at),
|
|
)
|
|
conn.execute("DROP TABLE asset_snapshot_runs")
|
|
conn.execute("UPDATE schema_meta SET value='11' WHERE key='schema_version'")
|
|
init_db(path)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM asset_snapshot_runs").fetchone()[0] == 0
|
|
|
|
before = hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
report = service.get_asset_snapshot_report(BASIS, db_path=path)
|
|
after = hashlib.sha256(Path(path).read_bytes()).hexdigest()
|
|
|
|
assert report["success"] is True
|
|
assert report["source"] == "legacy_asset_snapshots"
|
|
assert report["snapshot_run_id"] == "legacy-run-2"
|
|
assert report["snapshot_count"] == 2
|
|
assert [item["id"] for item in report["included_items"]] == ["new-a", "new-b"]
|
|
assert report["net_worth"] == 150
|
|
assert report["correction"] == {
|
|
"is_correction": True,
|
|
"correction_of_run_id": "legacy-run-1",
|
|
"correction_reason": "correct frozen close",
|
|
}
|
|
assert report["approval"] == {}
|
|
assert all("valuation_id" not in item for item in report["included_items"])
|
|
assert after == before
|