"""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). Pemilih buatan memakai bobot segar (revote ~sekarang); sejak V40 baris basi JUGA disimpan (banding/tampil/bayar menghitung umur saat query) — contoh basi membuktikan ia lolos ke snapshot, bukan dibuang. """ import os import sqlite3 import tempfile from datetime import datetime, timedelta, timezone import requests import db import get_voters as gv def _fresh_weight(staked_raw, days_ago=5): """Bobot utk revote ~sekarang: staked_raw × 2^(pekan/52) — kuantisasi pekan utuh sejak epoch 2000 (formula stake2vote Vexanium).""" weeks = ((datetime.now() - gv.WEIGHT_EPOCH).days - days_ago) // 7 return f'{staked_raw * 2 ** (weeks / 52.0):.17f}' 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': _fresh_weight(12000000)}, {'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': _fresh_weight(12000000)}, {'owner': 'ccc3', 'producers': [gv.TARGET_BP], 'staked': 10000000, 'last_vote_weight': _fresh_weight(10000000)}, ], 'more': True}, 'ccc3': {'rows': [ {'owner': 'ccc3', 'producers': [gv.TARGET_BP], 'staked': 10000000, 'last_vote_weight': _fresh_weight(10000000)}, {'owner': 'zzz9', 'producers': [gv.TARGET_BP], 'staked': 15000000, 'last_vote_weight': _fresh_weight(15000000, days_ago=900)}, ], '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() db.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', 'zzz9']) # 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', 'zzz9') for n in norm)) by = {n['owner']: n for n in norm} aaa1_w = _fresh_weight(12000000) check('V6 bobot verbatim TEXT', by['aaa1']['weight'] == aaa1_w) 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; V40: basi tetap disimpan check('V13 last_vote ikut dinormalisasi', 'last_vote' in by['aaa1']) check('V13 last_vote segar (≤ VEX_STALE_DAYS)', datetime.fromisoformat(by['aaa1']['last_vote']) >= datetime.now() - timedelta(days=28)) check('V40 basi (zzz9) tetap disimpan', by['zzz9']['owner'] == 'zzz9') check('V40 last_vote basi zzz9 ~900 hari lalu', (datetime.now() - datetime.fromisoformat(by['zzz9']['last_vote'])).days > 800) # V13: formula — staked_raw × 2^(pekan/52), kuantisasi pekan utuh sejak 2000 # 2^1 → 52 pekan setelah 2000-01-01 = 2000-12-30 (52×7 = 364 hari) lv = gv.derive_last_vote(f'{12000000 * 2 ** 1:.17f}', 1200.0) lv_dt = datetime.fromisoformat(lv) if lv else None check('V13 formula 2^1 (52 pekan) → 2000-12-30', lv_dt is not None and lv_dt == datetime(2000, 12, 30)) # 2^26 → 26×52 pekan = 1352 pekan setelah 2000 = 2025-11-29 lv = gv.derive_last_vote(f'{12000000 * 2 ** 26:.17f}', 1200.0) lv_dt = datetime.fromisoformat(lv) if lv else None check('V13 formula 2^26 (1352 pekan) → 2025-11-29', lv_dt is not None and lv_dt == datetime(2025, 11, 29)) 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) # V13/B12: regresi formula pekan utuh — baris nyata dari chain # ..tg: staked_raw 9000, weight → t×52 = 1291 pekan → 2024-09-28 lv = gv.derive_last_vote('267849632990.26596069335937500', 9000 / 10000) check('V13 regresi ..tg (1291 pekan) → 2024-09-28', lv == '2024-09-28T00:00:00') # 1.crownz: staked_raw 17115095, t×52 = 1314 pekan → 2025-03-08 lv = gv.derive_last_vote( f'{17115095 * 2 ** (1314 / 52.0):.17f}', 17115095 / 10000) check('V13 regresi 1.crownz (1314 pekan) → 2025-03-08', lv == '2025-03-08T00:00:00') # V13/V40: normalize menyimpan basi, membuang yang tak dapat diverifikasi old = gv.normalize({'owner': 'old8', 'producers': [gv.TARGET_BP], 'staked': 20000000, 'last_vote_weight': _fresh_weight(20000000, days_ago=900)}) check('V40 normalize basi (900 hari) → disimpan', old is not None and old['owner'] == 'old8') check('V13 normalize bobot kosong (unverifiable) → None', gv.normalize({'owner': 'unk7', 'producers': [gv.TARGET_BP], 'staked': 20000000, 'last_vote_weight': ''}) is None) check('V13 normalize segar → disimpan', gv.normalize({'owner': 'new6', 'producers': [gv.TARGET_BP], 'staked': 20000000, 'last_vote_weight': _fresh_weight(20000000)})['owner'] == 'new6') # V13: estimasi yang melewati now (unstake tanpa revote) diklamp ke now fixed_now = datetime(2026, 8, 5, 12, 0, 0) weeks_future = ((fixed_now - gv.WEIGHT_EPOCH).days + 14) // 7 # 2 pekan ke depan fut = gv.normalize({'owner': 'fut9', 'producers': [gv.TARGET_BP], 'staked': 20000000, 'last_vote_weight': f'{20000000 * 2 ** (weeks_future / 52.0):.17f}'}, now=fixed_now) check('V13 last_vote masa depan diklamp ke now', fut is not None and fut['last_vote'] == fixed_now.isoformat(timespec='seconds')) # 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': _fresh_weight(10000000)})['staked'] == 1000.0) # V8: node kadang memberi staked sebagai string → tetap dinormalisasi srow = {'owner': 'eee5', 'producers': [gv.TARGET_BP], 'staked': '10000010000', 'last_vote_weight': _fresh_weight(10000010000)} 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}, ] scanned_a = gv.persist(set_a) conn = sqlite3.connect(db.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}, ] scanned_b = gv.persist(set_b) conn = sqlite3.connect(db.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) # V62: stempel scan UTC-naive (konsisten dgn cutoff umur dashboard/ # distribusi yang pakai `datetime.now(timezone.utc)`). Dulu `datetime.now()` # (waktu lokal container, TZ=Asia/Jakarta UTC+7) → `first_seen_at` WIB # dibandingkan sbg string UTC → maturity pemilih BARU bergeser +7 jam # (B24). Guard: stamp ≈ now-UTC-naive, ⊖ WIB (offset ±2 mnt toleransi). utc_now = datetime.now(timezone.utc).replace(tzinfo=None) for s in (scanned_a, scanned_b): stamp = datetime.fromisoformat(s) delta = abs((utc_now - stamp).total_seconds()) check(f'V62 stempel scan UTC-naive ({s})', delta < 120 and stamp.tzinfo is None) wib_delta = abs((utc_now - timedelta(hours=7) - stamp).total_seconds()) check(f'V62 stempel scan ⊖ WIB ({s})', wib_delta > 60) # V41: first_seen — kemunculan pertama tiap owner, idempoten (⊥ overwrite) db.record_first_seen([{'owner': 'aaa1'}], '2000-01-01T00:00:00') conn = sqlite3.connect(db.DB_PATH) fs = {r[0]: r[1] for r in conn.execute( 'SELECT owner, first_seen_at FROM voter_first_seen').fetchall()} conn.close() check('V41 first_seen tercatat utk semua owner', set(fs) == {'aaa1', 'ccc3', 'ddd4'}) check('V41 first_seen = kemunculan pertama', fs['aaa1'] == scanned_a and fs['ccc3'] == scanned_a and fs['ddd4'] == scanned_b) check('V41 first_seen idempoten (record ulang ⊥ overwrite)', fs['aaa1'] == scanned_a and fs['aaa1'] != '2000-01-01T00:00:00') print('\nSEMUA UJI PASS') if __name__ == '__main__': main()