from datetime import datetime, timedelta import pytest from fansifter_common.utils import timezone from email_campaigns.audiences.enums import AudienceTarget from email_campaigns.audiences.models import Audience from email_campaigns.campaigns.enums import EmailCampaignStatus from email_campaigns.campaigns.models import CampaignBatch, DeliveryQuota, EmailCampaign from email_campaigns.campaigns.services import SendTimeEstimatorService from email_campaigns.emails.models import EmailDomain from tests.unit.types import CreateModel, CreatePgModel class TestSendTimeEstimatorService: """Unit tests for SendTimeEstimatorService.""" @pytest.fixture def base_time(self) -> datetime: return timezone.now() @pytest.mark.db def test_estimate_single_campaign_no_queue( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 3000, "yahoo": 2000}, ) email_domain = create_model( EmailDomain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) campaign = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=audience.id, email_domain=email_domain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="yahoo", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=500.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=campaign, send_at=base_time, ) assert estimated_seconds == 14400 assert emails_count == 5000 @pytest.mark.db def test_estimate_draft_campaign_empty_pending( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1000, recipients_count_by_email_provider={"google": 1000}, ) email_domain = create_model( EmailDomain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=None, created_at=base_time, audience_id=audience.id, email_domain=email_domain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=campaign, send_at=base_time, ) assert estimated_seconds == 3600 assert emails_count == 1000 @pytest.mark.db def test_estimate_draft_with_scheduled_in_queue( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: pending_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, ) target_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) email_domain = create_model( EmailDomain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, send_at=base_time, scheduled_at=base_time - timedelta(hours=1), created_at=base_time - timedelta(hours=1), audience_id=pending_audience.id, email_domain=email_domain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) target_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=target_audience.id, email_domain=email_domain, vendor_id=target_audience.vendor_id, subaccount_id=target_audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=target_campaign, send_at=base_time, ) assert estimated_seconds == 36000 assert emails_count == 5000 @pytest.mark.db def test_estimate_draft_with_in_progress_in_queue( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: pending_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=10000, recipients_count_by_email_provider={"google": 10000}, ) target_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) email_domain = create_model( EmailDomain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) pending_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, send_at=base_time - timedelta(hours=1), scheduled_at=base_time - timedelta(hours=1), created_at=base_time - timedelta(hours=1), audience_id=pending_audience.id, email_domain=email_domain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) target_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=target_audience.id, email_domain=email_domain, vendor_id=target_audience.vendor_id, subaccount_id=target_audience.subaccount_id, ) create_model( CampaignBatch, campaign_id=pending_campaign.id, domain_id=email_domain.id, provider="google", batch_size=10000, batch_offset=3000, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(minutes=5), last_hour_sent_emails=100, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=target_campaign, send_at=base_time, ) assert estimated_seconds == 43200 assert emails_count == 5000 @pytest.mark.db def test_estimate_draft_with_old_audience_fallback( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1000, recipients_count_by_email_provider=None, ) email_domain = create_model( EmailDomain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=audience.id, email_domain=email_domain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=100.0, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="yahoo", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=100.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=campaign, send_at=base_time, ) assert estimated_seconds == 36000 assert emails_count == 1000 @pytest.mark.db def test_estimate_draft_multi_provider_bottleneck( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=15000, recipients_count_by_email_provider={ "google": 10000, "yahoo": 5000, }, ) email_domain = create_model( EmailDomain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=audience.id, email_domain=email_domain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=10000.0, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="yahoo", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=100.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=campaign, send_at=base_time, ) assert estimated_seconds == 180000 assert emails_count == 15000 @pytest.mark.db def test_estimate_draft_with_preemption( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: pending_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, ) target_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=2000, recipients_count_by_email_provider={"google": 2000}, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) email_domain = create_model( EmailDomain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, send_at=base_time + timedelta(hours=3), scheduled_at=base_time - timedelta(hours=1), created_at=base_time - timedelta(hours=1), audience_id=pending_audience.id, email_domain=email_domain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) target_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=target_audience.id, email_domain=email_domain, vendor_id=target_audience.vendor_id, subaccount_id=target_audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=target_campaign, send_at=base_time, ) assert estimated_seconds == 7200 assert emails_count == 2000 @pytest.mark.db def test_estimate_draft_with_preemption_blocks_target( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: pending_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=3000, recipients_count_by_email_provider={"google": 3000}, ) target_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) email_domain = create_model( EmailDomain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, send_at=base_time + timedelta(minutes=90), scheduled_at=base_time - timedelta(hours=1), created_at=base_time - timedelta(hours=1), audience_id=pending_audience.id, email_domain=email_domain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) target_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=target_audience.id, email_domain=email_domain, vendor_id=target_audience.vendor_id, subaccount_id=target_audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=target_campaign, send_at=base_time, ) assert estimated_seconds == 28800 assert emails_count == 5000 @pytest.mark.db def test_estimate_draft_with_gap_between_campaigns( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, base_time: datetime, ) -> None: pending_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1000, recipients_count_by_email_provider={"google": 1000}, ) target_audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=500, recipients_count_by_email_provider={"google": 500}, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) email_domain = create_model( EmailDomain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, send_at=base_time, scheduled_at=base_time - timedelta(hours=2), created_at=base_time - timedelta(hours=2), audience_id=pending_audience.id, email_domain=email_domain, vendor_id=pending_audience.vendor_id, subaccount_id=pending_audience.subaccount_id, ) target_campaign = create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, send_at=base_time, scheduled_at=base_time, created_at=base_time, audience_id=target_audience.id, email_domain=email_domain, vendor_id=target_audience.vendor_id, subaccount_id=target_audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_sent_at=base_time - timedelta(hours=1), last_hour_sent_emails=0, last_hourly_quota=1000.0, ) estimated_seconds, emails_count = service.estimate_send_duration( campaign=target_campaign, send_at=base_time, ) assert estimated_seconds == 5400 assert emails_count == 500 @pytest.mark.db def test_estimate_send_duration_bulk_empty( self, service: SendTimeEstimatorService, ) -> None: result = service.estimate_send_duration_bulk([]) assert result == [] @pytest.mark.db def test_estimate_send_duration_bulk_single_campaign( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, ) -> None: audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 3000, "yahoo": 2000}, ) email_domain = create_model( EmailDomain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) campaign = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience.id, email_domain=email_domain, vendor_id=audience.vendor_id, subaccount_id=audience.subaccount_id, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_hourly_quota=1000.0, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="yahoo", last_hourly_quota=500.0, ) result = service.estimate_send_duration_bulk([campaign]) assert len(result) == 1 campaign_id, duration, emails_count = result[0] assert campaign_id == campaign.id assert 14000 <= duration <= 15000 assert emails_count == 5000 @pytest.mark.db def test_estimate_send_duration_bulk_multiple_campaigns( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, ) -> None: audience1 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1000, recipients_count_by_email_provider={"google": 1000}, ) audience2 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=2000, recipients_count_by_email_provider={"google": 2000}, ) email_domain = create_model(EmailDomain) campaign1 = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience1.id, email_domain=email_domain, ) campaign2 = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, audience_id=audience2.id, email_domain=email_domain, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_hourly_quota=1000.0, ) result = service.estimate_send_duration_bulk([campaign1, campaign2]) assert len(result) == 2 campaign_ids = {r[0] for r in result} assert campaign_ids == {campaign1.id, campaign2.id} @pytest.mark.db def test_estimate_send_duration_bulk_scheduled_uses_campaign_send_at( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, ) -> None: from fansifter_common.utils import timezone now = timezone.now() audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=10000, recipients_count_by_email_provider={"google": 10000}, ) email_domain = create_model(EmailDomain) campaign_future = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience.id, email_domain=email_domain, scheduled_at=now, send_at=now + timedelta(hours=2), ) campaign_soon = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience.id, email_domain=email_domain, scheduled_at=now, send_at=now + timedelta(minutes=30), ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_hourly_quota=1000.0, ) result = service.estimate_send_duration_bulk([campaign_future, campaign_soon]) assert len(result) == 2 result_dict = {r[0]: r[1] for r in result} duration_future = result_dict[campaign_future.id] duration_soon = result_dict[campaign_soon.id] assert duration_future > 10000 assert duration_soon > 10000 assert duration_future != duration_soon @pytest.mark.db def test_estimate_send_duration_bulk_in_progress_uses_now( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, ) -> None: from fansifter_common.utils import timezone now = timezone.now() audience = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=5000, recipients_count_by_email_provider={"google": 5000}, ) email_domain = create_model(EmailDomain) campaign_in_progress = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, audience_id=audience.id, email_domain=email_domain, send_at=now - timedelta(hours=1), ) create_model( CampaignBatch, campaign_id=campaign_in_progress.id, domain_id=email_domain.id, provider="google", batch_size=5000, batch_offset=2000, ) create_model( DeliveryQuota, domain_id=email_domain.id, provider="google", last_hourly_quota=1000.0, ) result = service.estimate_send_duration_bulk([campaign_in_progress]) assert len(result) == 1 campaign_id, duration, emails_count = result[0] assert campaign_id == campaign_in_progress.id assert duration > 1000 assert emails_count == 3000 @pytest.mark.db def test_estimate_send_duration_bulk_multiple_domains( self, service: SendTimeEstimatorService, create_model: CreateModel, create_pg_model: CreatePgModel, ) -> None: audience1 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1000, recipients_count_by_email_provider={"google": 600, "yahoo": 400}, ) audience2 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=1500, recipients_count_by_email_provider={"google": 900, "yahoo": 600}, ) audience3 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=2000, recipients_count_by_email_provider={"google": 1200, "yahoo": 800}, ) audience4 = create_pg_model( Audience, target=AudienceTarget.EMAIL, fan_count=2500, recipients_count_by_email_provider={"google": 1500, "yahoo": 1000}, ) domain1 = create_model(EmailDomain) domain2 = create_model(EmailDomain) campaign1_d1 = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience1.id, email_domain=domain1, ) campaign2_d1 = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, audience_id=audience2.id, email_domain=domain1, ) campaign1_d2 = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, audience_id=audience3.id, email_domain=domain2, ) campaign2_d2 = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, audience_id=audience4.id, email_domain=domain2, ) create_model( DeliveryQuota, domain_id=domain1.id, provider="google", last_hourly_quota=600.0, ) create_model( DeliveryQuota, domain_id=domain1.id, provider="yahoo", last_hourly_quota=400.0, ) create_model( DeliveryQuota, domain_id=domain2.id, provider="google", last_hourly_quota=1200.0, ) create_model( DeliveryQuota, domain_id=domain2.id, provider="yahoo", last_hourly_quota=800.0, ) result = service.estimate_send_duration_bulk( [campaign1_d1, campaign2_d1, campaign1_d2, campaign2_d2] ) assert len(result) == 4 results_by_id = {r[0]: (r[1], r[2]) for r in result} assert campaign1_d1.id in results_by_id assert campaign2_d1.id in results_by_id assert campaign1_d2.id in results_by_id assert campaign2_d2.id in results_by_id duration1_d1, emails1_d1 = results_by_id[campaign1_d1.id] duration2_d1, emails2_d1 = results_by_id[campaign2_d1.id] duration1_d2, emails1_d2 = results_by_id[campaign1_d2.id] duration2_d2, emails2_d2 = results_by_id[campaign2_d2.id] # All campaigns should have positive durations assert duration1_d1 > 1000 assert duration2_d1 > 1000 assert duration1_d2 > 1000 assert duration2_d2 > 1000 assert len({duration1_d1, duration2_d1, duration1_d2, duration2_d2}) >= 3 assert emails1_d1 == 1000 assert emails2_d1 == 1500 assert emails1_d2 == 2000 assert emails2_d2 == 2500 def test_build_batch_index_empty(self) -> None: result = SendTimeEstimatorService._build_batch_index([]) assert result == {} def test_build_batch_index_single_batch(self) -> None: batch = CampaignBatch( campaign_id="campaign1", domain_id="domain1", provider="google", batch_size=1000, batch_offset=0, ) result = SendTimeEstimatorService._build_batch_index([batch]) assert "campaign1" in result assert "google" in result["campaign1"] assert result["campaign1"]["google"] == batch def test_build_batch_index_multiple_batches_same_campaign(self) -> None: batch_google = CampaignBatch( campaign_id="campaign1", domain_id="domain1", provider="google", batch_size=1000, batch_offset=0, ) batch_yahoo = CampaignBatch( campaign_id="campaign1", domain_id="domain1", provider="yahoo", batch_size=500, batch_offset=0, ) result = SendTimeEstimatorService._build_batch_index( [batch_google, batch_yahoo] ) assert "campaign1" in result assert "google" in result["campaign1"] assert "yahoo" in result["campaign1"] assert result["campaign1"]["google"] == batch_google assert result["campaign1"]["yahoo"] == batch_yahoo def test_build_batch_index_multiple_campaigns(self) -> None: batch1_google = CampaignBatch( campaign_id="campaign1", domain_id="domain1", provider="google", batch_size=1000, batch_offset=100, ) batch1_yahoo = CampaignBatch( campaign_id="campaign1", domain_id="domain1", provider="yahoo", batch_size=500, batch_offset=50, ) batch2_google = CampaignBatch( campaign_id="campaign2", domain_id="domain1", provider="google", batch_size=2000, batch_offset=200, ) result = SendTimeEstimatorService._build_batch_index( [batch1_google, batch1_yahoo, batch2_google] ) assert "campaign1" in result assert "google" in result["campaign1"] assert "yahoo" in result["campaign1"] assert result["campaign1"]["google"] == batch1_google assert result["campaign1"]["yahoo"] == batch1_yahoo assert "campaign2" in result assert "google" in result["campaign2"] assert result["campaign2"]["google"] == batch2_google assert "yahoo" not in result["campaign2"] def test_build_batch_index_o1_lookup(self) -> None: batches = [ CampaignBatch( campaign_id=f"campaign{i}", domain_id="domain1", provider="google", batch_size=1000, batch_offset=0, ) for i in range(100) ] result = SendTimeEstimatorService._build_batch_index(batches) assert len(result) == 100 for i in range(100): campaign_id = f"campaign{i}" assert campaign_id in result assert "google" in result[campaign_id] assert result[campaign_id]["google"].campaign_id == campaign_id