277 lines
17 KiB
Python
277 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import json
|
|
import re
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).parents[1]
|
|
DOCS = ROOT / "docs"
|
|
REQ = DOCS / "requirements/req-003_account-valuation-history.md"
|
|
|
|
|
|
def text(path: str) -> str:
|
|
return (ROOT / path).read_text(encoding="utf-8")
|
|
|
|
|
|
def test_settlement_requirement_revision_is_additive_and_traceable() -> None:
|
|
source = REQ.read_text(encoding="utf-8")
|
|
_, raw, body = source.split("---", 2)
|
|
metadata = json.loads(raw)
|
|
assert metadata["id"] == "REQ-003"
|
|
assert metadata["revision"] == "1.2"
|
|
assert metadata["issue"] == "#18"
|
|
assert metadata["implementation_pr"] == "https://gitea.jsfamily.quest/jsfamily/jarca-ledger-mcp/pulls/19"
|
|
assert "#12" in body and "#13" in body
|
|
assert re.findall(r"^### REQ-003\.(\d+) —", body, re.M) == [f"{i:02d}" for i in range(1, 12)]
|
|
assert re.findall(r"^## (\d+)\. ", body, re.M) == [str(i) for i in range(1, 12)]
|
|
assert "schema 11→12" in body and "schema 12→13" in body
|
|
|
|
|
|
def test_draft_docs_cover_authority_dates_atomicity_and_read_only_contracts() -> None:
|
|
docs = "\n".join(text(path) for path in (
|
|
"docs/requirements/req-003_account-valuation-history.md", "docs/data-model.md",
|
|
"docs/interface-contract.md", "docs/project-structure.md", "docs/runtime-flow.md",
|
|
))
|
|
for marker in ("date_difference_days = valuation_date - basis_date", "age_days", "manual_substitute",
|
|
"candidate_amount", "authority amount NULL", "refresh_settlement_draft_targets",
|
|
"BEGIN IMMEDIATE", "persisted response", "one active draft", "correction_reason",
|
|
"unverified", "no-create", "transaction readiness"):
|
|
assert marker in docs, marker
|
|
|
|
|
|
def test_new_public_operations_are_fully_listed_from_contract() -> None:
|
|
interface = text("docs/interface-contract.md")
|
|
for operation in ("create_settlement_draft", "list_settlement_drafts", "get_settlement_draft",
|
|
"update_settlement_draft", "refresh_settlement_draft_targets", "set_settlement_draft_item",
|
|
"review_settlement_draft", "approve_settlement_draft", "finalize_settlement_draft",
|
|
"cancel_settlement_draft"):
|
|
assert f"{operation}(" in interface, operation
|
|
assert "review_fingerprint" in interface and "selection_mode" in interface
|
|
|
|
|
|
def test_comprehensive_migration_acceptance_has_explicit_legacy_test_mapping() -> None:
|
|
text = REQ.read_text(encoding="utf-8")
|
|
case = text.split("### REQ-003.11-AC26 —", 1)[1].split("### REQ-003.11-AC27 —", 1)[0]
|
|
assert "`tests/test_settlement_cashflow_schema.py::test_legacy_migration_is_idempotent_and_keeps_unstored_totals_null`" in case
|
|
|
|
|
|
def test_settlement_acceptance_cases_are_individually_named_and_real() -> None:
|
|
import ast
|
|
req = REQ.read_text(encoding="utf-8").split("---", 2)[2]
|
|
section = req.split("## 7. 인수 기준", 1)[1].split("## 8. 오류", 1)[0]
|
|
cases = re.findall(r"^### REQ-003\.11-AC(\d+) — (.+)$", section, re.M)
|
|
assert len(cases) == 27
|
|
assert [n for n, _ in cases] == [f"{i:02}" for i in range(1, 28)]
|
|
files = {str(p.relative_to(ROOT)): p for p in (ROOT / "tests").glob("test_*.py")}
|
|
for block in re.split(r"(?=^### REQ-003\.11-AC)", section, flags=re.M)[1:]:
|
|
ref = re.search(r"`(tests/[^`]+\.py)::(test_[A-Za-z0-9_]+)`", block)
|
|
assert ref, block.splitlines()[0]
|
|
path, func = ref.groups()
|
|
assert path in files, path
|
|
names = {n.name for n in ast.walk(ast.parse(files[path].read_text(encoding="utf-8"))) if isinstance(n, ast.FunctionDef)}
|
|
assert func in names, (path, func)
|
|
for label in ("판정 기준", "검증 분류", "상태 불변조건"):
|
|
assert label in block, (block.splitlines()[0], label)
|
|
assert "## 12." not in req
|
|
requirements = req.split("## 5. 요구사항", 1)[1].split("## 6.", 1)[0]
|
|
assert requirements.count("**영향 범위**") == 11
|
|
new_requirement = requirements.split("### REQ-003.11 —", 1)[1]
|
|
for dimension in ("인터페이스", "영속성", "후속 사용 주체", "마이그레이션", "호환성", "배포 전제 조건"):
|
|
assert f"- {dimension}:" in new_requirement
|
|
|
|
|
|
def test_settlement_metadata_names_match_actual_public_contract() -> None:
|
|
interface = text('docs/interface-contract.md').split('## settlement-draft-service', 1)[1]
|
|
model = text('docs/data-model.md').split('## settlement-drafts', 1)[1]
|
|
for field in ('valuation_date', 'recorded_at', 'date_difference_days', 'date_difference_direction', 'age_days', 'representative_liability_definition'):
|
|
assert field in interface and field in model, field
|
|
for field in ('allow_missing_exclusions', 'allow_carried_before', 'allow_manual_substitute'):
|
|
assert field in interface
|
|
assert 'test_date_metadata_public_contract_names_survive_all_closed_consumers' in interface
|
|
|
|
|
|
def test_draft_mcp_signature_documents_internal_database_boundary() -> None:
|
|
interface = text("docs/interface-contract.md").split("## settlement-draft-service", 1)[1]
|
|
assert re.search(r"db_path.*(?:internal|service|gateway).*(?:not|omitted|제외).*(?:MCP|inputSchema)", interface, re.I | re.S), \
|
|
"document must distinguish internal db_path from MCP inputSchema"
|
|
assert "db_path: str | None = None" in interface
|
|
service = ast.parse(text("jarca_ledger_mcp/application/settlement_draft_service.py"))
|
|
names = {n.name for n in ast.walk(service) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))}
|
|
assert "create_settlement_draft" in names
|
|
|
|
|
|
def test_legacy_close_fields_are_scoped_separately_from_draft_finalize() -> None:
|
|
interface = text("docs/interface-contract.md")
|
|
legacy = interface.split("## D04 snapshot close output contract", 1)[1].split("## settlement-draft-service", 1)[0]
|
|
draft = interface.split("## settlement-draft-service", 1)[1]
|
|
assert re.search(r"legacy direct close", legacy, re.I), "exact field list must be scoped to legacy close"
|
|
assert re.search(r"additive settlement-draft finalize response", legacy, re.I) and re.search(r"preserving common aliases", draft, re.I), "draft finalize output extension and retained common aliases must be distinguished"
|
|
assert not re.search(r"new.*(?:identical|same).*shape", draft, re.I)
|
|
|
|
|
|
def test_unregistered_liability_contract_separates_registration_from_unknown_debt() -> None:
|
|
interface = text("docs/interface-contract.md")
|
|
clause = interface.split("### Draft evidence public fields", 1)[1].split("## ", 1)[0]
|
|
assert re.search(r"unregistered_liabilities.*(?:no|not).*registered.*liabilit", clause, re.I | re.S)
|
|
assert re.search(r"not.*(?:unknown|real).*debt", clause, re.I | re.S)
|
|
assert re.search(r"not.*(?:proof.*debt.free|prove.*debt.free)", clause, re.I | re.S)
|
|
assert re.search(r"exclu.*(?:missing|completeness)|(?:missing|completeness).*exclu", clause, re.I | re.S)
|
|
|
|
|
|
def test_requirement_authority_scope_distinguishes_approved_drafts_and_automatic_selection() -> None:
|
|
req = REQ.read_text(encoding="utf-8")
|
|
constraints = req.split("## 6. 제약과 불변조건", 1)[1].split("## 7.", 1)[0]
|
|
assert re.search(r"automatic.*`exact`.*`as_of`", constraints, re.I), "future exclusion must explicitly scope automatic exact/as_of selection"
|
|
assert re.search(r"legacy direct close", constraints, re.I), \
|
|
"future exclusion must explicitly scope legacy direct close"
|
|
assert re.search(r"verified.*future.*manual_substitute", constraints, re.I | re.S), \
|
|
"verified future valuation may only enter an explicitly approved draft substitute"
|
|
assert all(term in constraints for term in ("reason", "review", "approval", "original")), \
|
|
"draft substitute must preserve reason, review, approval, and original dates"
|
|
assert re.search(r"unverified.*rejected.*unknown.*(?:any|all).*path", constraints, re.I | re.S), \
|
|
"unverified, rejected, or unknown amounts are never authority on any path"
|
|
assert "future_manual_asset" not in constraints
|
|
|
|
|
|
def test_legacy_metadata_field_lists_are_required_retained_fields_not_exhaustive_response_keys() -> None:
|
|
interface = text("docs/interface-contract.md")
|
|
legacy = interface.split("## D04 snapshot close output contract", 1)[1].split("## settlement-draft-service", 1)[0]
|
|
assert re.search(r"legacy direct close", legacy, re.I)
|
|
assert re.search(r"required top-level metadata fields", legacy, re.I)
|
|
assert re.search(r"not (?:an )?exhaustive", legacy, re.I)
|
|
assert "additive settlement-draft finalize response" in legacy
|
|
|
|
|
|
def test_settlement_draft_project_ownership_and_shared_policy_wiring() -> None:
|
|
doc = text("docs/project-structure.md")
|
|
required_rows = {
|
|
"jarca_ledger_mcp/service.py": "public re-export",
|
|
"jarca_ledger_mcp/adapters/mcp/tool_contracts.py": "inputSchema",
|
|
"jarca_ledger_mcp/adapters/streamlit/settlement_gateway.py": "auth-before-DB",
|
|
"dashboard/page_views/settlement_drafts.py": "render_settlement_drafts",
|
|
"dashboard/page_views/monthly_settlement.py": "page_monthly_settlement",
|
|
"dashboard/page_views/assets.py": "page_assets",
|
|
"jarca_ledger_mcp/application/account_valuation_service.py": "register_account_valuation",
|
|
"jarca_ledger_mcp/infrastructure/sqlite/schema.py": "schema 13",
|
|
}
|
|
for path, responsibility in required_rows.items():
|
|
rows = [row for row in doc.splitlines() if row.startswith("|") and f"`{path}`" in row]
|
|
assert rows and any(responsibility in row for row in rows), (path, responsibility)
|
|
assert (ROOT / path).is_file(), path
|
|
section = doc.split("### Shared settlement-draft policies", 1)[1]
|
|
for clause in ("one active draft per month", "idx_settlement_drafts_active_month", "WHERE status='active'",
|
|
"settlement_draft_service.py", "month-end", "unverified", "authority amount NULL",
|
|
"confirmed child", "explicit reselection", "review", "approval",
|
|
"tests/test_settlement_draft_adapters.py", "tests/test_monthly_settlement_drafts.py"):
|
|
assert clause in section, clause
|
|
|
|
|
|
def test_settlement_draft_runtime_cancel_and_recovery_contract() -> None:
|
|
doc = text("docs/runtime-flow.md")
|
|
assert "### Cancellation and reopening" in doc, "cancel/reopen lifecycle must be documented"
|
|
section = doc.split("### Cancellation and reopening", 1)[1]
|
|
for clause in ("cancel_settlement_draft(draft_id, expected_version, cancelled_by, reason, db_path=None)",
|
|
"active", "nonempty", "cancelled_by", "cancellation_reason", "cancelled_at",
|
|
"approval_fingerprint", "approval_version", "version", "updated_at", "BEGIN IMMEDIATE",
|
|
"retained", "account_valuations", "asset_snapshot_runs", "asset_snapshots",
|
|
"rollback", "no-write", "completed", "cancelled", "fresh create key",
|
|
"same create key", "latest prior run", "correction_of_run_id", "correction_reason",
|
|
"tests/test_settlement_draft_correction_contract.py"):
|
|
assert clause in section, clause
|
|
source = ast.parse(text("jarca_ledger_mcp/application/settlement_draft_service.py"))
|
|
cancel = next(n for n in source.body if isinstance(n, ast.FunctionDef) and n.name == "cancel_settlement_draft")
|
|
assert [arg.arg for arg in cancel.args.args] == ["draft_id", "expected_version", "cancelled_by", "reason", "db_path"]
|
|
|
|
|
|
def _comprehensive_section(doc, heading):
|
|
assert heading in doc, f"missing documented contract section: {heading}"
|
|
return doc.split(heading, 1)[1].split("\n### ", 1)[0]
|
|
|
|
|
|
def _comprehensive_table(section):
|
|
rows = {}
|
|
for line in section.splitlines():
|
|
if not line.startswith("|") or re.match(r"^\|\s*:?-+", line):
|
|
continue
|
|
cells = [c.strip().strip("` ") for c in line.strip().strip("|").split("|")]
|
|
if len(cells) == 4 and cells[0] not in {"Column", "Field", "field"}:
|
|
rows[cells[0]] = cells[1:]
|
|
return rows
|
|
|
|
|
|
def test_settlement_cashflow_model_documents_schema_and_legacy_reproduction(tmp_path):
|
|
import sqlite3
|
|
from jarca_ledger_mcp.db import init_db
|
|
doc = (ROOT / "docs/data-model.md").read_text(encoding="utf-8")
|
|
names = {
|
|
"asset_snapshot_runs": ("settlement_month", "currency", "income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth", "cashflow_policy_json", "aggregation_version", "snapshot_schema_version", "source_snapshot_json"),
|
|
"settlement_drafts": ("cashflow_policy_json", "aggregation_version"),
|
|
"settlement_cashflow_snapshots": ("id", "monthly_settlement_id", "settlement_run_id", "cashflow_kind", "category", "subcategory", "amount", "transaction_count", "currency"),
|
|
}
|
|
database = tmp_path / "synthetic-doc-schema.sqlite3"
|
|
init_db(str(database))
|
|
with sqlite3.connect(database) as conn:
|
|
for table, columns in names.items():
|
|
section = _comprehensive_section(doc, f"### {table} cashflow field contract")
|
|
observed = _comprehensive_table(section)
|
|
ddl = {r[1]: [r[2], "NOT NULL" if r[3] else "NULL", r[4] if r[4] is not None else "none"] for r in conn.execute(f"PRAGMA table_info({table})") if r[1] in columns}
|
|
assert observed == ddl, f"field types/nullability/defaults differ from actual DDL: {table}"
|
|
reproduction = _comprehensive_section(doc, "### Comprehensive snapshot evidence inputs")
|
|
for field in ("source_snapshot_json", "transactions", "assets", "transaction_date", "originalcashflow_kind", "applied_amount", "valuation_id", "valuation_date", "authority_source", "cashflow_policy_json", "legacy", "NULL"):
|
|
assert field in reproduction, field
|
|
for constraint in (
|
|
"ON DELETE RESTRICT", "UNIQUE", "transaction_count > 0",
|
|
"idx_settlement_cashflow_snapshots_run", "exactly one parent",
|
|
"(settlement_run_id,cashflow_kind,category,subcategory,currency)",
|
|
"(monthly_settlement_id,cashflow_kind,category,subcategory,currency)",
|
|
):
|
|
assert constraint in doc, constraint
|
|
|
|
|
|
def test_settlement_cashflow_interface_specifies_valid_policy_review_and_response_examples():
|
|
from jarca_ledger_mcp.application.settlement_cashflow_service import _normalize_cashflow_policy
|
|
doc = (ROOT / "docs/interface-contract.md").read_text(encoding="utf-8")
|
|
policy_section = _comprehensive_section(doc, "### Comprehensive cashflow policy input")
|
|
match = re.search(r"```json\s*(.*?)\s*```", policy_section, re.S)
|
|
assert match, "policy input needs an executable JSON example"
|
|
policy = json.loads(match.group(1))
|
|
assert _normalize_cashflow_policy(policy) == policy
|
|
assert set(policy["savings_categories"][0]) == {"category", "subcategory"}
|
|
assert set(policy["category_decisions"][0]) == {"category", "subcategory", "cashflow_kind", "reason"}
|
|
for field in ("income", "expense", "excluded", "duplicate", "nonempty", "expected_version", "review_fingerprint", "aggregation_version", "comprehensive-v1"):
|
|
assert field in policy_section, field
|
|
outputs = _comprehensive_section(doc, "### Comprehensive settlement response contract")
|
|
totals = _comprehensive_table(outputs)
|
|
keys = ("income_total", "expense_total", "net_cashflow", "total_assets", "total_liabilities", "net_worth")
|
|
assert set(keys) <= set(totals)
|
|
for key in keys:
|
|
assert totals[key] == ["integer", "required", "null"], key
|
|
for field in ("cashflow_items", "cashflow_blockers", "transaction_id", "reason", "validation_error", "cashflow review blockers must be resolved", "final_response", "cashflow_policy", "source_snapshot", "refund"):
|
|
assert field in outputs, field
|
|
examples = [json.loads(s) for s in re.findall(r"```json\s*(.*?)\s*```", outputs, re.S)]
|
|
review = next((x for x in examples if isinstance(x, dict) and "cashflow_blockers" in x and "review_fingerprint" in x), None)
|
|
assert review is not None, "review response example must use actual root fields"
|
|
assert review["success"] is True and all(type(review[key]) is int for key in keys)
|
|
assert review["expense_total"] < 0, "signed-refund example must retain a negative expense total"
|
|
assert all(set(x) == {"transaction_id", "reason"} for x in review["cashflow_blockers"])
|
|
saved = next((x for x in examples if isinstance(x, dict) and x.get("source") == "settlement_snapshot"), None)
|
|
assert saved is not None and saved["success"] is True and "run_id" in saved
|
|
assert all(key in saved for key in keys)
|
|
|
|
|
|
def test_settlement_cashflow_private_service_has_explicit_ownership_and_test_mapping():
|
|
doc = (ROOT / "docs/project-structure.md").read_text(encoding="utf-8")
|
|
rows = [line for line in doc.splitlines() if line.startswith("|") and "settlement_cashflow_service.py" in line]
|
|
assert len(rows) == 1, "private aggregation owner must have an explicit responsibility row"
|
|
for field in ("private", "_normalize_cashflow_policy", "_review_cashflow", "settlement_draft_service.py", "test_settlement_cashflow_policy.py", "test_comprehensive_settlement_adapters.py"):
|
|
assert field in rows[0], field
|
|
|
|
def test_cashflow_policy_documented_signature_matches_public_api():
|
|
from jarca_ledger_mcp import service
|
|
import inspect
|
|
section = _comprehensive_section((ROOT / "docs/interface-contract.md").read_text(), "### Comprehensive cashflow policy input")
|
|
assert list(inspect.signature(service.set_settlement_cashflow_policy).parameters) == ["draft_id", "expected_version", "updated_by", "cashflow_policy", "db_path"]
|
|
assert "set_settlement_cashflow_policy(draft_id, expected_version, updated_by, cashflow_policy, db_path=None)" in section
|
|
assert "set_settlement_draft_cashflow_policy(" not in section
|