databisnisid/test_get_voters.py

"""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

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)

    # 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'])

    print('\nSEMUA UJI PASS')


if __name__ == '__main__':
    main()