from datetime import date from lyra_worker.qobuz_gate import ( QobuzPacing, today_cap, is_allowed, next_gap_seconds, gate_open, record_download, ) # ---- QobuzPacing.from_config ------------------------------------------------- def test_from_config_defaults(): p = QobuzPacing.from_config({}) assert p.enabled is True assert p.active_start == 8 and p.active_end == 23 assert p.daily_cap == 40 assert p.warmup_start_date == "" def test_from_config_reads_values(): p = QobuzPacing.from_config({ "qobuz.pacing.enabled": "false", "qobuz.pacing.activeStartHour": "9", "qobuz.pacing.activeEndHour": "22", "qobuz.pacing.dailyCap": "25", "qobuz.pacing.warmupStartDate": "2026-07-15", }) assert p.enabled is False assert p.active_start == 9 and p.active_end == 22 assert p.daily_cap == 25 assert p.warmup_start_date == "2026-07-15" # ---- today_cap (warm-up ramp) ------------------------------------------------ def test_today_cap_no_warmup_uses_steady(): p = QobuzPacing(daily_cap=40, warmup_start_date="") assert today_cap(p, date(2026, 7, 20)) == 40 def test_today_cap_ramp_boundaries(): p = QobuzPacing(daily_cap=40, warmup_start_date="2026-07-15") assert today_cap(p, date(2026, 7, 15)) == 5 # day 0 assert today_cap(p, date(2026, 7, 17)) == 5 # day 2 assert today_cap(p, date(2026, 7, 18)) == 10 # day 3 assert today_cap(p, date(2026, 7, 21)) == 10 # day 6 assert today_cap(p, date(2026, 7, 22)) == 20 # day 7 assert today_cap(p, date(2026, 7, 28)) == 20 # day 13 assert today_cap(p, date(2026, 7, 29)) == 40 # day 14 assert today_cap(p, date(2026, 8, 30)) == 40 # long after def test_today_cap_steady_below_ramp_step_is_the_ceiling(): # A small steady cap is never exceeded by a ramp step. p = QobuzPacing(daily_cap=3, warmup_start_date="2026-07-15") assert today_cap(p, date(2026, 7, 29)) == 3 # ---- is_allowed (the gate decision) ----------------------------------------- _P = QobuzPacing(enabled=True, active_start=8, active_end=23, daily_cap=40) def test_disabled_always_allows(): p = QobuzPacing(enabled=False, active_start=8, active_end=23) assert is_allowed(p, now_hour=3, count_today=999, cap=0, now_epoch=0, next_allowed_epoch=1e18) is True def test_blocked_outside_active_hours(): assert is_allowed(_P, now_hour=7, count_today=0, cap=40, now_epoch=100, next_allowed_epoch=0) is False assert is_allowed(_P, now_hour=23, count_today=0, cap=40, now_epoch=100, next_allowed_epoch=0) is False def test_blocked_at_or_over_cap(): assert is_allowed(_P, now_hour=10, count_today=40, cap=40, now_epoch=100, next_allowed_epoch=0) is False assert is_allowed(_P, now_hour=10, count_today=41, cap=40, now_epoch=100, next_allowed_epoch=0) is False def test_blocked_before_next_allowed(): assert is_allowed(_P, now_hour=10, count_today=0, cap=40, now_epoch=100, next_allowed_epoch=200) is False def test_allowed_when_all_conditions_met(): assert is_allowed(_P, now_hour=10, count_today=5, cap=40, now_epoch=300, next_allowed_epoch=200) is True # ---- next_gap_seconds (spread + jitter) ------------------------------------- def test_next_gap_spreads_cap_across_active_window(): # 15h window (8..23) / cap 40 = 1350s base; ±30%; midpoint rand -> base. p = QobuzPacing(active_start=8, active_end=23, daily_cap=40) assert abs(next_gap_seconds(p, cap=40, rand01=0.5) - 1350.0) < 1.0 assert abs(next_gap_seconds(p, cap=40, rand01=0.0) - 1350.0 * 0.7) < 1.0 # low end assert abs(next_gap_seconds(p, cap=40, rand01=1.0) - 1350.0 * 1.3) < 1.0 # high end def test_next_gap_has_a_floor(): # A huge cap would give a tiny gap; a 60s floor protects the request rate. p = QobuzPacing(active_start=8, active_end=23, daily_cap=100000) assert next_gap_seconds(p, cap=100000, rand01=0.0) == 60.0 # ---- DB helpers: gate_open / record_download -------------------------------- def _set(conn, key, value): with conn.cursor() as cur: cur.execute( 'INSERT INTO "Config"(key,value,secret,"updatedAt") VALUES (%s,%s,false,now()) ' 'ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, "updatedAt"=now()', (key, value), ) conn.commit() def _get(conn, key): with conn.cursor() as cur: cur.execute('SELECT value FROM "Config" WHERE key=%s', (key,)) row = cur.fetchone() return row[0] if row else None def test_record_download_increments_and_sets_next_allowed(conn): _set(conn, "qobuz.pacing.dailyCap", "40") record_download(conn, rand01=0.5) assert _get(conn, "qobuz.pacing.countToday") == "1" assert _get(conn, "qobuz.pacing.countDay") is not None assert _get(conn, "qobuz.pacing.nextAllowedAt") is not None record_download(conn, rand01=0.5) assert _get(conn, "qobuz.pacing.countToday") == "2" def test_record_download_resets_counter_on_new_day(conn): # A stale count from a previous day must reset to 1, not accumulate. _set(conn, "qobuz.pacing.dailyCap", "40") _set(conn, "qobuz.pacing.countDay", "2000-01-01") _set(conn, "qobuz.pacing.countToday", "39") record_download(conn, rand01=0.5) assert _get(conn, "qobuz.pacing.countToday") == "1" def test_gate_open_with_always_open_window(conn): # An all-day window (0..24) removes the time-of-day dependency: no prior downloads and no # next-allowed set → the gate is open regardless of when the test runs. _set(conn, "qobuz.pacing.activeStartHour", "0") _set(conn, "qobuz.pacing.activeEndHour", "24") assert gate_open(conn) is True def test_gate_closed_by_empty_active_window(conn): # start == end is an empty window (h >= 5 AND h < 5 is never true) → always closed. _set(conn, "qobuz.pacing.activeStartHour", "5") _set(conn, "qobuz.pacing.activeEndHour", "5") assert gate_open(conn) is False def test_gate_closed_when_cap_reached_today(conn): _set(conn, "qobuz.pacing.activeStartHour", "0") _set(conn, "qobuz.pacing.activeEndHour", "24") _set(conn, "qobuz.pacing.dailyCap", "3") # Simulate 3 downloads already today. with conn.cursor() as cur: cur.execute("SELECT current_date::text") today = cur.fetchone()[0] _set(conn, "qobuz.pacing.countDay", today) _set(conn, "qobuz.pacing.countToday", "3") assert gate_open(conn) is False def test_gate_ignores_stale_count_from_a_previous_day(conn): # A high count from yesterday must not keep the gate closed today. _set(conn, "qobuz.pacing.activeStartHour", "0") _set(conn, "qobuz.pacing.activeEndHour", "24") _set(conn, "qobuz.pacing.dailyCap", "3") _set(conn, "qobuz.pacing.countDay", "2000-01-01") _set(conn, "qobuz.pacing.countToday", "99") assert gate_open(conn) is True