"""Uji verifikasi klaim reward BP (oracle §V32..V34). `./venv/bin/python test_claim.py` harus exit 0. Memakai DB sementara + mock jaringan/pyntelope — ⊥ klaim nyata. Verifikasi: jendela klaim (V32), parse respons `get_table_rows` bentuk asli `{rows:[...]}` + recovery window lewat, math fee (V33), siklus penuh klaim→fee, retry fee, resume, reward 0 → skipped, klaim yang mendarat walau timeout (V34). """ import os import sqlite3 import sys import tempfile from datetime import datetime, timedelta from decimal import Decimal import db import claim import telegram 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, exc=None): self._txid = txid self._exc = exc def id(self): return self._txid def send(self): if self._exc is not None: raise self._exc return {'processed': True} def fresh_db(tag): """DB sqlite sementara baru per bagian uji.""" tmp = tempfile.mkdtemp() db.DB_PATH = os.path.join(tmp, f'claim-{tag}.db') return db.DB_PATH def last_claim_row(): row = db.queryone( 'SELECT claim_id, run_date, claim_txid, reward, fee_amount, ' ' fee_status, fee_txid FROM claim_runs ORDER BY claim_id DESC ' 'LIMIT 1') return row def fake_post(data): return type('R', (), {'raise_for_status': lambda self: None, 'json': lambda self: data})() def main(): # Matikan sleep nyata (settle/poll) + Telegram selama uji. claim.time.sleep = lambda s: None _tg_token = telegram.TELEGRAM_BOT_TOKEN _tg_comm = telegram.TELEGRAM_COMMUNITY_CHAT_IDS _tg_int = telegram.TELEGRAM_INTERNAL_CHAT_IDS _tg_post = telegram.requests.post telegram.TELEGRAM_BOT_TOKEN = '' telegram.TELEGRAM_COMMUNITY_CHAT_IDS = [] telegram.TELEGRAM_INTERNAL_CHAT_IDS = [] try: # ————— V32: parse_claim_time — format chain ————— check('V32 parse format %f', claim.parse_claim_time('2026-08-05T07:01:44.500') == datetime(2026, 8, 5, 7, 1, 44, 500000)) check('V32 parse format detik', claim.parse_claim_time('2026-08-05T07:01:44') == datetime(2026, 8, 5, 7, 1, 44)) check('V32 parse kosong → None', claim.parse_claim_time(None) is None) check('V32 parse sampah → None', claim.parse_claim_time('x') is None) # ————— V32: fetch_producer_last_claim — bentuk asli get_table_rows ————— claim._post_json = lambda url, payload: { 'rows': [{'last_claim_time': '2026-08-05T07:01:44.500'}], 'more': False, 'next_key': ''} check('V32 parse respons asli {rows:[...]} → waktu', claim.fetch_producer_last_claim() == '2026-08-05T07:01:44.500') claim._post_json = lambda url, payload: {'rows': [], 'more': False, 'next_key': ''} check('V32 rows kosong → None', claim.fetch_producer_last_claim() is None) claim._post_json = lambda url, payload: {} check('V32 respons tak dikenal → None (⊥ crash)', claim.fetch_producer_last_claim() is None) # ————— V32: next_window = last + 24 jam; lewat → klaim segera ————— claim.fetch_producer_last_claim = lambda: '2026-08-05T07:01:44.500' check('V32 jendela = last + 24 jam', claim.next_window(datetime(2026, 8, 6, 5, 0, 0)) == datetime(2026, 8, 6, 7, 1, 44, 500000)) check('V32 window lewat (≥24 jam) → now (recovery, ⊥ deadlock)', claim.next_window(datetime(2026, 8, 6, 12, 0, 0)) == datetime(2026, 8, 6, 12, 0, 0)) claim.fetch_producer_last_claim = lambda: None check('V32 tanpa last → now (langsung coba)', claim.next_window(datetime(2026, 8, 6, 12, 0, 0)) == datetime(2026, 8, 6, 12, 0, 0)) # ————— V32/V33: siklus penuh — klaim → catat → fee terkirim ————— fresh_db('cycle') claim.fetch_producer_last_claim = lambda: ( datetime.now() - timedelta(hours=25) ).strftime('%Y-%m-%dT%H:%M:%S') balances = iter([Decimal('100.0000'), Decimal('110.0004')]) dist.fetch_balance = lambda: next(balances) claim.build_claim_action = lambda: FakeSigned('txclaim') dist.verify_txid = lambda txid: True r1 = claim.step() check('V32 step klaim sukses → claimed', r1 == 'claimed') row = last_claim_row() check('V32 klaim tercatat (reward 10.0004, fee 1.0000 floor)', row is not None and row[3] == 10.0004 and row[4] == 1.0 and row[2] == 'txclaim' and row[5] == 'pending') memos = [] def fake_fee_sign(to, amount, memo): memos.append((to, amount, memo)) return FakeSigned('txfee') dist.build_signed_transfer = fake_fee_sign r2 = claim.step() check('V34 fee terkirim → fee_sent', r2 == 'fee_sent') check('V33 fee ke wallet bpdbsjasprod', memos[-1][0] == 'bpdbsjasprod') check('V33 memo BP FEE + tanggal', memos[-1][2] == f'BP FEE {datetime.now().strftime("%Y-%m-%d")}') row = last_claim_row() check('V34 fee ditandai sent + txid', row[5] == 'sent' and row[6] == 'txfee') # ————— V33/V34: fee gagal → failed → diulang sampai terkirim ————— fresh_db('retry') db.record_claim('2026-08-05', 'txold', 10.0, 1.0, 'pending', '2026-08-05T08:00:00') dist.build_signed_transfer = lambda to, amount, memo: FakeSigned( 'txfail', exc=RuntimeError('net down')) dist.verify_txid = lambda txid: False check('V34 fee gagal → fee_failed', claim.step() == 'fee_failed') check('V34 status failed', last_claim_row()[5] == 'failed') dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('txok') check('V34 fee diulang → fee_sent', claim.step() == 'fee_sent') row = last_claim_row() check('V34 fee akhirnya sent', row[5] == 'sent' and row[6] == 'txok') # ————— V33: reward 0 → fee skipped; step lanjut ke jadwal ————— fresh_db('zero') claim.fetch_producer_last_claim = lambda: ( datetime.now() - timedelta(hours=25) ).strftime('%Y-%m-%dT%H:%M:%S') balances = iter([Decimal('100.0000'), Decimal('100.0000')]) dist.fetch_balance = lambda: next(balances) claim.build_claim_action = lambda: FakeSigned('txzero') dist.verify_txid = lambda txid: True check('V33 klaim reward 0 → claimed', claim.step() == 'claimed') row = last_claim_row() check('V33 fee 0 → skipped', row[4] == 0.0 and row[5] == 'skipped') # pending_claim_fee ⊥ memuat fee 0 → step lanjut ke jadwal (sleep) claim.fetch_producer_last_claim = lambda: ( datetime.now() - timedelta(hours=20) ).strftime('%Y-%m-%dT%H:%M:%S') check('V34 ⊥ fee pending → jadwal (sleep)', claim.step() == 'sleep') # ————— V34: klaim timeout tapi mendarat (verify True) → diukur ————— fresh_db('landed') claim.fetch_producer_last_claim = lambda: ( datetime.now() - timedelta(hours=25) ).strftime('%Y-%m-%dT%H:%M:%S') balances = iter([Decimal('50.0000'), Decimal('60.0000')]) dist.fetch_balance = lambda: next(balances) claim.build_claim_action = lambda: FakeSigned( 'txL', exc=RuntimeError('broadcast timeout')) dist.verify_txid = lambda txid: True check('V34 timeout tapi mendarat → claimed', claim._try_claim_once() == 'claimed') row = last_claim_row() check('V34 reward terukur', row[3] == 10.0 and row[2] == 'txL') # ————— V34: verify None + last_claim_time maju → dianggap mendarat ————— fresh_db('reconcile') claim.fetch_producer_last_claim = lambda: ( datetime.now() - timedelta(hours=1) ).strftime('%Y-%m-%dT%H:%M:%S') balances = iter([Decimal('10.0000'), Decimal('12.0000')]) dist.fetch_balance = lambda: next(balances) claim.build_claim_action = lambda: FakeSigned( 'txN', exc=RuntimeError('timeout')) dist.verify_txid = lambda txid: None check('V34 verify None + last maju → claimed', claim._try_claim_once() == 'claimed') check('V34 reward tetap diukur', last_claim_row()[3] == 2.0) # ————— V32: klaim ditolak → retry; poll deadline lewat → timeout ————— fresh_db('timeout') claim.build_claim_action = lambda: FakeSigned( 'txT', exc=RuntimeError('rejected')) dist.verify_txid = lambda txid: False dist.fetch_balance = lambda: Decimal('100.0000') check('V32 klaim ditolak → retry', claim._try_claim_once() == 'retry') check('V32 deadline lewat → timeout', claim.poll_claim(datetime.now() - timedelta(seconds=1)) == 'timeout') # ————— resume: fee failed dari DB → step mengirimnya ————— fresh_db('resume') db.record_claim('2026-08-05', 'txr', 10.0, 1.0, 'failed', '2026-08-05T08:00:00') dist.build_signed_transfer = lambda to, amount, memo: FakeSigned('txr1') dist.verify_txid = lambda txid: True check('V34 resume fee dari DB → fee_sent', claim.step() == 'fee_sent') check('V34 resume → sent', last_claim_row()[5] == 'sent' and last_claim_row()[6] == 'txr1') # ————— V36: klaim reward → kanal INTERNAL saja, ⊥ komunitas ————— tg = {'posts': []} def tg_post(url, json=None, timeout=None): tg['posts'].append((url, json['chat_id'], json['text'])) return fake_post({}) telegram.TELEGRAM_BOT_TOKEN = '123:TEST' telegram.TELEGRAM_COMMUNITY_CHAT_IDS = ['-1001'] telegram.TELEGRAM_INTERNAL_CHAT_IDS = ['-1002'] telegram.requests.post = tg_post telegram.notify_claim('10.0000', '1.0000', 'bpdbsjasprod') telegram.notify_claim_failure(RuntimeError('net down')) check('V36 klaim → internal saja', len(tg['posts']) == 2 and all(p[1] == '-1002' for p in tg['posts'])) check('V36 ⊥ komunitas utk klaim', not any(p[1] == '-1001' for p in tg['posts'])) check('V36 isi notif klaim', 'KLAIM REWARD SUKSES' in tg['posts'][0][2] and 'KLAIM GAGAL' in tg['posts'][1][2]) finally: claim.time.sleep = time.sleep telegram.TELEGRAM_BOT_TOKEN = _tg_token telegram.TELEGRAM_COMMUNITY_CHAT_IDS = _tg_comm telegram.TELEGRAM_INTERNAL_CHAT_IDS = _tg_int telegram.requests.post = _tg_post print('\nSEMUA UJI PASS') if __name__ == '__main__': import time sys.exit(main())