119 lines
5.7 KiB
Python
Executable File
119 lines
5.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Synthetic fixture DB builder for CI/AppTest/smoke checks.
|
|
|
|
The fixture intentionally contains only synthetic identifiers and summary data.
|
|
It must never point at the operating ledger database.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from jarca_ledger_mcp import service
|
|
from jarca_ledger_mcp.db import DEFAULT_DB_PATH, init_db
|
|
|
|
|
|
def assert_ci_fixture_db_path(path: Path) -> Path:
|
|
resolved = path.expanduser().resolve()
|
|
if resolved == DEFAULT_DB_PATH.resolve():
|
|
raise ValueError("refusing to use the operating ledger DB as a CI fixture")
|
|
if resolved.exists() and resolved.is_dir():
|
|
raise ValueError(f"fixture DB path is a directory: {resolved}")
|
|
allowed_names = {"fixture.sqlite3", "ci-ledger-fixture.sqlite3", "ledger-fixture.sqlite3"}
|
|
if resolved.name not in allowed_names:
|
|
raise ValueError(f"CI fixture DB filename must be one of {sorted(allowed_names)}: {resolved.name}")
|
|
return resolved
|
|
|
|
|
|
def create_ci_fixture_db(db_path: str | Path) -> dict[str, Any]:
|
|
path = assert_ci_fixture_db_path(Path(db_path))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
if path.exists():
|
|
path.unlink()
|
|
init_db(str(path))
|
|
service.seed_default_ledger_categories(db_path=str(path))
|
|
service.seed_default_category_rules(db_path=str(path))
|
|
|
|
account = service.create_ledger_account(
|
|
"CI Synthetic Checking",
|
|
institution="CI Synthetic Bank",
|
|
holder_name="CI Synthetic Holder",
|
|
last_balance=1_250_000,
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
statement_status="partial_import",
|
|
liquidity_bucket="cash",
|
|
memo="ci synthetic fixture account",
|
|
db_path=str(path),
|
|
)
|
|
if not account.get("success"):
|
|
raise RuntimeError(f"fixture account seed failed: {account.get('error_code')}")
|
|
account_id = account["account"]["id"]
|
|
|
|
for account_spec in [
|
|
dict(name="CI Synthetic Investment", type="investment", institution="CI Synthetic Securities", last_balance=5_000_000, liquidity_bucket="investment"),
|
|
dict(name="CI Synthetic Loan", type="loan", institution="CI Synthetic Bank", last_balance=2_000_000, liquidity_bucket="loan"),
|
|
dict(name="CI Synthetic Manual Asset", type="other", institution="Manual", liquidity_bucket="real_estate"),
|
|
]:
|
|
result = service.create_ledger_account(
|
|
last_synced_at="2026-06-30T23:59:00+09:00",
|
|
memo="ci synthetic fixture account",
|
|
db_path=str(path),
|
|
**account_spec,
|
|
)
|
|
if not result.get("success"):
|
|
raise RuntimeError(f"fixture account seed failed: {result.get('error_code')}")
|
|
|
|
budget = service.create_ledger_budget("2026-06", "식비", 50_000, memo="ci synthetic budget", db_path=str(path))
|
|
if not budget.get("success"):
|
|
raise RuntimeError(f"fixture budget seed failed: {budget.get('error_code')}")
|
|
|
|
tx_specs: list[dict[str, Any]] = [
|
|
dict(transaction_date="2026-06-25", type="income", amount=2_000_000, merchant="CI_SYNTHETIC_SALARY", category="수입", subcategory="급여", source="manual", memo="ci synthetic income"),
|
|
dict(transaction_date="2026-06-26", type="expense", amount=35_000, merchant="CI_SYNTHETIC_DINING", category="식비", subcategory="외식", source="manual", memo="ci synthetic expense"),
|
|
dict(transaction_date="2026-06-27", type="expense", amount=18_500, merchant="CI_SYNTHETIC_MART", category="", subcategory="", source="manual", memo="ci synthetic uncategorized", raw_summary="ci synthetic grocery"),
|
|
dict(transaction_date="2026-06-28", type="expense", amount=4_800, merchant="CI_SYNTHETIC_CAFE", category="", subcategory="", source="manual", memo="ci synthetic uncategorized", raw_summary="ci synthetic cafe"),
|
|
dict(transaction_date="2026-06-29", type="expense", amount=120_000, merchant="CI_SYNTHETIC_TRANSFER", category="이체", subcategory="확인필요", source="kb_bank_pdf", counterparty="CI_SYNTHETIC_TOSS", memo="cross_transfer_review: ci synthetic", raw_summary="ci synthetic transfer", balance_after=1_250_000),
|
|
]
|
|
transaction_ids: list[str] = []
|
|
for index, spec in enumerate(tx_specs, start=1):
|
|
result = service.create_ledger_transaction(
|
|
account_id=account_id,
|
|
external_id=f"ci-fixture-{index}",
|
|
db_path=str(path),
|
|
**spec,
|
|
)
|
|
if not result.get("success"):
|
|
raise RuntimeError(f"fixture transaction seed failed: {result.get('error_code')}")
|
|
transaction_ids.append(result["transaction"]["id"])
|
|
|
|
status = service.get_ledger_db_status(str(path), include_hash=False, include_integrity=True)
|
|
if not status.get("success") or status.get("integrity_check") != "ok":
|
|
raise RuntimeError(f"fixture health check failed: {status.get('error_code')}")
|
|
|
|
return {
|
|
"success": True,
|
|
"fixture_db": str(path),
|
|
"account_count": 4,
|
|
"transaction_count": len(transaction_ids),
|
|
"integrity_check": status.get("integrity_check"),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="Create a synthetic Jarca Ledger CI fixture DB.")
|
|
parser.add_argument("--db", required=True, help="Output fixture path, e.g. $RUNNER_TEMP/fixture.sqlite3")
|
|
args = parser.parse_args()
|
|
try:
|
|
result = create_ci_fixture_db(args.db)
|
|
except Exception as exc: # noqa: BLE001 - CLI wrapper reports deterministic failure
|
|
print(json.dumps({"success": False, "error_code": "fixture_seed_failed", "error": str(exc)}, ensure_ascii=False))
|
|
return 1
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|