dsutanto/databisnisid
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

klaim reward BP harian (claim service V32-V34) + kill-switch DISTRIBUTE_ENABLED (V35) + Telegram dua kanal komunitas/internal (V30/V36) + placement web/scan/dist/claim != server2U

Browse Source
This commit is contained in:
proitlab committed 2026-08-06 06:30:19 +07:00
1 parent d3a0e768a0
commit b1e6e93796
14 files changed
+878 -72
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

test_distribute.py
+66 -17
View File
@@ -71,10 +71,16 @@ def main():
# Oracle: Telegram ⊥ boleh mengirim nyata — nonaktifkan dulu, dipakai lagi §V30
_tg_token = telegram.TELEGRAM_BOT_TOKEN
_tg_chats = telegram.TELEGRAM_CHAT_IDS
_tg_comm = telegram.TELEGRAM_COMMUNITY_CHAT_IDS
_tg_int = telegram.TELEGRAM_INTERNAL_CHAT_IDS
_tg_post = telegram.requests.post
telegram.TELEGRAM_BOT_TOKEN = ''
telegram.TELEGRAM_CHAT_IDS = []
telegram.TELEGRAM_COMMUNITY_CHAT_IDS = []
telegram.TELEGRAM_INTERNAL_CHAT_IDS = []
# V35: kill-switch default OFF (payout ⊥ jalan tanpa diset true)
check('V35 default config off', dist.DISTRIBUTE_ENABLED is False)
dist.DISTRIBUTE_ENABLED = True # run-nyata di bawah pakai enabled
# ————— V20: fetch_balance — parse chain get_currency_balance —————
dist.requests.post = lambda url, json=None, timeout=30: fake_post(
@@ -223,20 +229,53 @@ def main():
check('V22 run status partial',
calls['run_updates'][0][1] == 'partial')
# ————— V30: notifikasi Telegram — best-effort, ⊥ dry-run/no-op —————
# ————— V35: kill-switch DISTRIBUTE_ENABLED=false → no-op —————
make_voters_db(db.DB_PATH, [('aaa1', '1.0', 1200.0, '2026-08-05T00:00:00',
'2026-08-05T00:00:00')])
db.query = _orig_query
dist.fetch_balance = lambda: Decimal('100.0000')
_orig_record_run = db.record_run
_orig_build = dist.build_signed_transfer
dist.DISTRIBUTE_ENABLED = False
calls['runs'] = []
calls['signs'] = []
db.record_run = lambda *a, **k: calls['runs'].append(a) or 99
dist.build_signed_transfer = (lambda to, amount, memo:
calls['signs'].append(memo)
or FakeSigned('txoff'))
buf = io.StringIO()
with redirect_stdout(buf):
rc = dist.run_distribution(dry_run=False)
out = buf.getvalue()
check('V35 disabled → exit 0', rc == 0)
check('V35 disabled → pesan dinonaktifkan', 'dinonaktifkan' in out)
check('V35 disabled → ⊥ tulis DB', calls['runs'] == [])
check('V35 disabled → ⊥ tanda tangan', calls['signs'] == [])
buf = io.StringIO()
with redirect_stdout(buf):
rc = dist.run_distribution(dry_run=True)
check('V35 disabled + dry-run → rencana tetap',
rc == 0 and 'RENCANA' in buf.getvalue())
dist.DISTRIBUTE_ENABLED = True
db.record_run = _orig_record_run
dist.build_signed_transfer = _orig_build
# ————— V30: notifikasi Telegram — dua kanal, komunitas brief/internal detail —————
tg = {'posts': []}
def tg_post(url, json=None, timeout=None):
tg['posts'].append((url, json['text']))
tg['posts'].append((url, json['chat_id'], json['text']))
return fake_post({})
telegram.TELEGRAM_BOT_TOKEN = '123:TEST'
telegram.TELEGRAM_CHAT_IDS = ['-100123']
telegram.TELEGRAM_COMMUNITY_CHAT_IDS = ['-1001']
telegram.TELEGRAM_INTERNAL_CHAT_IDS = ['-1002']
telegram.requests.post = tg_post
# send_text tanpa token → no-op senyap (⊥ post)
telegram.TELEGRAM_BOT_TOKEN = ''
telegram.send_text('tidak boleh terkirim')
telegram.send_internal('tidak boleh terkirim')
check('V30 ⊥ token → no-op senyap', tg['posts'] == [])
telegram.TELEGRAM_BOT_TOKEN = '123:TEST'
@@ -256,39 +295,49 @@ def main():
dist.run_distribution(dry_run=True)
check('V30 dry-run ⊥ notif', tg['posts'] == [])
# run nyata → notif mulai + selesai
# run nyata → komunitas 2 brief + internal 2 detail
calls['updates'] = []
calls['run_updates'] = []
with redirect_stdout(io.StringIO()):
rc = dist.run_distribution(dry_run=False)
check('V30 run nyata exit 0', rc == 0)
check('V30 2 notif (mulai + selesai)',
len(tg['posts']) == 2)
check('V30 mulai tercatat', 'DISTRIBUSI MULAI' in tg['posts'][0][1])
check('V30 selesai + status ok', 'DISTRIBUSI SELESAI' in tg['posts'][1][1]
and 'Status: ok' in tg['posts'][1][1])
check('V30 ringkasan jumlah', '2/2 pemilih' in tg['posts'][1][1])
comm = [p for p in tg['posts'] if p[1] == '-1001']
intern = [p for p in tg['posts'] if p[1] == '-1002']
check('V30 komunitas 2 notif brief', len(comm) == 2)
check('V30 komunitas brief MULAI', 'DISTRIBUSI MULAI' in comm[0][2]
and 'Pemilih' not in comm[0][2])
check('V30 komunitas brief SELESAI + status', 'DISTRIBUSI SELESAI'
in comm[1][2] and 'Status: ok' in comm[1][2]
and 'Terkirim' not in comm[1][2])
check('V30 internal 2 notif detail', len(intern) == 2)
check('V30 internal detail MULAI', 'DISTRIBUSI MULAI' in intern[0][2]
and 'Pemilih: 2' in intern[0][2])
check('V30 internal detail SELESAI', '2/2 pemilih' in intern[1][2]
and 'Gagal: 0' in intern[1][2])
# no-op (pemilih kosong) → ⊥ notif baru
db.query = lambda sql, params=(): []
with redirect_stdout(io.StringIO()):
dist.run_distribution(dry_run=False)
check('V30 no-op ⊥ notif', len(tg['posts']) == 2)
check('V30 no-op ⊥ notif', len(tg['posts']) == 4)
db.query = _orig_query
# failure (⊥ key) → notif gagal lewat main(), lalu re-raise
# failure (⊥ key) → notif gagal ke INTERNAL saja (ops detail)
dist.BP_PRIVATE_KEY = ''
try:
dist.main()
raise AssertionError('V30 gagal: main harusnya re-raise')
except RuntimeError:
check('V30 failure → re-raise RuntimeError', True)
check('V30 notif gagal terkirim',
len(tg['posts']) == 3 and 'DISTRIBUSI GAGAL' in tg['posts'][2][1])
new_posts = tg['posts'][4:]
check('V30 gagal → internal saja',
len(new_posts) == 1 and new_posts[0][1] == '-1002'
and 'DISTRIBUSI GAGAL' in new_posts[0][2])
dist.BP_PRIVATE_KEY = '5Ktest'
telegram.TELEGRAM_BOT_TOKEN = _tg_token
telegram.TELEGRAM_CHAT_IDS = _tg_chats
telegram.TELEGRAM_COMMUNITY_CHAT_IDS = _tg_comm
telegram.TELEGRAM_INTERNAL_CHAT_IDS = _tg_int
telegram.requests.post = _tg_post
print('\nSEMUA UJI PASS')