import decimal from datetime import UTC, datetime import pytest from dirty_equals import IsList from fansifter_common.auth.account import Account from ows_text_campaigns.audiences.models import AudienceTextFan from ows_text_campaigns.campaigns.enums import CampaignStatus from ows_text_campaigns.campaigns.models import ( Campaign, SmsAreaCodeMapping, SmsComplianceRule, SmsTimezone, ) from ows_text_campaigns.campaigns.repositories import CampaignRepository from ows_text_campaigns.campaigns.types import CampaignCriteria from tests.unit.types import CreateModel class TestCampaignRepository: @pytest.mark.db def test_exists_by_name_and_account( self, repository: CampaignRepository, create_model: CreateModel ) -> None: campaign = create_model(Campaign) exists = repository.exists_by_name_and_account( name=campaign.name.upper(), account=campaign.account ) assert exists @pytest.mark.db def test_exists_by_name_and_account_not_exist( self, repository: CampaignRepository ) -> None: exists = repository.exists_by_name_and_account( name="name", account=Account(vendor_id=10, subaccount_id=0) ) assert not exists @pytest.mark.db def test_count_by_criteria_empty(self, repository: CampaignRepository) -> None: total = repository.count_by_criteria( CampaignCriteria( search=None, vendor_ids=set(), subaccount_ids=set(), global_participant_ids=None, statuses=None, ) ) assert total == 0 @pytest.mark.db def test_count_by_criteria_search( self, repository: CampaignRepository, create_model: CreateModel ) -> None: campaign = create_model(Campaign, name="Test Campaign") total = repository.count_by_criteria( CampaignCriteria( search="test", vendor_ids={campaign.vendor_id}, subaccount_ids=set(), global_participant_ids=None, statuses=None, ) ) assert total == 1 @pytest.mark.db def test_count_by_criteria_statuses( self, repository: CampaignRepository, create_model: CreateModel ) -> None: campaign_1 = create_model(Campaign, status=CampaignStatus.SENT) campaign_2 = create_model(Campaign, status=CampaignStatus.IN_PROGRESS) total = repository.count_by_criteria( CampaignCriteria( search=None, vendor_ids={ campaign_1.vendor_id, campaign_2.vendor_id, }, subaccount_ids=set(), global_participant_ids=None, statuses=[CampaignStatus.SENT, CampaignStatus.IN_PROGRESS], ) ) assert total == 2 @pytest.mark.db def test_find_by_criteria_empty(self, repository: CampaignRepository) -> None: items = repository.find_by_criteria( CampaignCriteria( search=None, vendor_ids=set(), subaccount_ids=set(), global_participant_ids=None, statuses=None, ), order_by=["sendAt.desc"], limit=10, offset=0, ) assert items == [] @pytest.mark.db def test_find_by_criteria_order_by_with_limit( self, repository: CampaignRepository, create_model: CreateModel, ) -> None: campaign_1 = create_model(Campaign, name="Test Campaign1") campaign_2 = create_model(Campaign, name="Test Campaign2") items = repository.find_by_criteria( CampaignCriteria( search=None, vendor_ids={ campaign_1.vendor_id, campaign_2.vendor_id, }, subaccount_ids=set(), global_participant_ids=None, statuses=None, ), order_by=["name.asc"], limit=2, offset=0, ) assert items == [campaign_1, campaign_2] @pytest.mark.db def test_find_by_criteria_statuses( self, repository: CampaignRepository, create_model: CreateModel ) -> None: campaign_1 = create_model(Campaign, status=CampaignStatus.SENT) campaign_2 = create_model(Campaign, status=CampaignStatus.IN_PROGRESS) create_model(Campaign, status=CampaignStatus.SCHEDULED) items = repository.find_by_criteria( CampaignCriteria( search=None, vendor_ids={ campaign_1.vendor_id, campaign_2.vendor_id, }, subaccount_ids=None, global_participant_ids=None, statuses=[CampaignStatus.SENT, CampaignStatus.IN_PROGRESS], ), order_by=["name.asc"], limit=10, offset=0, ) assert items == IsList(campaign_1, campaign_2, check_order=False) @pytest.mark.db def test_find_by_criteria_global_participant_ids( self, repository: CampaignRepository, create_model: CreateModel, ) -> None: global_participant_id = "123" campaign = create_model( Campaign, global_participant_id=global_participant_id, ) items = repository.find_by_criteria( CampaignCriteria( search=None, vendor_ids={campaign.vendor_id}, subaccount_ids=None, global_participant_ids={global_participant_id}, statuses=None, ), order_by=["name.asc"], limit=10, offset=0, ) assert items == [campaign] @pytest.mark.db def test_get_send_timezone_resolves_shared_area_code_by_state( self, repository: CampaignRepository, create_model: CreateModel ) -> None: # Australian area code "8" spans several states/timezones. perth_timezone = create_model( SmsTimezone, iana="Australia/Perth", utc_offset_std=decimal.Decimal(8) ) adelaide_timezone = create_model( SmsTimezone, iana="Australia/Adelaide", utc_offset_std=decimal.Decimal("9.5"), ) western_australia_rule = create_model( SmsComplianceRule, country_iso2="AU", state_province="WA", timezone_id=perth_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) south_australia_rule = create_model( SmsComplianceRule, country_iso2="AU", state_province="SA", timezone_id=adelaide_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) create_model( SmsAreaCodeMapping, country_iso2="AU", state_province="WA", area_code="8", compliance_rule_id=western_australia_rule.id, ) create_model( SmsAreaCodeMapping, country_iso2="AU", state_province="SA", area_code="8", compliance_rule_id=south_australia_rule.id, ) campaign = create_model(Campaign, audience_id="aud-au") create_model( AudienceTextFan, audience_id="aud-au", fan_id="f1", fan_country="AU", fan_state="WA", fan_area_code="8", ) result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 3, 0, tzinfo=UTC), ) # Resolved to Perth via the fan's state, not fanned out into Adelaide too. assert [tz.iana_tz_id for tz in result] == ["Australia/Perth"] @pytest.mark.db def test_get_send_timezone_without_area_code_resolved_by_state( self, repository: CampaignRepository, create_model: CreateModel ) -> None: adelaide_timezone = create_model( SmsTimezone, iana="Australia/Adelaide", utc_offset_std=decimal.Decimal("9.5"), ) south_australia_rule = create_model( SmsComplianceRule, country_iso2="AU", state_province="SA", timezone_id=adelaide_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) create_model( SmsAreaCodeMapping, country_iso2="AU", state_province="SA", area_code="8", compliance_rule_id=south_australia_rule.id, ) campaign = create_model(Campaign, audience_id="aud-no-area") create_model( AudienceTextFan, audience_id="aud-no-area", fan_id="f1", fan_country="AU", fan_state="SA", fan_area_code=None, ) result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 3, 0, tzinfo=UTC), ) # No area code, but still matched by state instead of being dropped. assert [tz.iana_tz_id for tz in result] == ["Australia/Adelaide"] @pytest.mark.db def test_get_send_timezone_country_wide_default_rule( self, repository: CampaignRepository, create_model: CreateModel ) -> None: london_timezone = create_model( SmsTimezone, iana="Europe/London", utc_offset_std=decimal.Decimal(0) ) create_model( SmsComplianceRule, country_iso2="GB", state_province="", timezone_id=london_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) campaign = create_model(Campaign, audience_id="aud-gb") create_model( AudienceTextFan, audience_id="aud-gb", fan_id="f1", fan_country="GB", fan_state=None, fan_area_code=None, ) result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 12, 0, tzinfo=UTC), ) assert len(result) == 1 assert result[0].iana_tz_id == "Europe/London" assert result[0].state_provinces == [] @pytest.mark.db def test_get_send_timezone_before_safe_hours( self, repository: CampaignRepository, create_model: CreateModel ) -> None: new_york_timezone = create_model( SmsTimezone, iana="America/New_York", utc_offset_std=decimal.Decimal(-5) ) new_york_rule = create_model( SmsComplianceRule, country_iso2="US", state_province="NY", timezone_id=new_york_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) create_model( SmsAreaCodeMapping, country_iso2="US", state_province="NY", area_code="212", compliance_rule_id=new_york_rule.id, ) campaign = create_model(Campaign, audience_id="aud-us") create_model( AudienceTextFan, audience_id="aud-us", fan_id="f1", fan_country="US", fan_state="NY", fan_area_code="212", ) # 12:00 UTC on a Wednesday -> 07:00 EST, before the 09:00 safe start. result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 12, 0, tzinfo=UTC), ) assert len(result) == 1 send_timezone = result[0] assert send_timezone.iana_tz_id == "America/New_York" assert send_timezone.state_provinces == ["NY"] # quiet-hours bounds are the safe window, min < max (regression guard). assert send_timezone.quiet_hours_min == "09:00" assert send_timezone.quiet_hours_max == "20:00" assert send_timezone.is_in_safe_hours is False assert send_timezone.send_at_local == datetime(2025, 1, 15, 7, 0) # Next safe time is 09:00 EST == 14:00 UTC the same day. assert send_timezone.next_safe_time == datetime(2025, 1, 15, 14, 0, tzinfo=UTC) @pytest.mark.db def test_get_send_timezone_within_safe_hours( self, repository: CampaignRepository, create_model: CreateModel ) -> None: new_york_timezone = create_model( SmsTimezone, iana="America/New_York", utc_offset_std=decimal.Decimal(-5) ) new_york_rule = create_model( SmsComplianceRule, country_iso2="US", state_province="NY", timezone_id=new_york_timezone.id, safe_hours_start_weekday="09:00", safe_hours_end_weekday="20:00", safe_hours_start_sunday="09:00", safe_hours_end_sunday="20:00", priority=100, is_active=True, ) create_model( SmsAreaCodeMapping, country_iso2="US", state_province="NY", area_code="212", compliance_rule_id=new_york_rule.id, ) campaign = create_model(Campaign, audience_id="aud-us-safe") create_model( AudienceTextFan, audience_id="aud-us-safe", fan_id="f1", fan_country="US", fan_state="NY", fan_area_code="212", ) # 18:00 UTC -> 13:00 EST, inside the 09:00-20:00 window. result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 18, 0, tzinfo=UTC), ) assert len(result) == 1 assert result[0].is_in_safe_hours is True assert result[0].next_safe_time is None @pytest.mark.db def test_get_send_timezone_no_matching_fans( self, repository: CampaignRepository, create_model: CreateModel ) -> None: campaign = create_model(Campaign, audience_id="aud-empty") result = repository.get_send_timezone( campaign_id=campaign.id, send_at=datetime(2025, 1, 15, 12, 0, tzinfo=UTC), ) assert result == []