182 lines
10 KiB
Python
182 lines
10 KiB
Python
"""Synthetic regressions for frozen evidence and reviewed authority."""
|
|
import sqlite3
|
|
|
|
import pytest
|
|
|
|
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
|
|
|
|
|
|
@pytest.fixture
|
|
def case(tmp_path):
|
|
path = str(tmp_path / 'authority.sqlite3')
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account('Synthetic investment', type='investment', liquidity_bucket='investment', db_path=path))['account']['id']
|
|
val = ok(s.register_account_valuation(account, '2026-09-30', 500, valuation_kind='total_value', verification_status='verified', idempotency_key='evidence', db_path=path))['valuation']['id']
|
|
draft = ok(s.create_settlement_draft('2026-09', 'creator', 'create', db_path=path))['draft']
|
|
return path, account, val, draft
|
|
|
|
|
|
def approve(path, draft, **flags):
|
|
review = ok(s.review_settlement_draft(draft['id'], db_path=path))
|
|
ok(s.approve_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'reviewer', 'synthetic reviewed', 'ref', db_path=path, **flags))
|
|
return review
|
|
|
|
|
|
def counts(path):
|
|
with sqlite3.connect(path) as conn:
|
|
return tuple(conn.execute('SELECT COUNT(*) FROM ' + t).fetchone()[0] for t in ('asset_snapshot_runs', 'asset_snapshots'))
|
|
|
|
|
|
@pytest.mark.parametrize('sql', [
|
|
"UPDATE account_valuations SET amount=600 WHERE id=?",
|
|
"UPDATE account_valuations SET memo='changed' WHERE id=?",
|
|
"UPDATE account_valuations SET principal=123 WHERE id=?",
|
|
])
|
|
def test_full_valuation_drift_changes_fingerprint_and_blocks_write(case, sql):
|
|
path, account, val, draft = case
|
|
review = approve(path, draft)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute(sql, (val,))
|
|
changed = ok(s.review_settlement_draft(draft['id'], db_path=path))
|
|
assert changed['review_fingerprint'] != review['review_fingerprint']
|
|
assert any(x.get('review_needed') for x in changed['items'])
|
|
result = s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'final', db_path=path)
|
|
assert not result['success'], result
|
|
assert counts(path) == (0, 0)
|
|
|
|
|
|
def test_classification_drift_requires_explicit_reselection(case):
|
|
path, account, val, draft = case
|
|
review = approve(path, draft)
|
|
with sqlite3.connect(path) as conn:
|
|
conn.execute("UPDATE accounts SET liquidity_bucket='cash' WHERE id=?", (account,))
|
|
changed = ok(s.review_settlement_draft(draft['id'], db_path=path))
|
|
assert changed['review_fingerprint'] != review['review_fingerprint']
|
|
assert any(x.get('review_needed') for x in changed['items'])
|
|
assert not s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'final', db_path=path)['success']
|
|
assert counts(path) == (0, 0)
|
|
|
|
|
|
def test_corrected_parent_cannot_finalize_without_explicit_child_decision(case):
|
|
path, account, val, draft = case
|
|
previous = approve(path, draft)
|
|
child = ok(s.correct_account_valuation(val, amount=550, verification_status='verified', correction_reason='synthetic correction', idempotency_key='child', db_path=path))['valuation']['id']
|
|
review = ok(s.review_settlement_draft(draft['id'], db_path=path))
|
|
assert review['review_fingerprint'] != previous['review_fingerprint']
|
|
result = s.approve_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'reviewer', 'review again', 'ref2', db_path=path)
|
|
if result['success']:
|
|
assert not s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'final', db_path=path)['success']
|
|
assert counts(path) == (0, 0)
|
|
updated = ok(s.set_settlement_draft_item(draft['id'], 'account', account, draft['version'], 'selector', valuation_id=child, db_path=path))['draft']
|
|
fresh = approve(path, updated)
|
|
final = ok(s.finalize_settlement_draft(updated['id'], updated['version'], fresh['review_fingerprint'], 'final2', db_path=path))
|
|
assert final['totals']['net_worth'] == 550
|
|
|
|
|
|
def test_create_replay_uses_original_payload_after_memo_edit(case):
|
|
path, account, val, draft = case
|
|
ok(s.update_settlement_draft(draft['id'], draft['version'], 'editor', 'edited', db_path=path))
|
|
replay = ok(s.create_settlement_draft('2026-09', 'creator', 'create', db_path=path))
|
|
assert replay['draft']['id'] == draft['id']
|
|
assert not s.create_settlement_draft('2026-09', 'creator', 'create', memo='different original request', db_path=path)['success']
|
|
|
|
|
|
def test_empty_draft_cannot_create_empty_close(tmp_path):
|
|
path = str(tmp_path / 'empty.sqlite3')
|
|
init_db(path)
|
|
draft = ok(s.create_settlement_draft('2026-09', 'creator', 'empty', db_path=path))['draft']
|
|
review = approve(path, draft, allow_missing_exclusions=True)
|
|
assert review['complete'] is False
|
|
assert not s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'empty-final', db_path=path)['success']
|
|
assert counts(path) == (0, 0)
|
|
|
|
|
|
def test_manual_reselection_explicitly_refreezes_semantic_evidence(tmp_path):
|
|
path = str(tmp_path / 'manual.sqlite3')
|
|
init_db(path)
|
|
asset = ok(s.create_manual_asset('Synthetic home', 100, '2026-09-30', 'real_estate', db_path=path))['manual_asset']['id']
|
|
draft = ok(s.create_settlement_draft('2026-09', 'creator', 'manual', db_path=path))['draft']
|
|
old = approve(path, draft)
|
|
ok(s.update_manual_asset(asset, valuation_amount=110, db_path=path))
|
|
drift = ok(s.review_settlement_draft(draft['id'], db_path=path))
|
|
assert drift['review_fingerprint'] != old['review_fingerprint']
|
|
assert not s.finalize_settlement_draft(draft['id'], draft['version'], old['review_fingerprint'], 'stale', db_path=path)['success']
|
|
refreshed = ok(s.set_settlement_draft_item(draft['id'], 'manual', asset, draft['version'], 'selector', db_path=path))['draft']
|
|
new = approve(path, refreshed)
|
|
assert not any(x.get('review_needed') for x in new['items'])
|
|
final = ok(s.finalize_settlement_draft(refreshed['id'], refreshed['version'], new['review_fingerprint'], 'manual-final', db_path=path))
|
|
assert final['totals']['net_worth'] == 110
|
|
|
|
|
|
@pytest.mark.parametrize('operation,args', [
|
|
('list_settlement_drafts', ()), ('get_settlement_draft', ('missing',)), ('review_settlement_draft', ('missing',)),
|
|
])
|
|
def test_draft_reads_preserve_missing_db_error_and_no_create(tmp_path, operation, args):
|
|
path = tmp_path / 'absent-parent' / 'db.sqlite3'
|
|
result = getattr(s, operation)(*args, db_path=str(path))
|
|
assert result['error_code'] == 'db_not_initialized', result
|
|
assert not path.parent.exists()
|
|
|
|
|
|
def test_target_refresh_refreezes_manual_retains_account_selection_and_retires_targets(tmp_path):
|
|
path = str(tmp_path / 'refresh.sqlite3')
|
|
init_db(path)
|
|
manual = ok(s.create_manual_asset('Synthetic property', 100, '2026-09-30', 'real_estate', db_path=path))['manual_asset']['id']
|
|
account = ok(s.create_ledger_account('Synthetic investment', type='investment', liquidity_bucket='investment', db_path=path))['account']['id']
|
|
val = ok(s.register_account_valuation(account, '2026-09-30', 50, valuation_kind='total_value', verification_status='verified', idempotency_key='initial', db_path=path))['valuation']['id']
|
|
d = ok(s.create_settlement_draft('2026-09', 'creator', 'create', db_path=path))['draft']
|
|
ok(s.update_manual_asset(manual, valuation_amount=110, db_path=path))
|
|
child = ok(s.correct_account_valuation(val, amount=60, verification_status='verified', correction_reason='synthetic', idempotency_key='child', db_path=path))['valuation']['id']
|
|
added = ok(s.create_manual_asset('Synthetic new', 20, '2026-09-30', 'other', db_path=path))['manual_asset']['id']
|
|
result = ok(s.refresh_settlement_draft_targets(d['id'], d['version'], 'refresh actor', db_path=path))
|
|
review = result['review']
|
|
property_row = next(x for x in review['items'] if x['source_id'] == manual)
|
|
assert property_row['amount'] == 110 and not property_row['review_needed']
|
|
account_row = next(x for x in review['items'] if x['source_id'] == account)
|
|
assert account_row['valuation_id'] == val and account_row['valuation_id'] != child
|
|
assert account_row['review_needed']
|
|
assert any(x['source_id'] == added and x['amount'] == 20 for x in review['items'])
|
|
ok(s.deactivate_manual_asset(added, db_path=path))
|
|
refreshed = ok(s.refresh_settlement_draft_targets(d['id'], result['draft']['version'], 'refresh actor', db_path=path))
|
|
retired = next(x for x in refreshed['items'] if x['source_id'] == added)
|
|
assert retired['included'] is False and retired['exclusion_reason']
|
|
assert retired['selected_by'] == 'refresh actor'
|
|
|
|
|
|
@pytest.mark.parametrize('action', ['memo', 'cancel', 'refresh', 'selection'])
|
|
def test_any_draft_mutation_clears_all_approval_version_state(case, action):
|
|
path, account, val, draft = case
|
|
approve(path, draft)
|
|
before = ok(s.get_settlement_draft(draft['id'], db_path=path))['draft']
|
|
assert before['approval_version'] == draft['version'] and before['approved_at']
|
|
if action == 'memo':
|
|
result = s.update_settlement_draft(draft['id'], draft['version'], 'editor', 'edited', db_path=path)
|
|
elif action == 'cancel':
|
|
result = s.cancel_settlement_draft(draft['id'], draft['version'], 'editor', 'synthetic cancellation', db_path=path)
|
|
elif action == 'refresh':
|
|
result = s.refresh_settlement_draft_targets(draft['id'], draft['version'], 'editor', db_path=path)
|
|
else:
|
|
result = s.set_settlement_draft_item(draft['id'], 'account', account, draft['version'], 'editor', valuation_id=val, db_path=path)
|
|
changed = ok(result)['draft']
|
|
assert changed['version'] == draft['version'] + 1
|
|
assert changed['approval_version'] == 0 and changed['approved_at'] == ''
|
|
assert all(changed[key] == '' for key in ('approved_by', 'approval_fingerprint', 'approval_reason', 'approval_reference'))
|
|
assert changed['approval_flags'] == '{}'
|
|
|
|
|
|
def test_finalize_same_key_different_payload_conflicts_without_partial_write(case):
|
|
path, account, val, draft = case
|
|
review = approve(path, draft)
|
|
final = ok(s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'final-key', db_path=path))
|
|
before = counts(path)
|
|
assert s.finalize_settlement_draft(draft['id'], draft['version'], review['review_fingerprint'], 'final-key', db_path=path) == final
|
|
conflict = s.finalize_settlement_draft(draft['id'], draft['version'] + 1, review['review_fingerprint'], 'final-key', db_path=path)
|
|
assert not conflict['success'] and conflict['error_code'] == 'conflict'
|
|
assert counts(path) == before
|