130 lines
6.6 KiB
Python
130 lines
6.6 KiB
Python
"""Strict selection, conditional authority and public close regressions."""
|
|
import pytest
|
|
|
|
from jarca_ledger_mcp import service as s
|
|
from jarca_ledger_mcp.db import init_db
|
|
|
|
|
|
def ok(r):
|
|
assert r['success'], r
|
|
return r
|
|
|
|
|
|
def setup(tmp_path, currency='KRW'):
|
|
path = str(tmp_path / 'scope.sqlite3')
|
|
init_db(path)
|
|
account = ok(s.create_ledger_account('Synthetic investment', type='investment', currency=currency, liquidity_bucket='investment', db_path=path))['account']['id']
|
|
return path, account
|
|
|
|
|
|
def valuation(path, account, key='v', **kwargs):
|
|
values = dict(basis_date='2026-09-30', amount=100, valuation_kind='total_value', verification_status='verified', idempotency_key=key, db_path=path)
|
|
values.update(kwargs)
|
|
return ok(s.register_account_valuation(account, **values))['valuation']['id']
|
|
|
|
|
|
def draft(path):
|
|
return ok(s.create_settlement_draft('2026-09', 'creator', 'create', db_path=path))['draft']
|
|
|
|
|
|
def select(path, d, account, value, **kwargs):
|
|
return s.set_settlement_draft_item(d['id'], 'account', account, d['version'], 'selector', valuation_id=value, db_path=path, **kwargs)
|
|
|
|
|
|
def test_automatic_selection_preserves_representative_priority_and_verified_status(tmp_path):
|
|
path, account = setup(tmp_path)
|
|
representative = valuation(path, account)
|
|
valuation(path, account, 'cash', amount=30, valuation_kind='cash_balance')
|
|
valuation(path, account, 'candidate', amount=900, verification_status='unverified')
|
|
d = draft(path)
|
|
review = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
row = next(x for x in review['items'] if x['source_id'] == account)
|
|
assert row['valuation_id'] == representative
|
|
assert review['totals']['net_worth'] == 100
|
|
|
|
|
|
def test_automatic_ambiguous_representative_remains_missing_until_explicit_choice(tmp_path):
|
|
path, account = setup(tmp_path)
|
|
chosen = valuation(path, account)
|
|
valuation(path, account, 'other', amount=200)
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
assert r['missing_items'] and r['totals']['net_worth'] == 0
|
|
selected = ok(select(path, d, account, chosen))['draft']
|
|
assert ok(s.review_settlement_draft(selected['id'], db_path=path))['totals']['net_worth'] == 100
|
|
|
|
|
|
@pytest.mark.parametrize('status,mode', [('unverified','exact'), ('unverified','carried_before'), ('rejected','manual_substitute')])
|
|
def test_invalid_verification_never_selected_as_confirmed_authority(tmp_path, status, mode):
|
|
path, account = setup(tmp_path)
|
|
value = valuation(path, account, verification_status=status, basis_date='2026-09-29' if mode == 'carried_before' else '2026-09-30')
|
|
d = draft(path)
|
|
result = select(path, d, account, value, selection_mode=mode, selection_reason='synthetic reason')
|
|
assert not result['success'], result
|
|
assert ok(s.get_settlement_draft(d['id'], db_path=path))['draft']['version'] == d['version']
|
|
|
|
|
|
def test_non_krw_candidate_is_missing_null_authority_not_converted(tmp_path):
|
|
path, account = setup(tmp_path, currency='USD')
|
|
value = valuation(path, account, currency='USD')
|
|
d = draft(path)
|
|
ok(select(path, d, account, value))
|
|
r = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
assert r['missing_items'] and not r['included_items']
|
|
assert r['items'][0]['amount'] is None
|
|
assert r['items'][0]['candidate_amount'] == 100
|
|
assert r['totals']['net_worth'] == 0 and r['warnings']
|
|
|
|
|
|
def test_manual_loan_is_representative_liability_and_future_manual_is_not_substitute(tmp_path):
|
|
path = str(tmp_path / 'manual-loan.sqlite3')
|
|
init_db(path)
|
|
ok(s.create_manual_asset('Synthetic liability', 40, '2026-09-30', 'loan', db_path=path))
|
|
future = ok(s.create_manual_asset('Future property', 100, '2026-10-05', 'real_estate', db_path=path))['manual_asset']['id']
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
assert r['totals']['total_assets'] == 0 and r['totals']['total_liabilities'] == 40
|
|
result = s.set_settlement_draft_item(d['id'], 'manual', future, d['version'], 'selector', selection_mode='manual_substitute', selection_reason='not a valuation id', db_path=path)
|
|
assert not result['success'], result
|
|
|
|
|
|
@pytest.mark.parametrize('bad', [None, True, 1, [], {}])
|
|
def test_invalid_creator_is_rejected_before_database_creation(tmp_path, bad):
|
|
path = tmp_path / 'absent' / 'db.sqlite3'
|
|
result = s.create_settlement_draft('2026-09', bad, 'key', db_path=str(path))
|
|
assert not result['success'] and result['error_code'] == 'validation_error', result
|
|
assert not path.parent.exists()
|
|
|
|
|
|
@pytest.mark.parametrize('bad', ['yes', 1, None])
|
|
def test_approval_flags_require_actual_boolean_and_preserve_state(tmp_path, bad):
|
|
path, account = setup(tmp_path)
|
|
valuation(path, account)
|
|
d = draft(path)
|
|
r = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
result = s.approve_settlement_draft(d['id'], d['version'], r['review_fingerprint'], 'reviewer', 'synthetic reason', 'ref', allow_missing_exclusions=bad, db_path=path)
|
|
assert not result['success'] and result['error_code'] == 'validation_error', result
|
|
assert not ok(s.get_settlement_draft(d['id'], db_path=path))['draft']['approved_by']
|
|
|
|
|
|
def test_finalized_substitute_retains_dates_flags_and_public_consumer_evidence(tmp_path):
|
|
path, account = setup(tmp_path)
|
|
v = valuation(path, account, basis_date='2026-10-05')
|
|
d = draft(path)
|
|
d = ok(select(path, d, account, v, selection_mode='manual_substitute', selection_reason='synthetic after-date'))['draft']
|
|
r = ok(s.review_settlement_draft(d['id'], db_path=path))
|
|
ok(s.approve_settlement_draft(d['id'], d['version'], r['review_fingerprint'], 'reviewer', 'substitution accepted', 'ref', allow_manual_substitute=True, db_path=path))
|
|
final = ok(s.finalize_settlement_draft(d['id'], d['version'], r['review_fingerprint'], 'final', db_path=path))
|
|
assert final['contains_substitutions'] is True
|
|
assert final['has_approved_missing'] is False
|
|
assert final['contains_carried'] is False
|
|
assert final['valuation_date'] and 'age_days' in final and 'verification_status' in final and 'evidence_kind' in final
|
|
report = ok(s.get_asset_snapshot_report('2026-09-30', db_path=path))
|
|
assert report['net_worth'] == 100
|
|
assert report['close_metadata']['contains_substitutions'] is True
|
|
evidence = report['close_metadata']['applied_evidence'][0]
|
|
assert evidence['valuation_date'] == '2026-10-05'
|
|
assert evidence['date_difference_days'] == 5 and evidence['selection_mode'] == 'manual_substitute'
|
|
assert evidence['selection_reason'] == 'synthetic after-date'
|
|
assert s.finalize_settlement_draft(d['id'], d['version'], r['review_fingerprint'], 'final', db_path=path) == final
|