299 lines
12 KiB
Python
299 lines
12 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
|
|
import telegram
|
|
|
|
|
|
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
|
|
|
|
# Oracle: Telegram ⊥ boleh mengirim nyata — nonaktifkan dulu, dipakai lagi §V30
|
|
_tg_token = telegram.TELEGRAM_BOT_TOKEN
|
|
_tg_chats = telegram.TELEGRAM_CHAT_IDS
|
|
_tg_post = telegram.requests.post
|
|
telegram.TELEGRAM_BOT_TOKEN = ''
|
|
telegram.TELEGRAM_CHAT_IDS = []
|
|
|
|
# ————— 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, status, created,
|
|
txid=None, error=None):
|
|
calls['pays'].append((owner, amount, 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
|
|
signed = []
|
|
|
|
def fake_sign(to, amount, memo):
|
|
signed.append(memo)
|
|
return FakeSigned('tx' + to)
|
|
|
|
dist.build_signed_transfer = fake_sign
|
|
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[2] == 'pending']) == 2)
|
|
today = datetime.now().strftime('%Y-%m-%d')
|
|
check('V23 memo on-chain per run',
|
|
all(m == f'DATABISNISID PROFIT SHARE {today}' for m in signed))
|
|
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')
|
|
|
|
# ————— V30: notifikasi Telegram — best-effort, ⊥ dry-run/no-op —————
|
|
tg = {'posts': []}
|
|
|
|
def tg_post(url, json=None, timeout=None):
|
|
tg['posts'].append((url, json['text']))
|
|
return fake_post({})
|
|
|
|
telegram.TELEGRAM_BOT_TOKEN = '123:TEST'
|
|
telegram.TELEGRAM_CHAT_IDS = ['-100123']
|
|
telegram.requests.post = tg_post
|
|
|
|
# send_text tanpa token → no-op senyap (⊥ post)
|
|
telegram.TELEGRAM_BOT_TOKEN = ''
|
|
telegram.send_text('tidak boleh terkirim')
|
|
check('V30 ⊥ token → no-op senyap', tg['posts'] == [])
|
|
telegram.TELEGRAM_BOT_TOKEN = '123:TEST'
|
|
|
|
# dry-run → ⊥ notif Telegram
|
|
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')])
|
|
db.query = _orig_query
|
|
dist.fetch_balance = lambda: Decimal('100.0000')
|
|
dist.BP_PRIVATE_KEY = '5Ktest'
|
|
dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('txT' + to)
|
|
dist.DISTRIBUTE_MAX_ATTEMPTS = 3
|
|
dist.verify_txid = lambda txid: True
|
|
calls['runs'] = []
|
|
with redirect_stdout(io.StringIO()):
|
|
dist.run_distribution(dry_run=True)
|
|
check('V30 dry-run ⊥ notif', tg['posts'] == [])
|
|
|
|
# run nyata → notif mulai + selesai
|
|
calls['updates'] = []
|
|
calls['run_updates'] = []
|
|
with redirect_stdout(io.StringIO()):
|
|
rc = dist.run_distribution(dry_run=False)
|
|
check('V30 run nyata exit 0', rc == 0)
|
|
check('V30 2 notif (mulai + selesai)',
|
|
len(tg['posts']) == 2)
|
|
check('V30 mulai tercatat', 'DISTRIBUSI MULAI' in tg['posts'][0][1])
|
|
check('V30 selesai + status ok', 'DISTRIBUSI SELESAI' in tg['posts'][1][1]
|
|
and 'Status: ok' in tg['posts'][1][1])
|
|
check('V30 ringkasan jumlah', '2/2 pemilih' in tg['posts'][1][1])
|
|
|
|
# no-op (pemilih kosong) → ⊥ notif baru
|
|
db.query = lambda sql, params=(): []
|
|
with redirect_stdout(io.StringIO()):
|
|
dist.run_distribution(dry_run=False)
|
|
check('V30 no-op ⊥ notif', len(tg['posts']) == 2)
|
|
db.query = _orig_query
|
|
|
|
# failure (⊥ key) → notif gagal lewat main(), lalu re-raise
|
|
dist.BP_PRIVATE_KEY = ''
|
|
try:
|
|
dist.main()
|
|
raise AssertionError('V30 gagal: main harusnya re-raise')
|
|
except RuntimeError:
|
|
check('V30 failure → re-raise RuntimeError', True)
|
|
check('V30 notif gagal terkirim',
|
|
len(tg['posts']) == 3 and 'DISTRIBUSI GAGAL' in tg['posts'][2][1])
|
|
dist.BP_PRIVATE_KEY = '5Ktest'
|
|
|
|
telegram.TELEGRAM_BOT_TOKEN = _tg_token
|
|
telegram.TELEGRAM_CHAT_IDS = _tg_chats
|
|
telegram.requests.post = _tg_post
|
|
|
|
print('\nSEMUA UJI PASS')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|