113 lines
5.6 KiB
Python
113 lines
5.6 KiB
Python
"""Contract tests for the Issue 15 Streamlit gateway and MCP surface.
|
|
|
|
Fixtures are synthetic and isolated: each test receives an empty temporary SQLite
|
|
path; no production DB or external service is used. These assertions intentionally
|
|
fail until the approved adapter/API is implemented.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import inspect
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
OPERATIONS = {
|
|
"create_settlement_draft": ["month", "created_by", "idempotency_key", "basis_date", "memo", "correction_of_run_id", "correction_reason", "db_path"],
|
|
"list_settlement_drafts": ["month", "db_path"],
|
|
"get_settlement_draft": ["draft_id", "db_path"],
|
|
"update_settlement_draft": ["draft_id", "expected_version", "updated_by", "memo", "db_path"],
|
|
"refresh_settlement_draft_targets": ["draft_id", "expected_version", "updated_by", "db_path"],
|
|
"set_settlement_draft_item": ["draft_id", "source_type", "source_id", "expected_version", "selected_by", "valuation_id", "selection_mode", "selection_reason", "included", "exclusion_reason", "db_path"],
|
|
"review_settlement_draft": ["draft_id", "db_path"],
|
|
"approve_settlement_draft": ["draft_id", "expected_version", "review_fingerprint", "approved_by", "approval_reason", "approval_reference", "allow_missing_exclusions", "allow_carried_before", "allow_manual_substitute", "db_path"],
|
|
"finalize_settlement_draft": ["draft_id", "expected_version", "review_fingerprint", "idempotency_key", "db_path"],
|
|
"cancel_settlement_draft": ["draft_id", "expected_version", "cancelled_by", "reason", "db_path"],
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def db_path(tmp_path: Path) -> Path:
|
|
return tmp_path / "synthetic-settlement.sqlite3"
|
|
|
|
|
|
def test_gateway_exports_exact_contract_signature_and_delegates(monkeypatch, db_path):
|
|
module = importlib.import_module("jarca_ledger_mcp.adapters.streamlit.settlement_gateway")
|
|
service = importlib.import_module("jarca_ledger_mcp.application.settlement_draft_service")
|
|
monkeypatch.setenv("JARCA_LEDGER_DB_PATH", str(db_path))
|
|
calls = []
|
|
for name, expected_parameters in OPERATIONS.items():
|
|
gateway_fn = getattr(module, name)
|
|
service_fn = getattr(service, name)
|
|
assert list(inspect.signature(gateway_fn).parameters) == expected_parameters
|
|
assert inspect.signature(gateway_fn) == inspect.signature(service_fn)
|
|
marker = {"success": True, "operation": name}
|
|
def stub(*args, _marker=marker, _name=name, **kwargs):
|
|
calls.append((_name, kwargs.get("db_path")))
|
|
return _marker
|
|
monkeypatch.setattr(service, name, stub)
|
|
kwargs = {p: "synthetic" for p in expected_parameters}
|
|
kwargs.update(db_path=str(db_path))
|
|
kwargs["correction_of_run_id"] = ""
|
|
kwargs["correction_reason"] = ""
|
|
if "expected_version" in kwargs:
|
|
kwargs["expected_version"] = 1
|
|
if "included" in kwargs:
|
|
kwargs["included"] = True
|
|
if name == "create_settlement_draft":
|
|
kwargs.update(month="2026-09", basis_date="2026-09-30")
|
|
assert gateway_fn(**kwargs) == marker
|
|
assert calls[-1] == (name, str(db_path))
|
|
|
|
|
|
def test_gateway_exposes_existing_valuation_evidence_operations():
|
|
gateway = importlib.import_module("jarca_ledger_mcp.adapters.streamlit.settlement_gateway")
|
|
for name in ("register_account_valuation", "list_account_valuations", "correct_account_valuation"):
|
|
assert callable(getattr(gateway, name)), name
|
|
|
|
|
|
def test_mcp_contract_and_fastmcp_tool_calls_preserve_service_result(monkeypatch, db_path):
|
|
contracts = importlib.import_module("jarca_ledger_mcp.adapters.mcp.tool_contracts")
|
|
server = importlib.import_module("jarca_ledger_mcp.adapters.mcp.server")
|
|
monkeypatch.setenv("JARCA_LEDGER_DB_PATH", str(db_path))
|
|
calls = []
|
|
tools = getattr(server, "mcp", getattr(server, "server", None))._tool_manager._tools
|
|
for name in OPERATIONS:
|
|
assert name in contracts.TOOL_CONTRACTS
|
|
assert name in tools
|
|
marker = {"success": True, "draft": {"id": "fixture-draft", "version": 1}}
|
|
def stub(*args, _marker=marker, _name=name, **kwargs):
|
|
calls.append(_name)
|
|
return _marker
|
|
monkeypatch.setattr(contracts, name, stub)
|
|
tool = tools[name]
|
|
call = getattr(tool, "fn", getattr(tool, "function", None))
|
|
assert call is not None
|
|
kwargs = {p: "synthetic" for p in OPERATIONS[name]}
|
|
kwargs.pop("db_path", None)
|
|
if "expected_version" in kwargs: kwargs["expected_version"] = 1
|
|
if "included" in kwargs: kwargs["included"] = True
|
|
if name == "create_settlement_draft": kwargs.update(month="2026-09", basis_date="2026-09-30")
|
|
assert call(**kwargs) == marker
|
|
assert calls[-1] == name
|
|
|
|
|
|
def test_read_only_gateway_calls_do_not_create_missing_database(monkeypatch, db_path):
|
|
gateway = importlib.import_module("jarca_ledger_mcp.adapters.streamlit.settlement_gateway")
|
|
# Real delegation is required here: read paths must not create a missing file.
|
|
for name in ("list_settlement_drafts",):
|
|
getattr(gateway, name)(db_path=str(db_path))
|
|
assert not db_path.exists()
|
|
|
|
|
|
|
|
def test_dashboard_authentication_entrypoint_precedes_database_resolution():
|
|
app_source = (Path(__file__).parents[1] / "dashboard" / "app.py").read_text(encoding="utf-8")
|
|
assert app_source.index("require_dashboard_authentication()") < app_source.index("get_db_path()")
|
|
|
|
|
|
def test_service_reexports_every_lifecycle_operation():
|
|
service = importlib.import_module("jarca_ledger_mcp.application.settlement_draft_service")
|
|
assert all(callable(getattr(service, name, None)) for name in OPERATIONS)
|