64 lines
4.0 KiB
Python
64 lines
4.0 KiB
Python
"""Executable reproductions reconcile static audit hypotheses with real behavior."""
|
|
import sqlite3
|
|
|
|
from jarca_ledger_mcp import service as s
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def ok(result):
|
|
assert result['success'], result
|
|
return result
|
|
|
|
|
|
def test_empty_draft_persists_without_closing_and_has_consistent_public_shape(tmp_path):
|
|
path = str(tmp_path / 'empty.sqlite3')
|
|
init_db(path)
|
|
result = ok(s.create_settlement_draft('2026-09', 'creator', 'initial', db_path=path))
|
|
draft = result['draft']
|
|
assert draft['status'] == 'active' and draft['version'] == 1
|
|
assert draft['basis_date'] == '2026-09-30' and draft['correction_of_run_id'] == ''
|
|
assert result['items'] == [] and not result['review']['complete']
|
|
assert result['review']['totals'] == {'total_assets': 0, 'total_liabilities': 0, 'net_worth': 0}
|
|
assert ok(s.get_settlement_draft(draft['id'], db_path=path)) == result
|
|
with sqlite3.connect(path) as conn:
|
|
assert conn.execute('SELECT COUNT(*) FROM settlement_drafts').fetchone()[0] == 1
|
|
assert conn.execute('SELECT COUNT(*) FROM asset_snapshot_runs').fetchone()[0] == 0
|
|
assert conn.execute('SELECT COUNT(*) FROM asset_snapshots').fetchone()[0] == 0
|
|
|
|
|
|
def test_initial_draft_freezes_exact_verified_selector_compatibly(tmp_path):
|
|
path = str(tmp_path / 'selection.sqlite3')
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account('Synthetic cash', db_path=path))['account']['id']
|
|
verified = ok(s.register_account_valuation(account, '2026-09-30', 100, verification_status='verified', idempotency_key='verified', db_path=path))['valuation']['id']
|
|
unverified = ok(s.register_account_valuation(account, '2026-09-30', 999, verification_status='unverified', idempotency_key='unverified', db_path=path))['valuation']['id']
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE account_valuations SET recorded_at='2026-10-01T00:00:00Z' WHERE id=?", (verified,))
|
|
conn.execute("UPDATE account_valuations SET recorded_at='2026-10-02T00:00:00Z' WHERE id=?", (unverified,))
|
|
selected = ok(s.select_account_valuation(account, '2026-09-30', as_of=False, db_path=path))
|
|
assert selected['valuation']['id'] == verified
|
|
draft = ok(s.create_settlement_draft('2026-09', 'creator', 'initial', db_path=path))
|
|
item = next(x for x in draft['items'] if x['source_id'] == account)
|
|
assert item['valuation_id'] == verified and item['amount'] == 100 and item['verification_status'] == 'verified'
|
|
|
|
|
|
def test_refresh_manual_target_preserves_retired_history_and_allows_explicit_refreeze(tmp_path):
|
|
path = str(tmp_path / 'retired.sqlite3')
|
|
init_db(path)
|
|
asset = ok(s.create_manual_asset('Synthetic property', 100, '2026-09-30', 'real_estate', db_path=path))['manual_asset']['id']
|
|
created = ok(s.create_settlement_draft('2026-09', 'creator', 'initial', db_path=path))
|
|
draft, original = created['draft'], created['items'][0]
|
|
ok(s.deactivate_manual_asset(asset, db_path=path))
|
|
retired = ok(s.refresh_settlement_draft_targets(draft['id'], draft['version'], 'refresher', db_path=path))
|
|
row = next(x for x in retired['items'] if x['source_id'] == asset)
|
|
assert row['id'] == original['id'] and not row['included'] and row['exclusion_reason']
|
|
assert row['state_hash'] == original['state_hash'] and row['candidate_amount'] == 100
|
|
refrozen = ok(s.set_settlement_draft_item(draft['id'], 'manual_asset', asset, retired['draft']['version'], 'refreezer', included=False, exclusion_reason='Synthetic retired source confirmed', db_path=path))
|
|
row = refrozen['items'][0]
|
|
assert row['id'] == original['id'] and not row['included'] and row['amount'] is None
|
|
assert row['state_hash'] != original['state_hash'] and not row['review_needed']
|
|
assert row['selected_by'] == 'refreezer'
|
|
invalid = s.set_settlement_draft_item(draft['id'], 'manual_asset', asset, refrozen['draft']['version'], 'selector', included=True, db_path=path)
|
|
assert not invalid['success']
|
|
assert ok(s.get_settlement_draft(draft['id'], db_path=path))['draft']['version'] == refrozen['draft']['version']
|