from datetime import timedelta import freezegun import pytest from dirty_equals import IsList from fansifter_common.utils import timezone from app.models import Campaign, CampaignBatch from app.strategies import FcfsStrategy, QuotaAllocation, QuotaStrategyType from tests.unit.types import BuildModel class TestFcfsStrategy: """Tests for FcfsStrategy allocation logic.""" @pytest.fixture(scope="class") def strategy(self) -> FcfsStrategy: return FcfsStrategy() def test_strategy_type(self, strategy: FcfsStrategy) -> None: """Verify strategy type identifier""" assert strategy.strategy_type == QuotaStrategyType.FCFS @freezegun.freeze_time("2025-11-20 10:30:00") def test_fcfs_order_earliest_campaign_first( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test that earliest scheduled campaign gets all quota""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=30), send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=90), send_at=current_time - timedelta(minutes=90), created_at=current_time - timedelta(minutes=90), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 campaign_3 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=60), send_at=current_time - timedelta(minutes=60), created_at=current_time - timedelta(minutes=60), ) batch_3 = build_model( CampaignBatch, campaign_id=campaign_3.id, batch_size=1000, batch_offset=0, ) batch_3.campaign = campaign_3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=campaign_2.id, batch_id=batch_2.id, emails_to_send=100, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_send_at_constraint( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test that send_at acts as blocking constraint""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=90), send_at=current_time + timedelta(minutes=30), created_at=current_time - timedelta(minutes=90), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=30), send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 allocations = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=campaign_2.id, batch_id=batch_2.id, emails_to_send=100, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_preemption_scenario( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test time-boundary preemption scenario""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time, send_at=current_time, created_at=current_time, ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=10000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=5), send_at=current_time + timedelta(minutes=5), created_at=current_time - timedelta(minutes=5), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=5000, batch_offset=0, ) batch_2.campaign = campaign_2 allocations_before = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=10, current_time=current_time, ) assert allocations_before == IsList( QuotaAllocation( campaign_id=campaign_1.id, batch_id=batch_1.id, emails_to_send=10, ) ) time_after = current_time + timedelta(minutes=5) allocations_after = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=10, current_time=time_after, ) assert allocations_after == IsList( QuotaAllocation( campaign_id=campaign_2.id, batch_id=batch_2.id, emails_to_send=10, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_limited_by_remaining_recipients( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test allocation limited by batch remaining recipients""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=90), send_at=current_time - timedelta(minutes=90), created_at=current_time - timedelta(minutes=90), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=100, batch_offset=95, ) batch_1.campaign = campaign_1 allocations = strategy.allocate_quota( batches=[batch_1], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=5, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_no_eligible_campaigns( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test when all campaigns blocked by send_at""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=90), send_at=current_time + timedelta(minutes=90), created_at=current_time - timedelta(minutes=90), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 allocations = strategy.allocate_quota( batches=[batch_1], available_quota=100, current_time=current_time, ) assert allocations == [] @freezegun.freeze_time("2025-11-20 10:30:00") def test_tiebreaker_by_created_at( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test tiebreaker when scheduled_at identical""" current_time = timezone.now() same_scheduled_at = current_time - timedelta(minutes=90) campaign_1 = build_model( Campaign, scheduled_at=same_scheduled_at, send_at=same_scheduled_at, created_at=current_time - timedelta(minutes=120), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=same_scheduled_at, send_at=same_scheduled_at, created_at=current_time - timedelta(minutes=105), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 allocations = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=100, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_legacy_campaigns_highest_priority( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test that campaigns without scheduled_at (legacy) get highest priority""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(hours=2, minutes=30), send_at=current_time - timedelta(hours=2, minutes=30), created_at=current_time - timedelta(hours=2, minutes=30), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=None, send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 campaign_3 = build_model( Campaign, scheduled_at=None, send_at=current_time - timedelta(minutes=90), created_at=current_time - timedelta(minutes=90), ) batch_3 = build_model( CampaignBatch, campaign_id=campaign_3.id, batch_size=1000, batch_offset=0, ) batch_3.campaign = campaign_3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=100, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_spillover_to_next_campaign( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test quota spillover when priority campaign uses only part of quota""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=35), send_at=current_time - timedelta(minutes=35), created_at=current_time - timedelta(minutes=35), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=100, batch_offset=99, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time + timedelta(minutes=25), send_at=current_time - timedelta(minutes=5), created_at=current_time + timedelta(minutes=25), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 allocations = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=1, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=99, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_spillover_through_multiple_campaigns( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test quota spillover cascades through multiple campaigns""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=30), send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=10, batch_offset=5, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=20), send_at=current_time - timedelta(minutes=20), created_at=current_time - timedelta(minutes=20), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=20, batch_offset=15, ) batch_2.campaign = campaign_2 campaign_3 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=10), send_at=current_time - timedelta(minutes=10), created_at=current_time - timedelta(minutes=10), ) batch_3 = build_model( CampaignBatch, campaign_id=campaign_3.id, batch_size=1000, batch_offset=0, ) batch_3.campaign = campaign_3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=5, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=5, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=90, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_no_spillover_when_first_uses_all_quota( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test no spillover when priority campaign uses entire quota""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=30), send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=1000, batch_offset=0, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time + timedelta(minutes=30), send_at=current_time, created_at=current_time + timedelta(minutes=30), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 allocations = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=100, ) ) @freezegun.freeze_time("2025-11-20 10:30:00") def test_spillover_respects_send_at_constraint( self, strategy: FcfsStrategy, build_model: BuildModel ) -> None: """Test spillover skips campaigns blocked by send_at""" current_time = timezone.now() campaign_1 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=30), send_at=current_time - timedelta(minutes=30), created_at=current_time - timedelta(minutes=30), ) batch_1 = build_model( CampaignBatch, campaign_id=campaign_1.id, batch_size=10, batch_offset=5, ) batch_1.campaign = campaign_1 campaign_2 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=15), send_at=current_time + timedelta(minutes=30), created_at=current_time - timedelta(minutes=15), ) batch_2 = build_model( CampaignBatch, campaign_id=campaign_2.id, batch_size=1000, batch_offset=0, ) batch_2.campaign = campaign_2 campaign_3 = build_model( Campaign, scheduled_at=current_time - timedelta(minutes=10), send_at=current_time - timedelta(minutes=10), created_at=current_time - timedelta(minutes=10), ) batch_3 = build_model( CampaignBatch, campaign_id=campaign_3.id, batch_size=1000, batch_offset=0, ) batch_3.campaign = campaign_3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=5, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=95, ), check_order=False, )