Files
databisnisid/test_distribute_loop.py

157 lines
6.4 KiB
Python

"""Oracle jadwal distribusi (V37/V44): next_boundary + parser.
Murni logika kalender — ⊥ jaringan/DB. Verifikasi: first_run one-off diambil
bila lebih dekat dari mingguan; batas dihitung ulang tiap iterasi loop
(distribute_loop.main loop tak pernah keluar); first_run yang sudah lewat
dibuang → hanya jadwal mingguan; V44: jadwal multi-hari (CSV `wed,sat`) — event
terdekat di antara hari-hari yang terdaftar, tetap hormati gating first_run.
"""
from datetime import datetime
import distribute_loop as dl
CHECKS = 0
def check(tag, cond):
global CHECKS
CHECKS += 1
if not cond:
raise AssertionError(f'GAGAL: {tag}')
print(f'ok: {tag}')
def main():
# parse_weekday: nama + angka, kasus-tak-acak
check('parse_weekday sat', dl.parse_weekday('sat') == 5)
check('parse_weekday SUN upper', dl.parse_weekday('SUN') == 6)
check('parse_weekday mon', dl.parse_weekday('mon') == 0)
check('parse_weekday angka 0-6', dl.parse_weekday('3') == 3)
check('parse_weekday 6', dl.parse_weekday('6') == 6)
for bad in ('', 'xyz', '7', '-1'):
try:
dl.parse_weekday(bad)
raise AssertionError(f'GAGAL: parse_weekday {bad!r} harus raise')
except ValueError:
pass
check('parse_weekday invalid raise', True)
# parse_weekdays (V44): satu hari atau CSV → list int unik terurut
check('parse_weekdays sat', dl.parse_weekdays('sat') == [5])
check('parse_weekdays wed,sat', dl.parse_weekdays('wed,sat') == [2, 5])
check('parse_weekdays kebalikan', dl.parse_weekdays('sat,wed') == [2, 5])
check('parse_weekdays duplikat', dl.parse_weekdays('wed,sat,wed') == [2, 5])
check('parse_weekdays angka CSV', dl.parse_weekdays('2,5') == [2, 5])
try:
dl.parse_weekdays('')
raise AssertionError('GAGAL: parse_weekdays kosong harus raise')
except ValueError:
pass
try:
dl.parse_weekdays('xyz')
raise AssertionError('GAGAL: parse_weekdays tak valid harus raise')
except ValueError:
pass
check('parse_weekdays invalid raise', True)
# parse_first_run: kosong → None, format → date
check('parse_first_run kosong', dl.parse_first_run('') is None)
check('parse_first_run None', dl.parse_first_run(None) is None)
fr = dl.parse_first_run('2026-08-17')
check('parse_first_run format', fr.year == 2026 and fr.month == 8
and fr.day == 17)
try:
dl.parse_first_run('17/08/2026')
raise AssertionError('GAGAL: format salah harus raise')
except ValueError:
pass
check('parse_first_run format salah raise', True)
# — next_boundary: mingguan Sabtu jam 10 + one-off Senin 17 Agt —
# Awal (Kamis 6 Agt pagi): Sabtu 8 Agt jatuh SEBELUM launch 17 Agt → dilewati,
# event terdekat = one-off Senin 17 Agt
now = datetime(2026, 8, 6, 9, 0) # Kamis
t = dl.next_boundary(now, 10, 5, fr)
check('Sabtu sebelum launch dilewati → Senin 17 Agt',
t == datetime(2026, 8, 17, 10, 0))
# Sama jam sudah lewat di hari one-off → mingguan Sabtu berikutnya
now = datetime(2026, 8, 17, 11, 0) # Senin, jam lewat
t = dl.next_boundary(now, 10, 5, fr)
check('first_run lewat → Sabtu 22 Agt',
t == datetime(2026, 8, 22, 10, 0))
# Mingguan saja (⊥ first_run) → Sabtu terdekat dari sekarang
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, 5)
check('tanpa first_run → Sabtu 8 Agt',
t == datetime(2026, 8, 8, 10, 0))
# Hari yang sama, jam belum lewat → hari ini
t = dl.next_boundary(datetime(2026, 8, 22, 9, 0), 10, 5)
check('Sabtu jam belum lewat → hari ini',
t == datetime(2026, 8, 22, 10, 0))
# Hari yang sama, jam sudah lewat → minggu depan
t = dl.next_boundary(datetime(2026, 8, 22, 11, 0), 10, 5)
check('Sabtu jam lewat → Sabtu pekan depan',
t == datetime(2026, 8, 29, 10, 0))
# first_run = hari Sabtu (22 Agt) → Sabtu2 sebelum launch dilewati
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, 5,
dl.parse_first_run('2026-08-22'))
check('first_run Sabtu → Sabtu 8/15 Agt dilewati, event = 22 Agt',
t == datetime(2026, 8, 22, 10, 0))
# Satu hari sebelum launch (Sabtu 15 Agt pagi) → tetap 22 Agt
t = dl.next_boundary(datetime(2026, 8, 15, 9, 0), 10, 5,
dl.parse_first_run('2026-08-22'))
check('sehari sblm launch Sabtu → tetap 22 Agt',
t == datetime(2026, 8, 22, 10, 0))
# Hari lain dalam seminggu: Rabu (weekday 2) → next Rabu 10:00
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, 2)
check('Rabu → Rabu 12 Agt', t == datetime(2026, 8, 12, 10, 0))
# — V44: jadwal multi-hari Rabu & Sabtu ([2, 5]) —
# Sebelum launch: Rabu/Sabtu sebelum 17 Agt dilewati → one-off Senin 17 Agt
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, [2, 5], fr)
check('wed,sat sblm launch → one-off 17 Agt',
t == datetime(2026, 8, 17, 10, 0))
# Setelah launch (Senin 17 Agt siang) → Rabu 19 Agt (bukan Sabtu 22)
t = dl.next_boundary(datetime(2026, 8, 17, 11, 0), 10, [2, 5], fr)
check('wed,sat stlh launch → Rabu 19 Agt',
t == datetime(2026, 8, 19, 10, 0))
# Setelah Rabu 19 Agt siang → Sabtu 22 Agt (jeda terdekat berikutnya)
t = dl.next_boundary(datetime(2026, 8, 19, 11, 0), 10, [2, 5], fr)
check('wed,sat stlh Rabu lewat → Sabtu 22 Agt',
t == datetime(2026, 8, 22, 10, 0))
# Tanpa first_run: dari Kamis 6 Agt → Sabtu 8 Agt (lebih dekat dari Rabu 12)
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, [2, 5])
check('wed,sat tanpa fr → Sabtu 8 Agt',
t == datetime(2026, 8, 8, 10, 0))
# Rabu pagi sebelum jam → Rabu hari itu juga
t = dl.next_boundary(datetime(2026, 8, 12, 9, 0), 10, [2, 5])
check('wed,sat Rabu jam sblm → hari ini',
t == datetime(2026, 8, 12, 10, 0))
# Rabu siang setelah jam → Sabtu 15 Agt
t = dl.next_boundary(datetime(2026, 8, 12, 11, 0), 10, [2, 5])
check('wed,sat Rabu jam lewat → Sabtu 15 Agt',
t == datetime(2026, 8, 15, 10, 0))
# Skalar int tetap kompatibel (normalisasi ke list)
t = dl.next_boundary(datetime(2026, 8, 6, 9, 0), 10, 5)
check('skalar int kompatibel → Sabtu 8 Agt',
t == datetime(2026, 8, 8, 10, 0))
print(f'SEMUA UJI PASS ({CHECKS} checks)')
if __name__ == '__main__':
main()