218 lines
8.6 KiB
Python
218 lines
8.6 KiB
Python
"""Uji verifikasi distribusi profit-share (oracle §V19..V27).
|
|
|
|
`./venv/bin/python test_distribute.py` harus exit 0. Memakai DB sementara +
|
|
mock jaringan/pyntelope — ⊥ transfer nyata. Verifikasi: matematika porsi
|
|
(V20), memo per run (V23), urutan pending→sent/failed (V21), guard
|
|
anti-duplikat (V22), no-op (V25), dry-run & ⊥ key (V26).
|
|
"""
|
|
|
|
import io
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from contextlib import redirect_stdout
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
|
|
import db
|
|
import distribute as dist
|
|
|
|
|
|
def check(tag, cond):
|
|
if not cond:
|
|
raise AssertionError(f'GAGAL: {tag}')
|
|
print(f'ok: {tag}')
|
|
|
|
|
|
class FakeSigned:
|
|
"""Objek SignedTransaction tiruan: id deterministik + send terkontrol."""
|
|
|
|
def __init__(self, txid, fails=0):
|
|
self._txid = txid
|
|
self._fails = fails
|
|
|
|
def id(self):
|
|
return self._txid
|
|
|
|
def send(self):
|
|
if self._fails > 0:
|
|
self._fails -= 1
|
|
raise RuntimeError('broadcast timeout (mock)')
|
|
return {'processed': True}
|
|
|
|
|
|
def make_voters_db(path, rows):
|
|
conn = sqlite3.connect(path)
|
|
conn.execute('DROP TABLE IF EXISTS voters')
|
|
conn.execute('CREATE TABLE voters ('
|
|
'owner TEXT PRIMARY KEY, weight TEXT NOT NULL, '
|
|
'staked REAL NOT NULL, scanned_at TEXT NOT NULL, '
|
|
'last_vote TEXT)')
|
|
conn.executemany(
|
|
'INSERT INTO voters (owner, weight, staked, scanned_at, last_vote) '
|
|
'VALUES (?, ?, ?, ?, ?)',
|
|
rows,
|
|
)
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
|
|
def fake_post(data):
|
|
return type('R', (), {'raise_for_status': lambda self: None,
|
|
'json': lambda self: data})()
|
|
|
|
|
|
def main():
|
|
tmp = tempfile.mkdtemp()
|
|
db.DB_PATH = os.path.join(tmp, 'test.db')
|
|
_orig_query = db.query
|
|
|
|
# ————— V20: fetch_balance — parse chain get_currency_balance —————
|
|
dist.requests.post = lambda url, json=None, timeout=30: fake_post(
|
|
['52186.0417 VEX'])
|
|
check('V20 fetch_balance parse', dist.fetch_balance() == Decimal('52186.0417'))
|
|
dist.requests.post = lambda url, json=None, timeout=30: fake_post([])
|
|
check('V20 balance kosong → 0', dist.fetch_balance() == Decimal('0'))
|
|
|
|
# ————— V20: compute_shares — floor 4 desimal, sisa tetap di akun —————
|
|
shares = dist.compute_shares(Decimal('100.0000'),
|
|
[('a', 300.0), ('b', 700.0)])
|
|
check('V20 porsi proporsional stake',
|
|
shares == [('a', Decimal('30.0000')), ('b', Decimal('70.0000'))])
|
|
|
|
rem = dist.compute_shares(Decimal('0.0003'), [('a', 1.0), ('b', 1.0)])
|
|
check('V20 Σshare ≤ balance (sisa dibuang)',
|
|
sum(s for _, s in rem) == Decimal('0.0002'))
|
|
check('V20 sisa tetap di akun', rem == [('a', Decimal('0.0001')),
|
|
('b', Decimal('0.0001'))])
|
|
|
|
z = dist.compute_shares(Decimal('0.0002'),
|
|
[('a', 1.0), ('b', 10000.0)])
|
|
check('V20 porsi nol dibuang (a)', len(z) == 1 and z[0][0] == 'b')
|
|
check('V20 total_stake nol → []', dist.compute_shares(Decimal('1'),
|
|
[('x', 0.0)]) == [])
|
|
|
|
# ————— V26: dry-run — rencana tanpa tanda tangan / tanpa tulis DB —————
|
|
make_voters_db(db.DB_PATH, [('aaa1', '1.0', 1200.0, '2026-08-05T00:00:00',
|
|
'2026-08-05T00:00:00')])
|
|
dist.fetch_balance = lambda: Decimal('100.0000')
|
|
dist.BP_PRIVATE_KEY = ''
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = dist.run_distribution(dry_run=True)
|
|
out = buf.getvalue()
|
|
check('V26 dry-run exit 0', rc == 0)
|
|
check('V26 dry-run menampilkan rencana', 'RENCANA' in out and 'aaa1' in out)
|
|
conn = sqlite3.connect(db.DB_PATH)
|
|
tbls = [r[0] for r in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table'").fetchall()]
|
|
conn.close()
|
|
check('V26 dry-run ⊥ menulis DB', 'distribute_runs' not in tbls)
|
|
|
|
# ————— V25: no-op — pemilih kosong / saldo kecil —————
|
|
dist.fetch_balance = lambda: Decimal('100.0000')
|
|
db.query = lambda sql, params=(): []
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = dist.run_distribution(dry_run=False)
|
|
check('V25 pemilih kosong → no-op', rc == 0 and 'no-op' in buf.getvalue())
|
|
|
|
make_voters_db(db.DB_PATH, [('aaa1', '1.0', 1200.0, '2026-08-05T00:00:00',
|
|
'2026-08-05T00:00:00')])
|
|
dist.fetch_balance = lambda: Decimal('0.0000')
|
|
db.query = _orig_query
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = dist.run_distribution(dry_run=False)
|
|
check('V25 saldo kecil → no-op', rc == 0 and 'no-op' in buf.getvalue())
|
|
|
|
# ————— V26: run nyata tanpa key → RuntimeError —————
|
|
dist.fetch_balance = lambda: Decimal('100.0000')
|
|
dist.BP_PRIVATE_KEY = ''
|
|
try:
|
|
dist.run_distribution(dry_run=False)
|
|
raise AssertionError('V26 gagal: tanpa key harusnya RuntimeError')
|
|
except RuntimeError:
|
|
check('V26 tanpa VEX_BP_PRIVATE_KEY → RuntimeError', True)
|
|
|
|
# ————— V21/V23/V24: run nyata — pending→sent, memo per run —————
|
|
make_voters_db(db.DB_PATH, [('aaa1', '1.0', 300.0, '2026-08-05T00:00:00',
|
|
'2026-08-05T00:00:00'),
|
|
('bbb2', '1.0', 700.0, '2026-08-05T00:00:00',
|
|
'2026-08-05T00:00:00')])
|
|
dist.fetch_balance = lambda: Decimal('100.0000')
|
|
dist.BP_PRIVATE_KEY = '5Ktest'
|
|
dist.DISTRIBUTE_MAX_ATTEMPTS = 3
|
|
calls = {'runs': [], 'pays': [], 'updates': [], 'run_updates': []}
|
|
|
|
def fake_record_run(run_date, bal, voters, stake, sent, status, created):
|
|
calls['runs'].append((run_date, status))
|
|
return 7
|
|
|
|
def fake_record_payment(run_id, owner, amount, memo, status, created,
|
|
txid=None, error=None):
|
|
calls['pays'].append((owner, amount, memo, status))
|
|
return len(calls['pays'])
|
|
|
|
def fake_update(pid, status, txid=None, error=None, updated_at=None):
|
|
calls['updates'].append((pid, status, txid))
|
|
|
|
def fake_run_update(run_id, status, total_sent):
|
|
calls['run_updates'].append((run_id, status, total_sent))
|
|
|
|
db.record_run = fake_record_run
|
|
db.record_payment = fake_record_payment
|
|
db.update_payment_status = fake_update
|
|
db.update_run_status = fake_run_update
|
|
dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('tx' + to)
|
|
buf = io.StringIO()
|
|
with redirect_stdout(buf):
|
|
rc = dist.run_distribution(dry_run=False)
|
|
check('V21 run nyata exit 0', rc == 0)
|
|
check('V21 run tercatat (run_date)', calls['runs'][0][0]
|
|
== datetime.now().strftime('%Y-%m-%d'))
|
|
check('V21 2 payment pending tercatat',
|
|
len([p for p in calls['pays'] if p[3] == 'pending']) == 2)
|
|
today = datetime.now().strftime('%Y-%m-%d')
|
|
check('V23 memo per run',
|
|
all(p[2] == f'DATABISNISID PROFIT SHARE {today}'
|
|
for p in calls['pays']))
|
|
check('V21 semua sent + txid',
|
|
all(u[1] == 'sent' and u[2].startswith('tx') for u in calls['updates']))
|
|
check('V24 status run ok + total_sent',
|
|
calls['run_updates'] == [(7, 'ok', 100.0)])
|
|
|
|
# ————— V22: guard anti-duplikat — timeout lalu terverifikasi mendarat —————
|
|
dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('txL' + to, fails=1)
|
|
dist.verify_txid = lambda txid: True # tx sebenarnya mendarat
|
|
calls['updates'] = []
|
|
calls['run_updates'] = []
|
|
with redirect_stdout(io.StringIO()):
|
|
dist.run_distribution(dry_run=False)
|
|
check('V22 tx timeout tapi mendarat → sent tanpa resend',
|
|
all(u[1] == 'sent' for u in calls['updates']))
|
|
check('V22 terverifikasi dengan txid lama',
|
|
all(u[2].startswith('txL') for u in calls['updates']))
|
|
check('V22 run status ok',
|
|
calls['run_updates'][0][1] == 'ok')
|
|
|
|
# ————— V22: verifikasi gagal (None) → tahan, jangan resend —————
|
|
dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('txH' + to, fails=1)
|
|
dist.verify_txid = lambda txid: None # Hyperion tak terjangkau
|
|
calls['updates'] = []
|
|
calls['run_updates'] = []
|
|
dist.DISTRIBUTE_MAX_ATTEMPTS = 1 # sekali coba → ditahan
|
|
with redirect_stdout(io.StringIO()):
|
|
dist.run_distribution(dry_run=False)
|
|
check('V22 verifikasi None → ditahan (failed)',
|
|
all(u[1] == 'failed' for u in calls['updates']))
|
|
check('V22 run status partial',
|
|
calls['run_updates'][0][1] == 'partial')
|
|
|
|
print('\nSEMUA UJI PASS')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|