from dataclasses import dataclass from datetime import datetime from typing import List, Dict, Optional from db import db from models.projects_history import HistoryActionType from tests.test_utils import FixtureLoader from workers.gras_project_worker import GRASProjectWorker from tests import BaseTestCase from tests.workers.fixtures.project_worker_fixture import fixture from models import models, Project, ProjectHistoryItem from utils.snowflake.models import GRASProjectModel, GRASArtist @dataclass class ArtistModel: id: str name: str is_deleted: bool = False primary: bool = False @dataclass class HistoryItemModel: action: HistoryActionType old_value: Optional[List or Dict] = None new_value: Optional[List or Dict] = None class GRASWorkerTestCase(BaseTestCase): def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) def test_project_import_should_add_new_artists(self): sut = GRASProjectWorker() model = GRASProjectModel( id="GRAS_100", title="DJ GRAS feat. 100 id", rep_owner_key="rep-2", start_date=datetime.now(), deleted=False ) model.featured_artists = [ GRASArtist(id="30", name="Terry", is_main=False, is_primary=False, popularity=10), GRASArtist(id="190", name="Greg", is_main=False, is_primary=False, popularity=8), ] model.primary_artists = [ GRASArtist(id="20", name="Ched", is_main=False, is_primary=False, popularity=6), GRASArtist(id="39", name="Alexa", is_main=False, is_primary=False, popularity=2), ] sut.perform([model]) project = db.session.query(Project).filter(Project.gras_project_code == "GRAS_100").first() items = project.target_items self.soft_assert_equal(len(items), 6) self.__assert_target_item(items[0], ArtistModel(id="GRAS_140", name="Fibonacci")) self.__assert_target_item(items[1], ArtistModel(id="GRAS_20", name="Ched", primary=True)) self.__assert_target_item(items[2], ArtistModel(id="GRAS_30", name="Terry")) self.__assert_target_item(items[3], ArtistModel(id="GRAS_60", name="artist2e", is_deleted=True)) self.__assert_target_item(items[4], ArtistModel(id="GRAS_39", name="Alexa", primary=True)) self.__assert_target_item(items[5], ArtistModel(id="GRAS_190", name="Greg")) self.assert_all() def test_project_import_deleted_artist_should_remain_deleted(self): sut = GRASProjectWorker() model = GRASProjectModel( id="GRAS_100", title="DJ GRAS feat. 100 id", rep_owner_key="rep-2", start_date=datetime.now(), deleted=False ) model.primary_artists = [GRASArtist(id="40", name="Norman", is_main=False, is_primary=False, popularity=150)] sut.perform([model]) project = db.session.query(Project).filter(Project.gras_project_code == "GRAS_100").first() items = project.target_items self.soft_assert_equal(len(items), 3) self.__assert_target_item(items[0], ArtistModel(id="GRAS_140", name="Fibonacci")) self.__assert_target_item(items[1], ArtistModel(id="GRAS_40", name="Norman", is_deleted=True, primary=True)) self.__assert_target_item(items[2], ArtistModel(id="GRAS_60", name="artist2e", is_deleted=True)) self.assert_all() def test_project_import_artist_type_changed(self): sut = GRASProjectWorker() model = GRASProjectModel( id="GRAS_100", title="DJ GRAS feat. 100 id", rep_owner_key="rep-2", start_date=datetime.now(), deleted=False ) model.featured_artists = [GRASArtist(id="20", name="Ched", is_main=False, is_primary=False, popularity=6)] model.primary_artists = [GRASArtist(id="140", name="Fibonacci", is_main=False, is_primary=True, popularity=8)] sut.perform([model]) project = db.session.query(Project).filter(Project.gras_project_code == "GRAS_100").first() items = project.target_items self.soft_assert_equal(len(project.target_items), 3) self.__assert_target_item(items[0], ArtistModel(id="GRAS_140", name="Fibonacci", primary=True)) self.__assert_target_item(items[1], ArtistModel(id="GRAS_20", name="Ched")) self.__assert_target_item(items[2], ArtistModel(id="GRAS_60", name="artist2e", is_deleted=True)) self.assert_all() def test_project_import_artist_history_events_updated(self): sut = GRASProjectWorker() model = GRASProjectModel( id="GRAS_100", title="DJ GRAS feat. 100 id", rep_owner_key="rep-2", start_date=datetime.now(), deleted=False ) model.featured_artists = [GRASArtist(id="20", name="Ched", is_main=False, is_primary=False, popularity=6)] model.primary_artists = [GRASArtist(id="990", name="Simon", is_main=True, is_primary=True, popularity=96)] sut.perform([model]) items: List[ProjectHistoryItem] = ( db.session.query(ProjectHistoryItem) .join(Project) .filter(Project.gras_project_code == "GRAS_100") .order_by(ProjectHistoryItem.created_at.desc()) .all() ) self.assertEqual(len(items), 5) self.__assert_history_item( items[3], HistoryItemModel( action=HistoryActionType.ADD_PRIMARY_ARTISTS_TO_PROJECT, new_value=[{"name": "Simon", "id": 1}] ), ) self.__assert_history_item( items[2], HistoryItemModel( action=HistoryActionType.REMOVE_PRIMARY_ARTISTS_FROM_PROJECT, old_value=[{"name": "Ched", "id": 20}] ), ) self.__assert_history_item( items[1], HistoryItemModel( action=HistoryActionType.ADD_FEATURED_ARTISTS_TO_PROJECT, new_value=[{"name": "Ched", "id": 20}] ), ) self.__assert_history_item( items[0], HistoryItemModel( action=HistoryActionType.REMOVE_FEATURED_ARTISTS_FROM_PROJECT, old_value=[{"name": "Terry", "id": 30}] ), ) self.assert_all() def test_project_import_deleted_project_should_skip_import(self): sut = GRASProjectWorker() model = GRASProjectModel( id="GRAS_200", title="DJ GRAS", rep_owner_key="rep-2", start_date=datetime.now(), deleted=False ) model.featured_artists = [GRASArtist(id="90", name="Carl", is_main=False, is_primary=False, popularity=16)] model.primary_artists = [GRASArtist(id="1900", name="Lois", is_main=True, is_primary=True, popularity=86)] sut.perform([model]) project = db.session.query(Project).filter(Project.gras_project_code == "GRAS_200").first() self.soft_assert_equal(len(project.target_items), 1) self.soft_assert_equal(project.is_deleted, True) self.__assert_target_item(project.target_items[0], ArtistModel(id="GRAS_10", name="TestAtrist", primary=True)) self.assert_all() def __assert_target_item(self, item, artist: ArtistModel): self.soft_assert_equal(item.entity.external_id, artist.id) self.soft_assert_equal(item.entity.name, artist.name) self.soft_assert_equal(item.is_deleted, artist.is_deleted) self.soft_assert_equal(item.entity_type, 0 if artist.primary else 1) def __assert_history_item(self, item: ProjectHistoryItem, history_item: HistoryItemModel): self.soft_assert_equal(item.action_id, history_item.action.value) self.soft_assert_equal(item.old_value, history_item.old_value) self.soft_assert_equal(item.new_value, history_item.new_value)