from datetime import date from typing import List from db import db from utils.snowflake.tiktok_data.models import TikTokAdSetModel from models import models, CampaignAd, CampaignAdLink from tests import BaseTestCase from tests.test_utils import FixtureLoader from tests.workers.fixtures.google_campaigns_worker_fixture import fixture from workers.tiktok_ad_sets_worker import TikTokAdSetsWorker class TikTokAdSetTestCase(BaseTestCase): def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) @staticmethod def __default_ad_set_model() -> TikTokAdSetModel: model = TikTokAdSetModel( id=10697529516, campaign_id=10697529516, name="Maren Morris Entertainment", objective="REACH", genders=["FEMALE", "MALE", "NONE"], ages=["AGE_18_24", "AGE_13_17", "AGE_55_100", "AGE_25_34", "AGE_35_44", "NONE", "AGE_45_54"], start_date=date(2020, 7, 29), end_date=date(2020, 8, 12), country_codes=["US", "None"], budget_spend=1278.76, planned_budget=1500.0, destination_link="www.sony.com" ) return model def __campaign_ads_related_to_campaign(self, campaign_id: int) -> List[CampaignAd]: return ( db.session.query(CampaignAd) .filter(CampaignAd.campaign_id == campaign_id).all() ) def __get_campaign_ad_by_external_id(self, external_id: int) -> CampaignAd: return ( db.session.query(CampaignAd) .filter(CampaignAd.external_id == str(external_id)) .one_or_none() ) def __get_campaign_ad_link(self, campaign_ad: CampaignAd) -> CampaignAdLink: return db.session.query(CampaignAdLink).filter(CampaignAdLink.campaign_ad_id == campaign_ad.id).first() def test_import_should_add_new_campaign_ad_if_not_exist(self): sut = TikTokAdSetsWorker() model = self.__default_ad_set_model() model.campaign_id = 234567890 campaigns_before_import = self.__campaign_ads_related_to_campaign(100) sut.perform([model]) campaigns_after_import = self.__campaign_ads_related_to_campaign(100) campaign_ad = self.__get_campaign_ad_by_external_id(10697529516) link = self.__get_campaign_ad_link(campaign_ad) self.soft_assert_equal(len(campaigns_before_import) + 1, len(campaigns_after_import)) self.soft_assert_equal(link.link, "www.sony.com") self.assert_all() def test_ad_set_import_update_existing_ad_set(self): sut = TikTokAdSetsWorker() model = self.__default_ad_set_model() model.id = 101 model.name = "Campaign Ad 1 update" model.campaign_id = 234567890 sut.perform([model]) items: List[CampaignAd] = self.__campaign_ads_related_to_campaign(100) self.soft_assert_equal(len(items), 3) self.soft_assert_equal(items[-1].name, "Campaign Ad 1 update") self.assert_all() def test_ad_set_import_without_campaign(self): sut = TikTokAdSetsWorker() model = self.__default_ad_set_model() sut.perform([model]) campaign_ad = self.__get_campaign_ad_by_external_id(10697529516) self.soft_assert_equal(campaign_ad, None) self.assert_all() def test_ad_set_import_none_destination_link_should_not_create_link(self): sut = TikTokAdSetsWorker() model = self.__default_ad_set_model() model.destination_link = None model.campaign_id = 234567890 sut.perform([model]) campaign_ad = self.__get_campaign_ad_by_external_id(10697529516) link = self.__get_campaign_ad_link(campaign_ad) self.assertEqual(link, None)