feat: implement dcy-007 contract governance and ci gates

tests/test_contract_regression.py

@@ -0,0 +1,128 @@
+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")