104 lines
4.2 KiB
Python
104 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def test_review_decision_and_reconciliation_schema_are_additive_idempotent(tmp_path: Path) -> None:
|
|
db_path = str(tmp_path / "validation.sqlite3")
|
|
init_db(db_path)
|
|
init_db(db_path)
|
|
|
|
status = service.get_ledger_db_status(db_path, include_hash=False, include_integrity=True)
|
|
assert status["success"] is True
|
|
assert status["integrity_check"] == "ok"
|
|
assert status["table_counts"]["review_decisions"] == 0
|
|
assert status["table_counts"]["reconciliation_sessions"] == 0
|
|
assert service.health_check(db_path)["schema_version"] == "11"
|
|
|
|
|
|
def test_review_decision_round_trip_payload_sanitizes_sensitive_fields(tmp_path: Path) -> None:
|
|
db_path = str(tmp_path / "validation.sqlite3")
|
|
decision = service.upsert_review_decision(
|
|
"cross_transfer",
|
|
"tx_synthetic_1",
|
|
status="decided",
|
|
decision_type="account_transfer",
|
|
reason_code="matched_pair",
|
|
memo="synthetic memo",
|
|
operator="qc",
|
|
review_after="2026-07-31",
|
|
target_payload={"amount": 10000, "raw_payload": "SHOULD_NOT_PERSIST", "file_path": "/secret.pdf"},
|
|
db_path=db_path,
|
|
)
|
|
assert decision["success"] is True
|
|
saved = decision["review_decision"]
|
|
assert saved["status"] == "decided"
|
|
assert saved["decision_type"] == "account_transfer"
|
|
assert saved["target_payload"] == {"amount": 10000}
|
|
|
|
held = service.hold_review_decision("cross_transfer", "tx_synthetic_1", memo="hold", db_path=db_path)
|
|
assert held["success"] is True
|
|
assert held["review_decision"]["status"] == "held"
|
|
|
|
released = service.release_review_decision("cross_transfer", "tx_synthetic_1", memo="release", db_path=db_path)
|
|
assert released["success"] is True
|
|
assert released["review_decision"]["status"] == "pending"
|
|
|
|
listed = service.list_review_decisions("cross_transfer", "tx_synthetic_1", db_path=db_path)
|
|
assert listed["success"] is True
|
|
assert listed["count"] == 1
|
|
|
|
|
|
def test_reconciliation_session_round_trip_and_report_integration(tmp_path: Path) -> None:
|
|
db_path = str(tmp_path / "validation.sqlite3")
|
|
account = service.create_ledger_account(
|
|
"Synthetic Manual Account",
|
|
type="bank",
|
|
statement_status="manual_only",
|
|
last_balance=1_000_000,
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
db_path=db_path,
|
|
)["account"]
|
|
service.create_ledger_transaction("2026-06-01", "income", 700_000, account_id=account["id"], db_path=db_path)
|
|
|
|
session = service.upsert_reconciliation_session(
|
|
account["id"],
|
|
"2026-06-30",
|
|
expected_balance=1_000_000,
|
|
actual_balance=700_000,
|
|
status="open",
|
|
source_note="synthetic validation",
|
|
db_path=db_path,
|
|
)
|
|
assert session["success"] is True
|
|
assert session["reconciliation_session"]["difference"] == 300_000
|
|
|
|
report = service.get_balance_reconciliation_report(db_path=db_path)
|
|
assert report["success"] is True
|
|
item = report["items"][0]
|
|
assert item["reconciliation_session"]["id"] == session["session_id"]
|
|
assert item["status"] == "REVIEW_NEEDED"
|
|
|
|
closed = service.close_reconciliation_session(session_id=session["session_id"], db_path=db_path)
|
|
assert closed["success"] is True
|
|
assert closed["reconciliation_session"]["status"] == "closed"
|
|
indicators = service.get_review_operating_indicators(db_path=db_path)["indicators"]
|
|
assert indicators["last_reconciliation_closed_at"]
|
|
|
|
|
|
def test_additive_mcp_contract_surface_exposes_safe_review_tools() -> None:
|
|
from jarca_ledger_mcp.adapters.mcp import server
|
|
|
|
tool_names = set(server.mcp._tool_manager._tools.keys())
|
|
assert "list_review_decisions" in tool_names
|
|
assert "upsert_review_decision" in tool_names
|
|
assert "hold_review_decision" in tool_names
|
|
assert "release_review_decision" in tool_names
|
|
assert "list_reconciliation_sessions" in tool_names
|
|
assert "upsert_reconciliation_session" in tool_names
|
|
assert "close_reconciliation_session" in tool_names
|
|
assert not {name for name in tool_names if "delete_review" in name or "reset_review" in name}
|