from datetime import date from decimal import Decimal from typing import List from db import db from utils.snowflake.tiktok_data.models import TikTokCampaignModel from models import models, ProjectCampaign, Project from models.projects_history import HistoryActionType, ProjectHistoryItem from tests import BaseTestCase from tests.test_utils import FixtureLoader from tests.workers.fixtures.google_campaigns_worker_fixture import fixture from workers.tiktok_campaigns_worker import TikTokCampaignsWorker class TikTokCampaignTestCase(BaseTestCase): def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) @staticmethod def __default_campaign_model() -> TikTokCampaignModel: model = TikTokCampaignModel( id=10697529516, name="012345515 | Chris Brown | CA (Go Crazy Video) | RT TVIS", 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), marketing_account_id=8653387137, artists=["GRAS_10", "GRAS_20"], country_codes=["US", "None"], budget_spend=1278.76, planned_budget=1500.0, ) return model def __campaigns_relations_for_project(self, project_id: str) -> List[ProjectCampaign]: return ( db.session.query(ProjectCampaign) .join(Project, ProjectCampaign.project) .filter(Project.gras_project_code == project_id) .all() ) def __assosiated_projects_for_campaign(self, campaign_id: int) -> List[Project]: return db.session.query(Project).join(ProjectCampaign).filter(ProjectCampaign.campaign_id == campaign_id).all() def __history_items_for_campaign(self, campaign_id: int) -> List[ProjectHistoryItem]: return db.session.query(ProjectHistoryItem).filter(ProjectHistoryItem.campaign_id == campaign_id).all() def test_campaigns_import_should_add_new_campaign(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("GRAS_100") self.soft_assert_equal(len(items), 1) self.soft_assert_equal(items[0].status, 0) self.soft_assert_equal(items[0].campaign.planned_budget, Decimal("1500.00")) self.soft_assert_equal(items[0].campaign.budget_spend, Decimal("1278.76")) self.soft_assert_equal(items[0].campaign.provider_id, 1) self.soft_assert_equal(items[0].campaign.type_id, 8) # All tiktok campaign have type Social Ads self.soft_assert_equal(items[0].campaign.gratis, False) self.soft_assert_equal(len(items[0].campaign.platforms), 1) self.soft_assert_equal(items[0].campaign.objective_id, 13) self.soft_assert_equal(len(items[0].campaign.territories), 1) self.soft_assert_equal(items[0].campaign.territories[0].territory_id, 231) self.assert_all() def test_campaigns_import_should_map_provider(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() model.account = 6449900252 sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("GRAS_100") self.assertEqual(len(items), 1) self.assertEqual(items[0].campaign.provider_id, 1) def test_campaigns_import_update_existing_campaign_should_update_and_create_history_events(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() sut.perform([model]) model.territories = ["IO"] model.objective = "BRAND_AWARENESS" model.name = "012345515 | UPD Chris Brown | CA (Go Crazy Video)" model.start_date = date(2020, 7, 30) model.end_date = date(2020, 8, 11) sut.perform([model]) items = self.__campaigns_relations_for_project("GRAS_100") self.assertEqual(len(items), 1) self.assertEqual(items[0].status, 0) history_items = self.__history_items_for_campaign(items[0].campaign.id) self.assertEqual(len(history_items), 4) self.assertEqual(history_items[0].action_id, HistoryActionType.ADD_CAMPAIGN.value) self.assertEqual(history_items[1].action_id, HistoryActionType.CHANGE_CAMPAIGN.value) self.assertEqual(history_items[2].action_id, HistoryActionType.UPDATE_CAMPAIGN_TIMELINE.value) self.assertEqual(history_items[3].action_id, HistoryActionType.UPDATE_CAMPAIGN_OBJECTIVE.value) def test_campaigns_import_update_existing_campaign_should_not_reassign_approved_project(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() model.id = "234567890" model.artists = ["GRAS_10", "GRAS_60"] model.name = "Calboy - CA (Go Crazy Video)" model.start_date = date(2020, 9, 10) model.end_date = date(2020, 9, 12) sut.perform([model]) items = self.__assosiated_projects_for_campaign(100) self.assertEqual(len(items), 1) self.assertEqual(items[0].gras_project_code, "GRAS_200") def test_campaigns_import_should_assign_campaign_to_single_project_if_has_gras_id(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() model.name = "2001235655 | Calboy - CA (Go Crazy Video)" model.account_id = 2333825793 sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("2001235655") self.assertEqual(len(items), 1) self.assertEqual(items[0].campaign.name, "2001235655 | Calboy - CA (Go Crazy Video)") def test_campaigns_import_should_not_assign_campaign_to_single_project_if_has_gras_id_but_different_currency(self): sut = TikTokCampaignsWorker() model = self.__default_campaign_model() model.name = "2001235655 | Adele - Hello" model.marketing_account_id = 3029121817 model.artists = [] sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("2001235655") self.assertEqual(len(items), 0) def test_campaigns_import_should_assign_campaign_to_projects_with_artists_if_has_no_gras_id(self): # assign if project currency equal to marketing account currency sut = TikTokCampaignsWorker() model = self.__default_campaign_model() model.name = "7823432643728642 | Calboy - CA (Go Crazy Video)" model.marketing_account_id = 3029121842 sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("2001235655") self.assertEqual(len(items), 1)