"""Uji verifikasi invarian §V tanpa bergantung node publik (flaky). Oracle: `./venv/bin/python test_get_voters.py` harus exit 0. Dalam setiap contoh, 'last_vote_weight' berisi 17 desimal (bobot teliti). """ import os import sqlite3 import tempfile from datetime import datetime import requests import get_voters as gv class FakePayload: passes = 0 @staticmethod def call(payload): lb = payload['lower_bound'] pages = { '': {'rows': [ {'owner': gv.ZERO_OWNER, 'producers': [], 'staked': 10000, 'last_vote_weight': '0.00'}, {'owner': 'aaa1', 'producers': [gv.TARGET_BP], 'staked': 12000000, 'last_vote_weight': '12345.678901234567890'}, {'owner': 'bbb2', 'producers': ['wind'], 'staked': 9000, 'last_vote_weight': '999.0'}, ], 'more': True}, 'bbb2': {'rows': [ {'owner': 'aaa1', 'producers': [gv.TARGET_BP], 'staked': 12000000, 'last_vote_weight': '12345.678901234567890'}, {'owner': 'ccc3', 'producers': [gv.TARGET_BP], 'staked': 10000000, 'last_vote_weight': '0.50000000000000000'}, ], 'more': True}, 'ccc3': {'rows': [ {'owner': 'ccc3', 'producers': [gv.TARGET_BP], 'staked': 10000000, 'last_vote_weight': '0.50000000000000000'}, ], 'more': False}, } FakePayload.passes += 1 if FakePayload.passes > 20: raise AssertionError('V2 gagal: loop tidak berterminasi') return pages[lb] def check(tag, cond): if not cond: raise AssertionError(f'GAGAL: {tag}') print(f'ok: {tag}') def main(): tmp = tempfile.mkdtemp() gv.DB_PATH = os.path.join(tmp, 'test.db') # V1: ulang ≥3x saat node timeout → hits, lalu sukses calls = {'n': 0} def fake_post(url, json=None, timeout=30): calls['n'] += 1 if calls['n'] < 3: raise requests.exceptions.Timeout('node timeout') resp = type('R', (), {'ok': True, 'raise_for_status': lambda self: None, 'json': lambda self: {'rows': [], 'more': False}})() return resp gv.requests.post = fake_post out = gv.fetch_with_retry({}) check(f'V1 retry ≥3 lalu sukses ({calls["n"]} panggilan)', calls['n'] == 3 and out == {'rows': [], 'more': False}) # bila gagal terus → mengangkat setelah retries habis def always_fail(url, json=None, timeout=30): raise requests.exceptions.Timeout('node timeout') gv.requests.post = always_fail try: gv.fetch_with_retry({}, max_retries=3) raise AssertionError('V1 gagal: seharusnya mengangkat setelah retries habis') except requests.exceptions.Timeout: check('V1 gagal total → error dinaikkan', True) # V2 + V3: paginasi berakhir & baris duplikat antar-halaman dibuang gv.fetch_with_retry = FakePayload.call raw = gv.scan_voters() owners = [r['owner'] for r in raw] check('V2/V3 pembuangan duplikat', owners == [gv.ZERO_OWNER, 'aaa1', 'bbb2', 'ccc3']) # V4 + V5 + V6: saringan sentinel & BP target terjadi di normalize norm = [gv.normalize(r) for r in raw] norm = [n for n in norm if n is not None] norm_owners = [n['owner'] for n in norm] check('V4 sentinel "0" dilewati', gv.ZERO_OWNER not in norm_owners) check('V5 hanya BP target', all(n['owner'] in ('aaa1', 'ccc3') for n in norm)) by = {n['owner']: n for n in norm} check('V6 bobot verbatim TEXT', by['aaa1']['weight'] == '12345.678901234567890') check('V6 bobot bukan float-rounded', by['aaa1']['weight'] != str(float(by['aaa1']['weight']))) check('staked ÷10000', by['aaa1']['staked'] == 1200.0) check('staked batas ≥1000 ÷10000', by['ccc3']['staked'] == 1000.0) # V13: last_vote diturunkan di normalize dari rasio bobot/stake check('V13 last_vote ikut dinormalisasi', 'last_vote' in by['aaa1']) check('V13 bobot/stake rasio kecil → sebelum 2000', by['ccc3']['last_vote'] is not None and by['ccc3']['last_vote'] < '2000-01-01') # V13: formula — staked_raw × 2^26 → ~2026-01-01 (±2 hari, toleransi 365.25 hari/tahun) lv = gv.derive_last_vote(f'{12000000 * 2 ** 26}.00000000000000000', 1200.0) lv_dt = datetime.fromisoformat(lv) if lv else None check('V13 formula 2^26 → ~2026-01-01', lv_dt is not None and abs((lv_dt - datetime(2026, 1, 1)).days) <= 2) check('V13 rasio 1.0 → epoch 2000', gv.derive_last_vote('12000000.00000000000000000', 1200.0) == '2000-01-01T00:00:00') check('V13 bobot nol → None', gv.derive_last_vote('0', 1200.0) is None) check('V13 bobot kosong → None', gv.derive_last_vote('', 1200.0) is None) check('V13 bobot tak valid → None', gv.derive_last_vote('abc', 1200.0) is None) check('V13 stake nol → None', gv.derive_last_vote('1.0', 0.0) is None) # V9: staked < 1000 VEX → dibuang; tepat 1000 → disimpan check('V9 di bawah 1000 dibuang', gv.normalize({'owner': 'low9', 'producers': [gv.TARGET_BP], 'staked': 9999999, 'last_vote_weight': '1.0'}) is None) check('V9 tepat 1000 disimpan', gv.normalize({'owner': 'eq10', 'producers': [gv.TARGET_BP], 'staked': 10000000, 'last_vote_weight': '1.0'})['staked'] == 1000.0) # V8: node kadang memberi staked sebagai string → tetap dinormalisasi srow = {'owner': 'eee5', 'producers': [gv.TARGET_BP], 'staked': '10000010000', 'last_vote_weight': '42.0'} snorm = gv.normalize(srow) check('V8 staked string → float ÷10000', snorm['staked'] == 1000001.0) # V7: replace-snapshot — tiap persist mengganti tabel, ⊥ append/owner ganda set_a = [ {'owner': 'aaa1', 'weight': '111.0', 'staked': 1200.0}, {'owner': 'ccc3', 'weight': '222.0', 'staked': 1000.0}, ] gv.persist(set_a) conn = sqlite3.connect(gv.DB_PATH) cnt = conn.execute('SELECT COUNT(*) FROM voters').fetchone()[0] conn.close() check('V7 snapshot pertama = 2', cnt == 2) set_b = [ # aaa1 weight berubah; ccc3 dihapus; ddd4 baru {'owner': 'aaa1', 'weight': '999.0', 'staked': 5000.0}, {'owner': 'ddd4', 'weight': '333.0', 'staked': 2000.0}, ] gv.persist(set_b) conn = sqlite3.connect(gv.DB_PATH) rows = conn.execute('SELECT owner, weight, staked, scanned_at FROM voters').fetchall() by = {r[0]: r for r in rows} cols = [r[1] for r in conn.execute('PRAGMA table_info(voters)').fetchall()] pk = [r[1] for r in conn.execute('PRAGMA table_info(voters)').fetchall() if r[5]] conn.close() check('V7 count = |set_b|', len(rows) == 2) check('V7 ccc3 (dihapus) tak ada', 'ccc3' not in by) check('V7 ddd4 (baru) ada', 'ddd4' in by) check('V7 bobot aaa1 terbaru', by['aaa1'][1] == '999.0') stamps = set(r[3] for r in rows) check('V7 semua baris cap waktu scan ini', len(stamps) == 1) check('V7 owner PRIMARY KEY', 'owner' in cols and pk == ['owner']) check('V7 kolom last_vote ada', 'last_vote' in cols) print('\nSEMUA UJI PASS') if __name__ == '__main__': main()