from datetime import date from decimal import Decimal from typing import List from db import db from utils.snowflake.facebook_data.models import FacebookCampaignModel 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 GRASCampaignUrl from workers.facebook_campaigns_worker import FacebookCampaignsWorker class FacebookCampaignTestCase(BaseTestCase): def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) def __default_campaign_model(self) -> FacebookCampaignModel: model = FacebookCampaignModel( id="10697529516", name="[FB] Chris Brown - CA (Go Crazy Video) - RT TVIS", objective="REACH", platforms=["instagram"], genders=["Female", "Male", "Undetermined"], ages=["45-54", "35-44", "25-34", "55-64", "65+", "18-24", "Unknown"], 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"])], account_id=8653387137, artists=["GRAS_10", "GRAS_20"], country_codes=["BO", "BR"], spend=1278.76, budget=1500.0, linkfire_links=[], ) 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 = FacebookCampaignsWorker() 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) self.assertEqual(items[0].campaign.type_id, 8) self.assertEqual(items[0].campaign.gratis, False) self.assertEqual(len(items[0].campaign.platforms), 1) def test_campaigns_import_should_map_provider(self): sut = FacebookCampaignsWorker() 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 = FacebookCampaignsWorker() model = self.__default_campaign_model() sut.perform([model]) model.territories = ["IO"] model.objective = "BRAND_AWARENESS" model.name = "[FB UPDATED] 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 = FacebookCampaignsWorker() 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 = FacebookCampaignsWorker() 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 = FacebookCampaignsWorker() model = self.__default_campaign_model() model.name = "2001235655 | Adele - Hello" model.artists = [] model.account_id = 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 = FacebookCampaignsWorker() model = self.__default_campaign_model() model.name = "7823432643728642 | Calboy - CA (Go Crazy Video)" model.account_id = 3029121842 sut.perform([model]) items: List[ProjectCampaign] = self.__campaigns_relations_for_project("2001235655") self.assertEqual(len(items), 1)