from datetime import datetime, timedelta import freezegun import pytest from dirty_equals import IsList, IsPartialDataclass from fansifter_common.utils import timezone from app.models import CampaignBatch from app.strategies import FairDistributionStrategy, QuotaAllocation, QuotaStrategyType from tests.unit.types import BuildModel class TestFairDistributionStrategy: """Tests for FairDistributionStrategy allocation logic.""" @pytest.fixture(scope="class") def strategy(self) -> FairDistributionStrategy: return FairDistributionStrategy() def test_strategy_type(self, strategy: FairDistributionStrategy) -> None: """Verify strategy type identifier""" assert strategy.strategy_type == QuotaStrategyType.FAIR_DISTRIBUTION @freezegun.freeze_time("2025-11-20 10:15:00") def test_even_split_three_batches( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test quota split evenly among 3 batches""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=5), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=3), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=1), ) batch_3.id = 3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=99, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=33, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=33, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=33, ), ) @freezegun.freeze_time("2025-11-20 10:15:00") def test_limited_by_remaining_recipients( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test allocation limited by batch remaining recipients""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=100, batch_offset=90, updated_at=current_time, ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time, ) batch_2.id = 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=10, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=90, ), check_order=False, ) def test_rotation_by_minute( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test that batch order rotates based on minute""" base_time = datetime(2025, 11, 20, 10, 0, tzinfo=timezone.UTC) batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=10), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=5), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=1), ) batch_3.id = 3 allocations_min0 = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=base_time.replace(minute=0), ) assert allocations_min0 == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=34, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=33, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=33, ), check_order=False, ) allocations_min1 = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=base_time.replace(minute=1), ) assert allocations_min1 == IsList( QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=34, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=33, ), QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=33, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:15:00") def test_no_batches(self, strategy: FairDistributionStrategy) -> None: """Test with no batches returns empty list""" current_time = timezone.now() allocations = strategy.allocate_quota( batches=[], available_quota=100, current_time=current_time, ) assert allocations == [] @freezegun.freeze_time("2025-11-20 10:15:00") def test_zero_quota( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test with zero quota returns empty list""" current_time = timezone.now() batches = [ build_model( CampaignBatch, campaign_id="camp_a", batch_size=1000, batch_offset=0, updated_at=current_time, ) ] allocations = strategy.allocate_quota( batches=batches, available_quota=0, current_time=current_time, ) assert allocations == [] @freezegun.freeze_time("2025-11-20 10:15:00") def test_single_batch_no_rotation( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test that single batch returns same list (no rotation needed)""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time, ) batch_1.id = 1 batches = [batch_1] allocations = strategy.allocate_quota( batches=batches, available_quota=100, current_time=current_time, ) assert allocations == [ QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=100 ) ] @freezegun.freeze_time("2025-11-20 12:03:00") def test_rotation_remainder_zero( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test rotation when minute % batch_count = 0 (no rotation, sorted order)""" base_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=10), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=5), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=1), ) batch_3.id = 3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=90, current_time=base_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=30, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 12:04:00") def test_rotation_by_one_position( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test rotation when minute % batch_count = 1 (rotate by 1)""" base_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=10), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=5), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=1), ) batch_3.id = 3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=90, current_time=base_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=30, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 12:05:00") def test_rotation_by_two_positions( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test rotation when minute % batch_count = 2 (rotate by 2)""" base_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=10), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=5), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=base_time - timedelta(minutes=1), ) batch_3.id = 3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=90, current_time=base_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=30, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=30, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:15:00") def test_redistribution_with_multiple_capped_batches( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test redistribution when multiple batches are capped""" current_time = timezone.now() # Batches are sorted by updated_at, so order is stable batch_1 = build_model( CampaignBatch, batch_size=100, batch_offset=90, # 10 remaining updated_at=current_time - timedelta(minutes=3), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=100, batch_offset=80, # 20 remaining updated_at=current_time - timedelta(minutes=2), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, # 1000 remaining updated_at=current_time - timedelta(minutes=1), ) batch_3.id = 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=10, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=20, ), QuotaAllocation( campaign_id=batch_3.campaign_id, batch_id=batch_3.id, emails_to_send=70, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:15:00") def test_even_distribution_with_full_capacity( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test even distribution when all batches have full capacity.""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=3), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=2), ) batch_2.id = 2 batch_3 = build_model( CampaignBatch, batch_size=1000, batch_offset=0, updated_at=current_time - timedelta(minutes=1), ) batch_3.id = 3 allocations = strategy.allocate_quota( batches=[batch_1, batch_2, batch_3], available_quota=100, current_time=current_time, ) assert allocations == IsList( IsPartialDataclass(emails_to_send=34), IsPartialDataclass(emails_to_send=33), IsPartialDataclass(emails_to_send=33), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:00:00") def test_capacity_equals_quota_share_redistributes_remainder( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test that when a batch's capacity exactly equals its quota share, leftover quota is redistributed to batches with remaining capacity.""" current_time = timezone.now() # minute=0, 3 batches → rotation remainder=0, sorted order: A, B, C batch_a = build_model( CampaignBatch, batch_size=2, batch_offset=0, # capacity 2 updated_at=current_time - timedelta(minutes=3), ) batch_a.id = 1 batch_b = build_model( CampaignBatch, batch_size=3, batch_offset=0, # capacity 3 updated_at=current_time - timedelta(minutes=2), ) batch_b.id = 2 batch_c = build_model( CampaignBatch, batch_size=1000, batch_offset=0, # capacity 1000 updated_at=current_time - timedelta(minutes=1), ) batch_c.id = 3 # Round 1: share=3 each. A caps at 2 (overflow 1), B caps at 3 (exact), C gets 3. # Round 2: remaining=1 should go to C, not be lost. allocations = strategy.allocate_quota( batches=[batch_a, batch_b, batch_c], available_quota=9, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_a.campaign_id, batch_id=batch_a.id, emails_to_send=2, ), QuotaAllocation( campaign_id=batch_b.campaign_id, batch_id=batch_b.id, emails_to_send=3, ), QuotaAllocation( campaign_id=batch_c.campaign_id, batch_id=batch_c.id, emails_to_send=4, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:00:00") def test_quota_exceeds_total_capacity( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test that quota exceeding total batch capacity allocates up to capacity.""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=10, batch_offset=0, updated_at=current_time - timedelta(minutes=2), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=20, batch_offset=0, updated_at=current_time - timedelta(minutes=1), ) batch_2.id = 2 allocations = strategy.allocate_quota( batches=[batch_1, batch_2], available_quota=500, current_time=current_time, ) assert allocations == IsList( QuotaAllocation( campaign_id=batch_1.campaign_id, batch_id=batch_1.id, emails_to_send=10, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=20, ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:00:00") def test_tiny_remainder_skips_full_batches( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test that when distributing a tiny remainder (quota_per_batch==0), full batches are skipped and remaining quota goes to the next batch.""" current_time = timezone.now() # minute=0, 3 batches → sorted order: A, B, C (no rotation) batch_a = build_model( CampaignBatch, batch_size=1, batch_offset=0, # capacity 1 updated_at=current_time - timedelta(minutes=3), ) batch_a.id = 1 batch_b = build_model( CampaignBatch, batch_size=1, batch_offset=0, # capacity 1 updated_at=current_time - timedelta(minutes=2), ) batch_b.id = 2 batch_c = build_model( CampaignBatch, batch_size=1000, batch_offset=0, # capacity 1000 updated_at=current_time - timedelta(minutes=1), ) batch_c.id = 3 allocations = strategy.allocate_quota( batches=[batch_a, batch_b, batch_c], available_quota=5, current_time=current_time, ) # Total capacity is 1002, quota is 5 → all 5 should be allocated assert allocations == IsList( QuotaAllocation( campaign_id=batch_a.campaign_id, batch_id=batch_a.id, emails_to_send=1 ), QuotaAllocation( campaign_id=batch_b.campaign_id, batch_id=batch_b.id, emails_to_send=1 ), QuotaAllocation( campaign_id=batch_c.campaign_id, batch_id=batch_c.id, emails_to_send=3 ), check_order=False, ) @freezegun.freeze_time("2025-11-20 10:00:00") def test_all_batches_exactly_filled( self, strategy: FairDistributionStrategy, build_model: BuildModel ) -> None: """Test when quota exactly matches total capacity of all batches.""" current_time = timezone.now() batch_1 = build_model( CampaignBatch, batch_size=50, batch_offset=0, updated_at=current_time - timedelta(minutes=2), ) batch_1.id = 1 batch_2 = build_model( CampaignBatch, batch_size=50, batch_offset=0, updated_at=current_time - timedelta(minutes=1), ) batch_2.id = 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=50, ), QuotaAllocation( campaign_id=batch_2.campaign_id, batch_id=batch_2.id, emails_to_send=50, ), check_order=False, )