179 lines
11 KiB
Python
179 lines
11 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.1"
|
|
assert metadata["issue"] == "#15"
|
|
assert metadata["implementation_pr"] == "https://gitea.jsfamily.quest/jsfamily/jarca-ledger-mcp/pulls/16"
|
|
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_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) == 17
|
|
assert [n for n, _ in cases] == [f"{i:02}" for i in range(1, 18)]
|
|
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"]
|