from datetime import date from decimal import Decimal from typing import List from db import db 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 utils.snowflake.google_data.models import GoogleCampaignModel, GRASCampaignUrl, GRASCampaignNetworkType, LinkfireLink from workers.google_campaigns_worker import GoogleCampaignsWorker class GoogleCampaignTestCase(BaseTestCase): def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) def __default_campaign_model(self) -> GoogleCampaignModel: model = GoogleCampaignModel( id="10697529516", name="Chris Brown - CA (Go Crazy Video) - RT TVIS", bidding_type="cpv", types=[GRASCampaignNetworkType("YouTube Videos", "YouTube Videos")], genders=["Female", "Male", "Undetermined"], ages=["45-54", "35-44", "25-34", "55-64", "65 or more", "18-24", "Undetermined"], start_date=date(2020, 7, 29), end_date=date(2020, 8, 12), links=[GRASCampaignUrl("www.youtube.com/GoCrazy", ["https://www.youtube.com/watch?v=dPhwbZBvW2o"])], territories=[2124], account=8653387137, spend=1278.76, budget=1500.0, linkfire_links=[LinkfireLink(link_id="2b190f7c-cf68-4741-9c06-2718c473f6e5", link_url="lnk.to/Blink_182")], ) model.artists = ["GRAS_10", "GRAS_20"] 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, Project.project_campaigns) .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 = GoogleCampaignsWorker() model = self.__default_campaign_model() sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("GRAS_100") self.assertEqual(len(items), 1) self.assertEqual(items[0].status, 0) self.assertEqual(items[0].campaign.planned_budget, Decimal("1500.00")) self.assertEqual(items[0].campaign.budget_spend, Decimal("1278.76")) self.assertEqual(items[0].campaign.provider_id, 1) campaign_link = items[0].campaign.linkfire_campaigns[0] link = campaign_link.linkfire_link self.assertEqual(link.link_id, "2b190f7c-cf68-4741-9c06-2718c473f6e5") self.assertEqual(link.link_url, "lnk.to/Blink_182") self.assertEqual(campaign_link.add_type, 0) def test_campaigns_import_should_map_provider(self): sut = GoogleCampaignsWorker() 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, 3) def test_campaigns_import_update_existing_campaign_should_update_and_create_history_events(self): sut = GoogleCampaignsWorker() model = self.__default_campaign_model() sut.perform([model]) model.territories = [2036, 2840] model.bidding_type = "cpm" model.name = "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(history_items[0].action_id, HistoryActionType.ADD_CAMPAIGN.value) self.assertEqual(history_items[1].action_id, HistoryActionType.CHANGE_CAMPAIGN_LINKFIRE_LINKS.value) self.assertEqual(history_items[2].action_id, HistoryActionType.CHANGE_CAMPAIGN.value) self.assertEqual(history_items[3].action_id, HistoryActionType.UPDATE_CAMPAIGN_TIMELINE.value) self.assertEqual(history_items[4].action_id, HistoryActionType.UPDATE_CAMPAIGN_OBJECTIVE.value) def test_campaigns_import_update_existing_campaign_should_not_reassign_approved_project(self): sut = GoogleCampaignsWorker() 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_add_new_campaign_should_not_unassign_previous(self): sut = GoogleCampaignsWorker() model = self.__default_campaign_model() model.id = "45677567856" model.artists = ["GRAS_10"] model.account = 2333825793 model.name = "Calboy - CA (Go Crazy Video)" model.start_date = date(2020, 12, 6) model.end_date = date(2020, 12, 10) sut.perform([model]) items = sorted(self.__campaigns_relations_for_project("GRAS_300"), key=lambda x: x.campaign.external_id) self.assertEqual(len(items), 2) self.assertEqual(items[0].status, 0) self.assertEqual(items[0].campaign.external_id, "45677567856") self.assertEqual(items[1].status, 0) self.assertEqual(items[1].campaign.external_id, "677567856") items = self.__campaigns_relations_for_project("GRAS_400") self.assertEqual(len(items), 1) self.assertEqual(items[0].status, 0) self.assertEqual(items[0].campaign.external_id, "45677567856") def test_campaigns_import_should_assign_campaign_to_single_project_if_has_prs_code(self): sut = GoogleCampaignsWorker() model = self.__default_campaign_model() model.name = "Calboy - CA (Go Crazy Video) | PRS = 012345515" sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("GRAS_100") self.assertEqual(len(items), 1) def test_campaigns_import_should_not_assign_campaign_to_single_project_if_has_gras_id_but_different_currency(self): sut = GoogleCampaignsWorker() model = self.__default_campaign_model() model.name = "2001235655 | Adele - Hello" model.artists = [] model.account = 3029121817 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 = GoogleCampaignsWorker() model = self.__default_campaign_model() model.name = "Calboy - CA (Go Crazy Video) | PRS = 01234551436346" model.account = 3029121842 sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("2001235655") self.assertEqual(len(items), 1)