146 lines
6.4 KiB
Python
146 lines
6.4 KiB
Python
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp.application.account_valuation_service import register_account_valuation
|
|
from jarca_ledger_mcp.application.monthly_settlement_service import create_monthly_settlement
|
|
from jarca_ledger_mcp.infrastructure.sqlite.migrations import init_db
|
|
|
|
|
|
@pytest.fixture
|
|
def ledger(tmp_path):
|
|
path = tmp_path / "actual-service.sqlite3"
|
|
init_db(str(path))
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"INSERT INTO accounts(id,name,currency,created_at,updated_at) VALUES(?,?,?,?,?)",
|
|
("acct-1", "주계좌", "KRW", "2026-10-01T00:00:00Z", "2026-10-01T00:00:00Z"),
|
|
)
|
|
conn.commit()
|
|
settlement = create_monthly_settlement("2026-10", db_path=str(path))
|
|
return path, settlement["settlement_id"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("amount", "expected_amount"),
|
|
[(0, 0), (123456, 123456)],
|
|
)
|
|
def test_actual_service_saves_zero_and_amount_with_fk_and_unverified(ledger, amount, expected_amount):
|
|
path, settlement_id = ledger
|
|
result = register_account_valuation(
|
|
"acct-1", "2026-10-31", amount, "KRW", monthly_settlement_id=settlement_id, db_path=str(path)
|
|
)
|
|
assert result["success"] is True
|
|
with sqlite3.connect(path) as conn:
|
|
row = conn.execute(
|
|
"SELECT amount,currency,basis_date,monthly_settlement_id,verification_status,source_type,source_id FROM account_snapshot"
|
|
).fetchone()
|
|
assert row == (expected_amount, "KRW", "2026-10-31", settlement_id, "unverified", "account", "acct-1")
|
|
|
|
|
|
@pytest.mark.parametrize("existing_amount", [None, 0])
|
|
def test_actual_service_rejects_duplicate_even_when_existing_amount_is_null_or_zero(ledger, existing_amount):
|
|
path, settlement_id = ledger
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"""INSERT INTO account_snapshot
|
|
(id,basis_date,source_type,source_id,name,amount,currency,monthly_settlement_id,created_at)
|
|
VALUES(?,?,?,?,?,?,?,?,?)""",
|
|
("existing", "2026-10-31", "account", "acct-1", "주계좌", existing_amount, "KRW", settlement_id, "2026-10-31T00:00:00Z"),
|
|
)
|
|
conn.commit()
|
|
result = register_account_valuation(
|
|
account_id="acct-1", basis_date="2026-10-31", amount=99, currency="KRW",
|
|
monthly_settlement_id=settlement_id, db_path=str(path),
|
|
)
|
|
assert result == {
|
|
"success": False,
|
|
"error_code": "duplicate",
|
|
"error": "account evaluation already registered",
|
|
"snapshot_id": "existing",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["closed", "approved"])
|
|
def test_actual_service_blocks_closed_and_approved_settlements(ledger, status):
|
|
path, settlement_id = ledger
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(
|
|
"""UPDATE monthly_settlement
|
|
SET status=?, approved_by='test', approved_at='2026-10-31T00:00:00Z',
|
|
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-31T00:00:00Z' ELSE closed_at END
|
|
WHERE id=?""",
|
|
(status, status, status, settlement_id),
|
|
)
|
|
conn.commit()
|
|
result = register_account_valuation(
|
|
"acct-1", "2026-10-31", 10, "KRW", monthly_settlement_id=settlement_id, db_path=str(path)
|
|
)
|
|
assert result["success"] is False
|
|
assert result["error_code"] in {"closed_settlement", "invalid_settlement_status"}
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM account_snapshot").fetchone()[0] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("kwargs", "error_code"),
|
|
[
|
|
({"currency": "USD"}, "validation_error"),
|
|
({"basis_date": "2026-09-30"}, "validation_error"),
|
|
({"monthly_settlement_id": "missing"}, "not_found"),
|
|
],
|
|
)
|
|
def test_actual_service_validates_currency_date_and_settlement_and_rolls_back(ledger, kwargs, error_code):
|
|
path, settlement_id = ledger
|
|
args = {
|
|
"account_id": "acct-1", "basis_date": "2026-10-31", "amount": 10, "currency": "KRW",
|
|
"monthly_settlement_id": settlement_id, "db_path": str(path),
|
|
}
|
|
args.update(kwargs)
|
|
result = register_account_valuation(**args)
|
|
assert result["success"] is False
|
|
assert result["error_code"] == error_code
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM account_snapshot").fetchone()[0] == 0
|
|
assert conn.execute("SELECT COUNT(*) FROM monthly_settlement").fetchone()[0] == 1
|
|
|
|
|
|
def test_mcp_preserves_legacy_full_argument_contract_and_monthly_dispatch(ledger, monkeypatch):
|
|
from jarca_ledger_mcp.adapters.mcp import server
|
|
path, settlement_id = ledger
|
|
monkeypatch.setenv("JARCA_LEDGER_DB_PATH", str(path))
|
|
legacy = server.register_account_valuation(
|
|
"acct-1", "2026-10-31", 150, "KRW", "balance", "manual", "statement-1",
|
|
"verified", "legacy memo", "legacy-key", "2026-10-31T10:00:00+09:00",
|
|
"Asia/Seoul", 100, 100, "cost_basis", "Initial cost", 50, 0.5,
|
|
"cost_basis", None, None, None, None,
|
|
)
|
|
assert legacy["success"], legacy
|
|
assert legacy["valuation"]["cost_basis"] == 100
|
|
assert legacy["valuation"]["source_ref"] == "statement-1"
|
|
monthly = server.register_account_valuation(
|
|
"acct-1", "2026-10-31", 150, monthly_settlement_id=settlement_id,
|
|
liquidity_bucket="cash",
|
|
)
|
|
assert monthly["success"], monthly
|
|
assert monthly["verification_status"] == "unverified"
|
|
valuations = server.list_account_valuations("acct-1")["valuations"]
|
|
assert {item["id"] for item in valuations} == {
|
|
legacy["valuation"]["id"], monthly["snapshot_id"],
|
|
}
|
|
assert valuations == sorted(
|
|
valuations,
|
|
key=lambda item: (item["basis_date"], item["recorded_at"], item["id"]),
|
|
)
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute("SELECT COUNT(*) FROM account_valuations").fetchone() == (1,)
|
|
assert conn.execute("SELECT COUNT(*) FROM account_snapshot").fetchone() == (1,)
|
|
advertised = server.mcp._tool_manager._tools["register_account_valuation"].parameters
|
|
assert {"source_ref", "cost_basis", "monthly_settlement_id"} <= advertised["properties"].keys()
|
|
assert "db_path" not in advertised["properties"]
|
|
assert {"create_manual_asset", "list_manual_assets", "update_manual_asset", "deactivate_manual_asset"} <= server.mcp._tool_manager._tools.keys()
|