from datetime import UTC, datetime, timedelta import pytest from app import quotacalc # https://www.twilio.com/docs/sendgrid/ui/sending-email/warming-up-an-ip-address DAILY_WARMUP_QUOTAS = [ 20, 28, 39, 55, 77, 108, 151, 211, 295, 413, 579, 810, 1133, 1587, 2222, 3111, 4356, 6098, 8583, 11953, 16734, 23427, 32798, 45917, 64284, 89998, 125997, 176395, 246953, 345735, 484029, 677640, 948696, 1328175, 1859444, 2603222, 3644511, 5102316, 7143242, 10000539, 14000754, 19601056, ] DAILY_COOLDOWN_QUOTAS = [ 10000, 9963, 9945, 9918, 9878, 9820, 9734, 9608, 9426, 9168, 8807, 8320, 7685, 6899, 5986, 5000, 4013, 3100, 2314, 1679, 1192, 831, 573, 391, 265, 179, 121, 81, 54, 36, 20, 20, ] MAX_HOURLY_QUOTA = 1_800_000.0 START_TIME = datetime(year=2025, month=1, day=1, tzinfo=UTC) @pytest.mark.parametrize( ("warmup_days", "expected_quota"), [ pytest.param( warmup_days, expected_quota, id=f"warmup_days_{warmup_days}", ) for warmup_days, expected_quota in enumerate(DAILY_WARMUP_QUOTAS, start=0) ], ) def test_calculate_quota_with_warmup( warmup_days: float, expected_quota: float, ) -> None: """Test quota exponential growth during warmup.""" quota = quotacalc.calculate_quota_with_warmup( last_hourly_quota=quotacalc.MIN_QUOTA, warmup_days=warmup_days, ) assert quota == pytest.approx(expected_quota, rel=0.1) def test_calculate_quota_with_warmup_custom_growth_rate() -> None: """ Should increase quota faster when a higher custom growth_rate is provided, and slower when a lower rate is used. """ warmup_days = 5 base_quota = quotacalc.MIN_QUOTA # Default rate quota_default = quotacalc.calculate_quota_with_warmup( last_hourly_quota=base_quota, warmup_days=warmup_days, ) # Slower growth rate quota_slow = quotacalc.calculate_quota_with_warmup( last_hourly_quota=base_quota, warmup_days=warmup_days, growth_rate=quotacalc.GROWTH_RATE / 2, ) # Faster growth rate quota_fast = quotacalc.calculate_quota_with_warmup( last_hourly_quota=base_quota, warmup_days=warmup_days, growth_rate=quotacalc.GROWTH_RATE * 2, ) # The faster growth rate should yield a higher quota, # and the slower rate should yield a lower one. assert quota_slow < quota_default < quota_fast def test_calculate_quota_with_warmup_custom_rate_min_quota() -> None: """ Should return MIN_QUOTA when last_hourly_quota <= 0 even with custom growth_rate. """ quota = quotacalc.calculate_quota_with_warmup( last_hourly_quota=0, warmup_days=3, growth_rate=0.5, ) assert quota == quotacalc.MIN_QUOTA @pytest.mark.parametrize( ("cooldown_days", "expected_quota"), [ pytest.param( cooldown_days, expected_quota, id=f"cooldown_days_{cooldown_days}", ) for cooldown_days, expected_quota in enumerate(DAILY_COOLDOWN_QUOTAS, start=0) ], ) def test_calculate_quota_after_cooldown( cooldown_days: float, expected_quota: float, ) -> None: """Test quota exponential decay during cooldown.""" quota = quotacalc.calculate_quota_after_cooldown( last_hourly_quota=quotacalc.COOLDOWN_START_QUOTA, cooldown_days=cooldown_days, ) assert quota == pytest.approx(expected_quota, rel=0.1) @pytest.mark.parametrize( ( "last_sent_at", "last_hourly_quota", "current_time", "expected_current_hourly_quota", ), [ # Warm-up path (less than 5 minutes or no last send) pytest.param( None, None, START_TIME, DAILY_WARMUP_QUOTAS[0], id="warmup_initial", ), pytest.param( None, 1000, START_TIME, 1000, id="warmup_initial_with_last", ), pytest.param( START_TIME, quotacalc.MIN_QUOTA, START_TIME + timedelta(days=1), DAILY_WARMUP_QUOTAS[1], id="warmup_1day", ), pytest.param( START_TIME, DAILY_WARMUP_QUOTAS[5], START_TIME + timedelta(days=6), DAILY_WARMUP_QUOTAS[6], id="warmup_6day", ), pytest.param( START_TIME, DAILY_WARMUP_QUOTAS[10], START_TIME + timedelta(days=11), DAILY_WARMUP_QUOTAS[11], id="warmup_11day", ), pytest.param( START_TIME - timedelta(minutes=1), MAX_HOURLY_QUOTA - 0.0000001, START_TIME, MAX_HOURLY_QUOTA, id="warmup_max_quota", ), pytest.param( START_TIME - timedelta(minutes=1), MAX_HOURLY_QUOTA + 1000, START_TIME, MAX_HOURLY_QUOTA, id="warmup_max_quota_capped", ), # Cooldown path (> 5 minutes, i.e., days difference) pytest.param( START_TIME - timedelta(days=1), 10000.0, START_TIME, DAILY_COOLDOWN_QUOTAS[1], id="cooldown_1day", ), pytest.param( START_TIME - timedelta(days=10), 10000.0, START_TIME, DAILY_COOLDOWN_QUOTAS[10], id="cooldown_10days", ), pytest.param( START_TIME - timedelta(days=15), 10000.0, START_TIME, DAILY_COOLDOWN_QUOTAS[15], id="cooldown_midpoint", ), pytest.param( START_TIME - timedelta(days=30), 10000.0, START_TIME, DAILY_COOLDOWN_QUOTAS[30], id="cooldown_full_reset", ), ], ) def test_get_current_hourly_quota( last_sent_at: datetime | None, last_hourly_quota: float | None, current_time: datetime | None, expected_current_hourly_quota: float, ) -> None: """Test switching between warmup and cooldown quota paths.""" current_hourly_quota = quotacalc.calculate_current_hourly_quota( last_sent_at=last_sent_at, last_hourly_quota=last_hourly_quota, current_time=current_time, ) assert current_hourly_quota == pytest.approx(expected_current_hourly_quota, rel=0.1) @pytest.mark.parametrize( ( "last_sent_at", "last_hour_sent_emails", "current_hourly_quota", "current_time", "expected_available_quota", ), [ pytest.param( None, 0, 60, START_TIME.replace(minute=0), 1, id="initial_no_last_sent", ), pytest.param( None, 0, 20, None, 1, id="initial_no_last_sent_min_quota", ), pytest.param( START_TIME - timedelta(minutes=1), 0, 60, START_TIME, 1, id="recent_send_1min_ago", ), pytest.param( START_TIME - timedelta(minutes=2), 0, 60, START_TIME, 1, id="delay_under_max_sending_delay", ), pytest.param( START_TIME - timedelta(minutes=10), 0, 60, START_TIME + timedelta(minutes=2), 1, id="delay_exceeds_max_cap", ), pytest.param( START_TIME - timedelta(minutes=1), 2, 60, START_TIME, 0, id="already_sent_in_hour", ), ], ) def test_calculate_available_quota( last_sent_at: datetime | None, last_hour_sent_emails: int, current_hourly_quota: float, current_time: datetime | None, expected_available_quota: int, ) -> None: """Test available quota at any given minute.""" result = quotacalc.calculate_available_quota( last_sent_at=last_sent_at, last_hour_sent_emails=last_hour_sent_emails, current_hourly_quota=current_hourly_quota, current_time=current_time, ) assert result == expected_available_quota