129 lines
4.6 KiB
Python
129 lines
4.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp import service
|
|
|
|
SNAPSHOT_PATH = Path(__file__).parent / "contract_snapshots" / "response_contract_v1_0.json"
|
|
TYPE_MAP = {
|
|
"bool": bool,
|
|
"dict": dict,
|
|
"float": float,
|
|
"int": int,
|
|
"list": list,
|
|
"none": type(None),
|
|
"str": str,
|
|
}
|
|
|
|
|
|
def _load_contract() -> dict[str, Any]:
|
|
with SNAPSHOT_PATH.open(encoding="utf-8") as f:
|
|
return json.load(f)
|
|
|
|
|
|
def _assert_type(value: Any, spec: str | list[str], path: str) -> None:
|
|
names = spec if isinstance(spec, list) else [spec]
|
|
expected = tuple(TYPE_MAP[name] for name in names)
|
|
# bool is a subclass of int; contract int should not silently accept bool.
|
|
if "int" in names and isinstance(value, bool):
|
|
pytest.fail(f"{path} expected int-compatible contract type, got bool")
|
|
assert isinstance(value, expected), f"{path} expected {names}, got {type(value).__name__}"
|
|
|
|
|
|
def _assert_subset_schema(payload: dict[str, Any], schema: dict[str, Any], path: str) -> None:
|
|
for key, spec in schema.get("required", {}).items():
|
|
assert key in payload, f"{path}.{key} missing from response contract"
|
|
value = payload[key]
|
|
if isinstance(spec, dict):
|
|
_assert_type(value, spec["type"], f"{path}.{key}")
|
|
if spec.get("type") == "dict":
|
|
_assert_subset_schema(value, spec, f"{path}.{key}")
|
|
else:
|
|
_assert_type(value, spec, f"{path}.{key}")
|
|
|
|
for list_key, item_schema in schema.get("items", {}).items():
|
|
items = payload.get(list_key, [])
|
|
assert isinstance(items, list), f"{path}.{list_key} expected list for item schema"
|
|
for index, item in enumerate(items):
|
|
assert isinstance(item, dict), f"{path}.{list_key}[{index}] expected dict"
|
|
_assert_subset_schema(item, item_schema, f"{path}.{list_key}[{index}]")
|
|
|
|
|
|
def _build_contract_fixture(db_path: str) -> None:
|
|
account = service.create_ledger_account(
|
|
"Synthetic Checking",
|
|
type="bank",
|
|
institution="Synthetic Bank",
|
|
account_number_masked="SYN-0001",
|
|
last_balance=1_000_000,
|
|
last_synced_at="2026-06-30T12:00:00+09:00",
|
|
statement_status="manual_only",
|
|
liquidity_bucket="cash",
|
|
db_path=db_path,
|
|
)
|
|
assert account["success"] is True
|
|
account_id = account["account"]["id"]
|
|
assert service.create_ledger_transaction(
|
|
"2026-06-01",
|
|
"income",
|
|
3_000_000,
|
|
category="수입",
|
|
merchant="Synthetic Employer",
|
|
account_id=account_id,
|
|
db_path=db_path,
|
|
)["success"] is True
|
|
assert service.create_ledger_transaction(
|
|
"2026-06-02",
|
|
"expense",
|
|
50_000,
|
|
category="식비",
|
|
merchant="Synthetic Market",
|
|
account_id=account_id,
|
|
db_path=db_path,
|
|
)["success"] is True
|
|
|
|
|
|
def test_response_contract_metadata_versioning_policy() -> None:
|
|
contract = _load_contract()
|
|
assert contract["contract_version"] == "1.0"
|
|
assert contract["runtime_exposure"] == "unexposed"
|
|
assert "remove_required_key" in contract["policy"]["breaking_changes"]
|
|
assert "add_optional_field" in contract["policy"]["additive_changes"]
|
|
|
|
|
|
def test_health_and_status_success_response_contract(tmp_path: Path) -> None:
|
|
contract = _load_contract()["scopes"]
|
|
db_path = str(tmp_path / "contract-fixture.sqlite3")
|
|
_build_contract_fixture(db_path)
|
|
|
|
health = service.health_check(db_path)
|
|
assert health["success"] is True
|
|
_assert_subset_schema(health, contract["health"], "health")
|
|
|
|
status = service.get_ledger_db_status(db_path, include_hash=False, include_integrity=False)
|
|
assert status["success"] is True
|
|
_assert_subset_schema(status, contract["status"], "status")
|
|
|
|
|
|
def test_list_report_and_mutation_success_envelope_contract(tmp_path: Path) -> None:
|
|
contract = _load_contract()["scopes"]
|
|
db_path = str(tmp_path / "contract-fixture.sqlite3")
|
|
_build_contract_fixture(db_path)
|
|
|
|
mutation = service.create_ledger_account("Synthetic Savings", type="bank", liquidity_bucket="deposit", db_path=db_path)
|
|
assert mutation["success"] is True
|
|
_assert_subset_schema(mutation, contract["mutation_success_envelope"], "mutation_success_envelope")
|
|
|
|
accounts = service.list_ledger_accounts(db_path=db_path)
|
|
assert accounts["success"] is True
|
|
assert accounts["count"] >= 2
|
|
_assert_subset_schema(accounts, contract["list"], "list")
|
|
|
|
report = service.get_monthly_cashflow_report("2026-06", db_path=db_path)
|
|
assert report["success"] is True
|
|
_assert_subset_schema(report, contract["report"], "report")
|