167 lines
6.7 KiB
Python
167 lines
6.7 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import inspect
|
|
import sqlite3
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp.application import account_valuation_service as valuation
|
|
from jarca_ledger_mcp.application import (
|
|
assets_service,
|
|
categorization_services,
|
|
cross_transfer_service,
|
|
reconciliation_service,
|
|
system_services,
|
|
)
|
|
from jarca_ledger_mcp.infrastructure.sqlite import connection
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def test_read_queries_missing_nested_db_return_stable_error_without_creating_path(tmp_path):
|
|
db_path = tmp_path / "missing" / "nested" / "ledger.sqlite3"
|
|
calls = [
|
|
lambda: system_services.list_ledger_accounts(db_path=str(db_path)),
|
|
lambda: valuation.list_account_valuations("account", db_path=str(db_path)),
|
|
lambda: assets_service.list_manual_assets(db_path=str(db_path)),
|
|
lambda: assets_service.preview_asset_snapshot_close(db_path=str(db_path)),
|
|
lambda: reconciliation_service.get_balance_reconciliation_report(db_path=str(db_path)),
|
|
]
|
|
|
|
for read in calls:
|
|
result = read()
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "db_not_initialized"
|
|
assert result["error"]
|
|
assert not db_path.exists()
|
|
assert not db_path.parent.exists()
|
|
|
|
|
|
def test_read_only_connection_rejects_writes_and_does_not_commit(tmp_path):
|
|
db_path = tmp_path / "ledger.sqlite3"
|
|
system_services.create_ledger_account("checking", db_path=str(db_path)) # first-use writer initializes
|
|
before = _sha256(db_path)
|
|
|
|
with connection.read_db(str(db_path)) as conn:
|
|
with pytest.raises(sqlite3.OperationalError):
|
|
conn.execute("INSERT INTO accounts (id, name, type) VALUES ('x', 'x', 'bank')")
|
|
|
|
assert _sha256(db_path) == before
|
|
|
|
|
|
def test_read_suite_does_not_initialize_or_change_initialized_database(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "ledger.sqlite3"
|
|
system_services.create_ledger_account("checking", db_path=str(db_path))
|
|
before = _sha256(db_path)
|
|
def fail_init(*args, **kwargs):
|
|
raise AssertionError("read path called init_db")
|
|
|
|
monkeypatch.setattr("jarca_ledger_mcp.infrastructure.sqlite.migrations.init_db", fail_init)
|
|
monkeypatch.setattr("jarca_ledger_mcp.infrastructure.sqlite.connection.init_db", fail_init, raising=False)
|
|
reads = [
|
|
lambda: system_services.list_ledger_accounts(db_path=str(db_path)),
|
|
lambda: valuation.list_account_valuations("account", db_path=str(db_path)),
|
|
lambda: assets_service.list_manual_assets(db_path=str(db_path)),
|
|
lambda: assets_service.preview_asset_snapshot_close(db_path=str(db_path)),
|
|
lambda: reconciliation_service.get_balance_reconciliation_report(db_path=str(db_path)),
|
|
]
|
|
for read in reads:
|
|
assert read()["success"] is True
|
|
assert _sha256(db_path) == before
|
|
|
|
|
|
def test_first_use_writer_still_initializes_database(tmp_path):
|
|
db_path = tmp_path / "new" / "ledger.sqlite3"
|
|
result = system_services.create_ledger_account("checking", db_path=str(db_path))
|
|
assert result["success"] is True
|
|
assert db_path.is_file()
|
|
|
|
|
|
def test_read_schema_error_is_not_misreported_as_uninitialized(tmp_path):
|
|
db_path = tmp_path / "empty.sqlite3"
|
|
sqlite3.connect(db_path).close()
|
|
|
|
result = system_services.list_ledger_accounts(db_path=str(db_path))
|
|
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "db_error"
|
|
|
|
|
|
def test_nested_categorization_preview_propagates_missing_db_envelope(tmp_path):
|
|
db_path = tmp_path / "missing" / "ledger.sqlite3"
|
|
|
|
result = categorization_services.preview_categorization(db_path=str(db_path))
|
|
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "db_not_initialized"
|
|
assert not db_path.parent.exists()
|
|
|
|
|
|
def test_public_read_signatures_remain_positional_compatible():
|
|
assert list(inspect.signature(valuation.list_account_valuations).parameters)[:2] == ["account_id", "include_rejected"]
|
|
assert list(inspect.signature(assets_service.preview_asset_snapshot_close).parameters)[:2] == ["basis_date", "db_path"]
|
|
assert list(inspect.signature(reconciliation_service.get_balance_reconciliation_report).parameters)[:2] == ["account_id", "baseline_balance"]
|
|
|
|
|
|
def test_preview_categorization_initialized_db_preserves_success_envelope(tmp_path):
|
|
db_path = tmp_path / "ledger.sqlite3"
|
|
system_services.create_ledger_account("checking", db_path=str(db_path))
|
|
result = categorization_services.preview_categorization(db_path=str(db_path))
|
|
assert result["success"] is True
|
|
assert result["matched_count"] == 0
|
|
assert result["unmatched_count"] == 0
|
|
|
|
|
|
def test_apply_categorization_rules_missing_db_initializes_writer_or_returns_stable_error(tmp_path):
|
|
db_path = tmp_path / "nested" / "ledger.sqlite3"
|
|
result = categorization_services.apply_categorization_rules(db_path=str(db_path))
|
|
assert result["success"] is True
|
|
assert result["preview"]["matched_count"] == 0
|
|
|
|
|
|
def test_cross_transfer_review_queue_missing_nested_db_returns_stable_error_without_creating_path(tmp_path):
|
|
db_path = tmp_path / "missing" / "nested" / "ledger.sqlite3"
|
|
|
|
result = cross_transfer_service.list_cross_transfer_review_queue(db_path=str(db_path))
|
|
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "db_not_initialized"
|
|
assert result["error"]
|
|
assert not db_path.exists()
|
|
assert not db_path.parent.exists()
|
|
|
|
|
|
def test_cross_transfer_preview_missing_nested_db_returns_stable_error_without_creating_path(tmp_path):
|
|
db_path = tmp_path / "missing" / "nested" / "ledger.sqlite3"
|
|
|
|
result = cross_transfer_service.preview_cross_account_transfers(db_path=str(db_path))
|
|
|
|
assert result["success"] is False
|
|
assert result["error_code"] == "db_not_initialized"
|
|
assert result["error"]
|
|
assert not db_path.exists()
|
|
assert not db_path.parent.exists()
|
|
|
|
|
|
def test_cross_transfer_read_paths_do_not_initialize_or_mutate_initialized_db(tmp_path, monkeypatch):
|
|
db_path = tmp_path / "ledger.sqlite3"
|
|
system_services.create_ledger_account("checking", db_path=str(db_path))
|
|
before = _sha256(db_path)
|
|
|
|
def fail_init(*args, **kwargs):
|
|
raise AssertionError("cross-transfer read path called init_db")
|
|
|
|
monkeypatch.setattr("jarca_ledger_mcp.infrastructure.sqlite.migrations.init_db", fail_init)
|
|
monkeypatch.setattr("jarca_ledger_mcp.infrastructure.sqlite.connection.init_db", fail_init, raising=False)
|
|
|
|
queue = cross_transfer_service.list_cross_transfer_review_queue(db_path=str(db_path))
|
|
preview = cross_transfer_service.preview_cross_account_transfers(db_path=str(db_path))
|
|
|
|
assert queue["success"] is True
|
|
assert preview["success"] is True
|
|
assert _sha256(db_path) == before
|