102 lines
4.4 KiB
Python
102 lines
4.4 KiB
Python
"""Uji integrasi backend MariaDB/MySQL (opt-in, ⊥ dijalankan otomatis).
|
|
|
|
Oracle utama (test_get_voters.py / test_dashboard.py) tetap pakai sqlite agar
|
|
hermetik. Suite ini hanya jalan saat `VEX_DB_BACKEND=mysql` (dengan VEX_DB_*
|
|
yang menunjuk container docker-compose):
|
|
|
|
VEX_DB_BACKEND=mysql VEX_DB_HOST=127.0.0.1 VEX_DB_PORT=3306 \
|
|
VEX_DB_USER=databisnis VEX_DB_PASS=databisnis VEX_DB_NAME=databisnisid \
|
|
./venv/bin/python test_mariadb.py
|
|
|
|
Selain itu ia `skip` (exit 0 tanpa menjalankan apa pun). Menyalakan backend
|
|
mysql hanya lewat env — jangan pernah menyetel default sqlite ke mysql.
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import get_voters as gv
|
|
|
|
|
|
def check(tag, cond):
|
|
if not cond:
|
|
raise AssertionError(f'GAGAL: {tag}')
|
|
print(f'ok: {tag}')
|
|
|
|
|
|
def main():
|
|
if os.getenv('VEX_DB_BACKEND') != 'mysql':
|
|
print('skip: VEX_DB_BACKEND != mysql (backend sqlite default, tetap aman)')
|
|
return
|
|
|
|
import db
|
|
|
|
# ————— snapshot roundtrip: replace_snapshot → query —————
|
|
rows_a = [
|
|
{'owner': 'aaa1', 'weight': '111.00000000000000000', 'staked': 5000.0, 'last_vote': '2026-07-01T00:00:00'},
|
|
{'owner': 'ccc3', 'weight': '222.00000000000000000', 'staked': 3000.0, 'last_vote': '2026-07-02T00:00:00'},
|
|
]
|
|
scanned_a = gv.persist(rows_a)
|
|
got = db.query('SELECT owner, weight, staked, scanned_at, last_vote '
|
|
'FROM voters ORDER BY staked DESC')
|
|
check('mysql roundtrip: 2 baris sesuai urutan', [r[0] for r in got] == ['aaa1', 'ccc3'])
|
|
check('mysql roundtrip: weight verbatim TEXT', got[0][1] == '111.00000000000000000')
|
|
check('mysql roundtrip: staked float', isinstance(got[0][2], float) and got[0][2] == 5000.0)
|
|
check('mysql roundtrip: scanned_at terisi', got[0][3] == scanned_a)
|
|
check('mysql roundtrip: last_vote TEXT', got[0][4] == '2026-07-01T00:00:00')
|
|
|
|
# ————— snapshot ganti: baris lama hilang, tanpa PK clash —————
|
|
rows_b = [
|
|
{'owner': 'aaa1', 'weight': '999.00000000000000000', 'staked': 9000.0, 'last_vote': '2026-07-03T00:00:00'},
|
|
{'owner': 'ddd4', 'weight': '333.00000000000000000', 'staked': 2000.0, 'last_vote': '2026-07-04T00:00:00'},
|
|
]
|
|
gv.persist(rows_b)
|
|
got = db.query('SELECT owner, weight, staked, last_vote FROM voters ORDER BY owner')
|
|
by = {r[0]: r for r in got}
|
|
check('mysql ganti snapshot: |set_b|', len(got) == 2)
|
|
check('mysql ganti snapshot: ccc3 dihapus', 'ccc3' not in by)
|
|
check('mysql ganti snapshot: ddd4 baru', 'ddd4' in by)
|
|
check('mysql ganti snapshot: bobot terbaru', by['aaa1'][1] == '999.00000000000000000')
|
|
|
|
# ————— placeholder `%s` (LIKE) & LIMIT/OFFSET diterjemahkan —————
|
|
like = db.query(
|
|
"SELECT owner FROM voters WHERE owner LIKE %s ESCAPE '!' "
|
|
'ORDER BY owner LIMIT %s OFFSET %s',
|
|
['%aa%', 10, 0],
|
|
)
|
|
check('mysql LIKE %s + LIMIT %s', [r[0] for r in like] == ['aaa1'])
|
|
like_none = db.query(
|
|
"SELECT owner FROM voters WHERE owner LIKE %s ESCAPE '!'",
|
|
['%zzz%'],
|
|
)
|
|
check('mysql LIKE tak cocok → kosong', like_none == [])
|
|
|
|
# ————— queryone & COALESCE/agregat —————
|
|
cnt = db.queryone('SELECT COUNT(*) FROM voters')[0]
|
|
check('mysql COUNT(*)', cnt == 2)
|
|
scanned = db.queryone("SELECT COALESCE(MAX(scanned_at), '') FROM voters")[0]
|
|
check('mysql MAX(scanned_at) non-empty', isinstance(scanned, str) and scanned != '')
|
|
|
|
# ————— V41: voter_first_seen — INSERT IGNORE idempoten (⊥ overwrite) —————
|
|
db.record_first_seen([{'owner': 'aaa1'}, {'owner': 'ddd4'}], '2026-08-01T00:00:00')
|
|
db.record_first_seen([{'owner': 'aaa1'}], '2026-08-02T00:00:00')
|
|
fs = {r[0]: r[1] for r in db.query(
|
|
'SELECT owner, first_seen_at FROM voter_first_seen ORDER BY owner')}
|
|
check('mysql first_seen idempoten',
|
|
fs == {'aaa1': '2026-08-01T00:00:00', 'ddd4': '2026-08-01T00:00:00'})
|
|
|
|
# ————— rank via correlated-count (global, sama semantik sqlite) —————
|
|
ranked = db.query(
|
|
'SELECT v.owner, (SELECT COUNT(*) FROM voters v2 '
|
|
'WHERE v2.staked > v.staked '
|
|
' OR (v2.staked = v.staked AND v2.owner <= v.owner)) AS rn '
|
|
'FROM voters v ORDER BY v.staked DESC, v.owner ASC'
|
|
)
|
|
check('mysql rank global', [r[1] for r in ranked] == [1, 2])
|
|
|
|
print('selesai: semua uji MariaDB lolos')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|