from typing import get_args import pytest from dirty_equals import IsList from fansifter_common.auth.account import Account from email_campaigns.campaigns.enums import EmailCampaignStatus from email_campaigns.campaigns.models import EmailCampaign from email_campaigns.campaigns.repositories import EmailCampaignRepository from email_campaigns.campaigns.types import EmailCampaignCriteria, EmailCampaignOrderBy from email_campaigns.emails.models import EmailAnalyticsV3Dbt, EmailDomain from email_campaigns.rosters.types import FanDataListId from tests.unit.types import CreateModel class TestEmailCampaignRepository: @pytest.mark.db def test_count_by_criteria_empty(self, repository: EmailCampaignRepository) -> None: total = repository.count_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=None, ) ) assert total == 0 @pytest.mark.db def test_count_by_criteria_search( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: create_model(EmailCampaign, name="Test Campaign") total = repository.count_by_criteria( EmailCampaignCriteria( search="test", campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=None, audience_id=None, ) ) assert total == 1 @pytest.mark.db def test_count_by_criteria_statuses( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: create_model(EmailCampaign, status=EmailCampaignStatus.SENT) create_model(EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS) total = repository.count_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=[EmailCampaignStatus.SENT, EmailCampaignStatus.IN_PROGRESS], audience_id=None, ) ) assert total == 2 @pytest.mark.db def test_find_by_criteria_empty(self, repository: EmailCampaignRepository) -> None: items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=None, audience_id=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: EmailCampaignRepository, create_model: CreateModel, ) -> None: campaign_1 = create_model(EmailCampaign, name="Test Campaign1") campaign_2 = create_model(EmailCampaign, name="Test Campaign2") items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=None, audience_id=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: EmailCampaignRepository, create_model: CreateModel, ) -> None: campaign_1 = create_model(EmailCampaign, status=EmailCampaignStatus.SENT) campaign_2 = create_model(EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS) create_model(EmailCampaign, status=EmailCampaignStatus.SCHEDULED) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=[EmailCampaignStatus.SENT, EmailCampaignStatus.IN_PROGRESS], audience_id=None, ), 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: EmailCampaignRepository, create_model: CreateModel, ) -> None: global_participant_id = "123" campaign = create_model( EmailCampaign, global_participant_id=global_participant_id, custom_list_id=None, ) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids={FanDataListId.from_artist_id(global_participant_id)}, statuses=None, audience_id=None, ), order_by=["name.asc"], limit=10, offset=0, ) assert items == [campaign] @pytest.mark.db def test_find_by_criteria_custom_list_ids( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: custom_list_id_1 = "123" custom_list_id_2 = "234" campaign = create_model( EmailCampaign, global_participant_id=None, custom_list_id=custom_list_id_1, ) create_model( EmailCampaign, global_participant_id=None, custom_list_id=custom_list_id_2, ) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids={FanDataListId.from_custom_list_id(custom_list_id_1)}, statuses=None, audience_id=None, ), order_by=["name.asc"], limit=10, offset=0, ) assert items == [campaign] @pytest.mark.db def test_find_by_criteria_fandata_list_ids( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: custom_list_id_1 = "123" custom_list_id_2 = "234" global_participant_id = "123" campaign_1 = create_model( EmailCampaign, global_participant_id=None, custom_list_id=custom_list_id_1, ) campaign_2 = create_model( EmailCampaign, global_participant_id=None, custom_list_id=global_participant_id, ) create_model( EmailCampaign, global_participant_id=None, custom_list_id=custom_list_id_2, ) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids={ FanDataListId.from_custom_list_id(custom_list_id_1), FanDataListId.from_artist_id(global_participant_id), }, statuses=None, audience_id=None, ), order_by=["name.asc"], limit=10, offset=0, ) assert items == IsList( campaign_1, campaign_2, check_order=False, ) @pytest.mark.db def test_exists_by_name_and_account( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: campaign = create_model(EmailCampaign) 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: EmailCampaignRepository ) -> None: exists = repository.exists_by_name_and_account( name="name", account=Account(vendor_id=10, subaccount_id=0) ) assert not exists @pytest.mark.db @pytest.mark.parametrize("order_by", get_args(EmailCampaignOrderBy)) def test_find_by_criteria_order_by( self, order_by: EmailCampaignOrderBy, repository: EmailCampaignRepository ) -> None: repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=None, audience_id=None, ), order_by=[order_by], limit=10, offset=0, ) @pytest.mark.db def test_find_by_criteria_order_by_analytics_field( self, repository: EmailCampaignRepository, create_model: CreateModel ) -> None: campaign_1 = create_model( EmailCampaign, ) campaign_2 = create_model( EmailCampaign, ) create_model( EmailAnalyticsV3Dbt, email_id=campaign_1.id, email_type=campaign_1.email_type, sends=10, ) create_model( EmailAnalyticsV3Dbt, email_id=campaign_2.id, email_type=campaign_2.email_type, sends=5, ) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=None, audience_id=None, ), order_by=["sends.asc"], limit=10, offset=0, ) assert items == [campaign_2, campaign_1] @pytest.mark.db def test_count_by_criteria_audience_id( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: audience_id = "test-audience-id" create_model(EmailCampaign, audience_id=audience_id) create_model(EmailCampaign, audience_id=audience_id) create_model(EmailCampaign, audience_id="other-audience-id") total = repository.count_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=set(), subaccount_ids=set(), fandata_list_ids=None, statuses=None, audience_id=audience_id, ) ) assert total == 2 @pytest.mark.db def test_find_by_criteria_audience_id( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: audience_id = "test-audience-id" campaign_1 = create_model(EmailCampaign, audience_id=audience_id) campaign_2 = create_model(EmailCampaign, audience_id=audience_id) create_model(EmailCampaign, audience_id="other-audience") items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=None, audience_id=audience_id, ), 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_audience_id_and_status( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: audience_id = "test-audience-id" campaign_1 = create_model( EmailCampaign, audience_id=audience_id, status=EmailCampaignStatus.SCHEDULED ) create_model( EmailCampaign, audience_id=audience_id, status=EmailCampaignStatus.DRAFT ) create_model( EmailCampaign, audience_id="other-audience", status=EmailCampaignStatus.SCHEDULED, ) items = repository.find_by_criteria( EmailCampaignCriteria( search=None, campaign_ids=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=[EmailCampaignStatus.SCHEDULED], audience_id=audience_id, ), order_by=["name.asc"], limit=10, offset=0, ) assert items == [campaign_1] @pytest.mark.db def test_search_by_criteria_campaign_ids( self, repository: EmailCampaignRepository, create_model: CreateModel ) -> None: campaign_1 = create_model(EmailCampaign) create_model(EmailCampaign) items = repository.search_by_criteria( EmailCampaignCriteria( search=None, campaign_ids={campaign_1.id}, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=None, audience_id=None, ), order_by=["createdAt.desc"], limit=10, ) assert items == [campaign_1] @pytest.mark.db def test_search_by_criteria_audience_id_and_status( self, repository: EmailCampaignRepository, create_model: CreateModel ) -> None: audience = "audience" campaign_1 = create_model( EmailCampaign, audience_id=audience, status=EmailCampaignStatus.SCHEDULED ) create_model( EmailCampaign, audience_id=audience, status=EmailCampaignStatus.DRAFT ) items = repository.search_by_criteria( EmailCampaignCriteria( campaign_ids=None, search=None, vendor_ids=None, subaccount_ids=None, fandata_list_ids=None, statuses=[EmailCampaignStatus.SCHEDULED], audience_id=audience, ), order_by=["createdAt.desc"], limit=10, ) assert items == [campaign_1] @pytest.mark.db def test_find_pending_campaigns_empty( self, repository: EmailCampaignRepository ) -> None: result = repository.find_pending_campaigns(domain_id="non-existent") assert result == [] @pytest.mark.db def test_find_pending_campaigns_returns_scheduled_and_in_progress( self, repository: EmailCampaignRepository, create_model: CreateModel, ) -> None: email_domain_1 = create_model(EmailDomain) email_domain_2 = create_model(EmailDomain) create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, email_domain=email_domain_2, ) scheduled = create_model( EmailCampaign, status=EmailCampaignStatus.SCHEDULED, email_domain=email_domain_1, ) in_progress = create_model( EmailCampaign, status=EmailCampaignStatus.IN_PROGRESS, email_domain=email_domain_1, ) create_model( EmailCampaign, status=EmailCampaignStatus.DRAFT, email_domain=email_domain_1, ) create_model( EmailCampaign, status=EmailCampaignStatus.SENT, email_domain=email_domain_1, ) result = repository.find_pending_campaigns(domain_id=email_domain_1.id) assert result == IsList(scheduled, in_progress, check_order=False)