databisnisid/test_dashboard.py

"""Uji verifikasi dashboard (oracle).

`./venv/bin/python test_dashboard.py` harus exit 0.
Memakai DB sementara, bukan voters.db asli (read-only invariant tetap diverifikasi).
"""

import os
import sqlite3
import tempfile

import dashboard


def check(tag, cond):
    if not cond:
        raise AssertionError(f'GAGAL: {tag}')
    print(f'ok: {tag}')


def make_db(path, rows):
    conn = sqlite3.connect(path)
    conn.execute('CREATE TABLE voters ('
                 'owner TEXT PRIMARY KEY, weight TEXT NOT NULL, '
                 'staked REAL NOT NULL, scanned_at TEXT NOT NULL)')
    conn.executemany(
        'INSERT INTO voters (owner, weight, staked, scanned_at) VALUES (?, ?, ?, ?)',
        rows,
    )
    conn.commit()
    conn.close()


def main():
    tmp = tempfile.mkdtemp()
    db = os.path.join(tmp, 'voters.db')
    rows = []
    for i in range(1, 126):  # 125 baris -> 3 halaman
        rows.append((f'acct{i:03d}', f'{i}.00000000000000000', float(i) * 10.0, '2026-08-04T00:00:00'))
    rows[5] = (rows[5][0], rows[5][1], 5.0 * 10.0, rows[5][3])  # non-monotonic guard
    make_db(db, rows)

    dashboard.DB_PATH = db  # arahkan app ke DB uji (read-only tetap dipertahankan)
    client = dashboard.app.test_client()

    # V10: urutan staked DESC (terbesar dulu), 50 per halaman
    resp = client.get('/?page=1')
    check('V10 halaman 1 → 200', resp.status_code == 200)
    html = resp.get_data(as_text=True)
    check('V10 judul halaman', 'DAFTAR PEMILIH' in html)

    staked_vals = []
    for i in range(1, 4):
        resp = client.get(f'/?page={i}')
        check(f'V10 halaman {i} → 200', resp.status_code == 200)
        staked_vals += _extract_stakes(resp.get_data(as_text=True))
    check('V10 total 125 baris ditarik', len(staked_vals) == 125)
    check('V10 urut non-naik', all(staked_vals[i] >= staked_vals[i + 1]
                                   for i in range(len(staked_vals) - 1)))

    # halaman melebihi total → dibatasi (tidak error)
    resp = client.get('/?page=99')
    check('V10 halaman ekstra → 200', resp.status_code == 200)

    # V10: read-only — tidak ada tabel baru / perubahan pada DB asli
    before = _table_hashes(db)
    client.get('/')
    check('V10 read-only (DB tak berubah)', _table_hashes(db) == before)

    # ————— V11: search server-side ?q= (substring, case-insensitive) —————
    resp = client.get('/?q=acct12')
    check('V11 search ?q= → 200', resp.status_code == 200)
    html = resp.get_data(as_text=True)
    check('V11 hanya akun cocok', 'acct125' in html and 'acct120' in html
          and 'acct119' not in html)
    check('V11 caption HASIL · 6', 'HASIL · 6 AKUN' in html)
    q_vals = _extract_stakes(html)
    check('V11 hasil search urut DESC', q_vals == sorted(q_vals, reverse=True)
          and len(q_vals) == 6)

    # pager membawa q
    check('V11 pager mempertahankan q', 'q=acct12' in html)

    # halaman ekstra saat filter → dipatok ke total_pages filter
    resp = client.get('/?q=acct12&page=99')
    check('V11 clamp halaman saat search', resp.status_code == 200)

    # search tanpa hasil → status kosong
    resp = client.get('/?q=zzzz')
    html = resp.get_data(as_text=True)
    check('V11 tak ada hasil → pesan kosong', 'TAK ADA AKUN COCOK' in html)

    # ————— V11: /api/search (JSON, rank global via ROW_NUMBER) —————
    resp = client.get('/api/search?q=acct12')
    data = resp.get_json()
    check('V11 /api/search shape', set(data.keys()) == {'query', 'count', 'cap', 'results'})
    check('V11 /api/search count', data['count'] == 6)
    owners = [r['owner'] for r in data['results']]
    ranks = [r['rank'] for r in data['results']]
    check('V11 /api/search cocok substring',
          all('acct12' in o for o in owners))
    check('V11 /api/search rank global',
          ranks == sorted(ranks) and ranks == list(range(1, 7)))
    check('V11 /api/search rank sejajar staked tertinggi',
          [r['owner'] for r in data['results']] ==
          ['acct125', 'acct124', 'acct123', 'acct122', 'acct121', 'acct120'])
    check('V11 /api/search urut staked DESC',
          [r['staked'] for r in data['results']] == sorted(
              [r['staked'] for r in data['results']], reverse=True))
    check('V11 /api/search bobot string verbatim',
          isinstance(data['results'][0]['weight'], str))

    resp = client.get('/api/search?q=zzzz')
    check('V11 /api/search kosong', resp.get_json()['count'] == 0)

    # rank HARUS global, bukan dalam-filter: 'acct00' menyentuh baris stake kecil
    resp = client.get('/api/search?q=acct00')
    data = resp.get_json()
    check('V11 rank global (bukan dalam-filter)',
          [r['rank'] for r in data['results']] == list(range(117, 126)))

    # ————— env config: default = konstanta lama —————
    import config
    check('V11 config default TARGET_BP', config.TARGET_BP == 'databisnisid')
    check('V11 config default API_NODE', config.API_NODE == 'https://v2.vexascan.com:2096')
    check('V11 config default PAGE_SIZE', config.PAGE_SIZE == 50)
    check('V11 config default MIN_STAKED_VEX', config.MIN_STAKED_VEX == 1000.0)
    check('V11 config default DB_PATH', config.DB_PATH == 'voters.db')


def _extract_stakes(text):
    """Ambil nilai stake (dalam urutan kemunculan) dari baris HTML."""
    vals = []
    for token in text.split('class="list-row"')[1:]:
        stake_token = token.split('<span class="stake"')[1].split('</span>')[0].split('>', 1)[1]
        vals.append(_parse_vex(stake_token))
    return vals


def _parse_vex(text):
    # format '1.749.791,2303' → float
    return float(text.replace('.', '').replace(',', '.'))


def _table_hashes(path):
    conn = sqlite3.connect(path)
    try:
        rows = conn.execute('SELECT owner, weight, staked, scanned_at '
                            'FROM voters ORDER BY owner').fetchall()
    finally:
        conn.close()
    return tuple(rows)


if __name__ == '__main__':
    main()