import copy import elasticsearch import json import pytest from collections import defaultdict from datetime import date, datetime, timedelta from flask import url_for from http import HTTPStatus from typing import Iterable, List import src.legacy.redis_db.keys as keys from apollo_main_db.apollo.models import ApolloUserOnboarding, Genre, Globalization, StarredContent from apollo_main_db.spotify.models import ViewPlaylistTypeEnum from src.cache import redis_client from src.constants.core import ANDROID, IOS, VENDOR_APPLE, VENDOR_SPOTIFY from src.constants.http_status import HTTP_200_OK, HTTP_201_CREATED, HTTP_400_BAD_REQUEST, HTTP_401_UNAUTHORIZED, \ HTTP_404_NOT_FOUND, HTTP_409_CONFLICT from src.legacy.apollo_api.constants import NPSQuestionType, NPSResultStatus, TwilioMessageStatus from src.legacy.spotify.constants import SPOTIFY_PLAYLIST_IMAGE_URL_MASK from src.legacy.vendor_api.vendor import VendorAPI from tests.api.tracks.test_tracks_playlists import get_hh_track from tests.legacy.apollo_api import factories as apollo_factories from tests.legacy.apple_music import factories from tests.legacy.spotify.factories import SpotifyArtistFactory, SpotifyHotHitsPlaylistTrackStatsFactory, \ SpotifyPersonalizedPlaylistFactory, SpotifyTrack2Factory, SpotifyTrackArtistFactory from tests.legacy.util import generate_view_playlists pytest_plugins = [ "tests.legacy.vendor_api.fixtures.vendor_responses", "tests.legacy.apollo_api.fixtures.compared_tracks_responses", ] @pytest.mark.parametrize( "is_mobile,is_portal,on_boarding,client_type", ( (True, True, True, "mobile"), (True, False, True, "mobile"), (True, True, True, "portal"), (False, True, True, "portal"), (True, False, False, None), (False, True, True, None), ), ) def test_onboarding_retrieving( db_session, client, user_id, patch_auth_user, is_mobile: bool, is_portal: bool, on_boarding: bool, client_type ): patch_auth_user(user_id) apollo_factories.OnboardingFactory(user_id=user_id, is_mobile=is_mobile, is_portal=is_portal) assert len(db_session.query(ApolloUserOnboarding).all()) == 1 url = url_for("apollo_api_blueprint.onboarding", client_type=client_type) response = client.get(url) response_user_onboarding = response.json["user_onboarding"] assert response_user_onboarding == on_boarding if response_user_onboarding: assert response.json["created_at"] else: with pytest.raises(KeyError): assert response.json["created_at"] @pytest.mark.parametrize( "is_mobile,is_portal,client_type,created", ( (True, False, "mobile", True), (False, False, "mobile", True), (True, True, "mobile", True), (False, True, "mobile", True), (True, False, "portal", True), (False, True, "portal", True), (True, True, "portal", True), (False, False, "portal", True), (True, False, "mobile", False), (False, True, "mobile", False), (True, False, "portal", False), (False, True, "portal", False), (True, False, None, True), (False, True, None, True), (True, True, None, True), (False, False, None, True), (True, False, None, False), (False, True, None, False), ), ) def test_onboarding_post( db_session, client, user_id, patch_auth_user, is_mobile: bool, is_portal: bool, created: bool, client_type ): patch_auth_user(user_id) portal = client_type in ["portal", None] mobile = client_type == "mobile" if created: apollo_factories.OnboardingFactory(user_id=user_id, is_mobile=is_mobile, is_portal=is_portal) assert len(db_session.query(ApolloUserOnboarding).all()) == 1 else: assert len(db_session.query(ApolloUserOnboarding).all()) == 0 url = url_for("apollo_api_blueprint.onboarding") request_data = json.dumps(dict(client_type=client_type, timestamp=1681974898012)) if client_type else json.dumps({}) client.post(url, data=request_data, content_type="application/json") on_boarding = db_session.query(ApolloUserOnboarding).filter_by(user_id=user_id).first() if created: if portal: assert on_boarding.is_portal is True assert on_boarding.is_mobile == is_mobile elif mobile: assert on_boarding.is_mobile is True assert on_boarding.is_portal == is_portal else: if portal: assert on_boarding.is_portal is True assert on_boarding.is_mobile is False elif mobile: assert on_boarding.is_mobile is True assert on_boarding.is_portal is False @pytest.mark.parametrize( "client_type,conflict, status", ( ("mobile", True, HTTP_409_CONFLICT), ("portal", True, HTTP_409_CONFLICT), (None, True, HTTP_409_CONFLICT), ("mobile", False, HTTP_200_OK), ("portal", False, HTTP_200_OK), ("portal", False, HTTP_200_OK), ), ) def test_onboarding_conflict_post( db_session, client, user_id, patch_auth_user, conflict: bool, client_type, status, mocker ): patch_auth_user(user_id) if conflict: apollo_factories.OnboardingFactory(user_id=user_id) mocked_database = mocker.patch("src.legacy.apollo_api.views.session.query") class SessionQueryMock: def filter_by(self, user_id): return SessionQueryMock() def first(self): return None mocked_database.return_value = SessionQueryMock() url = url_for("apollo_api_blueprint.onboarding") request_data = json.dumps(dict(client_type=client_type)) if client_type else json.dumps({}) response = client.post(url, data=request_data, content_type="application/json") assert response.status_code == status def test_markets_retrieving_full_name(db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) tw = apollo_factories.ApolloMarketFactory.create(id=1, name="Taiwan", full_name="Taiwan") nt = apollo_factories.ApolloMarketFactory.create(id=2, name="Netherlands", full_name="the Netherlands") url = url_for("apollo_api_blueprint.markets", extended=True) response = client.get(url) response.json[0]["id"] = nt.id response.json[0]["code"] = nt.code response.json[0]["name"] = nt.name response.json[0]["full_name"] = nt.full_name response.json[1]["id"] = tw.id response.json[1]["code"] = tw.code response.json[1]["name"] = tw.name response.json[0]["full_name"] = tw.full_name def test_markets_retrieving(db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) primary_markets = [ apollo_factories.ApolloMarketFactory.create(id=(i + 1), code=_code, is_primary=True) for i, _code in enumerate(("us", "au")) ] non_primary_markets = [ apollo_factories.ApolloMarketFactory.create(id=(i + 100), code=_code, is_primary=False) for i, _code in enumerate(("ua", "mk", "si")) ] primary_market_ids = [m.id for m in primary_markets] non_primary_market_ids = [m.id for m in non_primary_markets] url = url_for("apollo_api_blueprint.markets", extended=True) response = client.get(url) assert len(response.json) == 5 for i in response.json: assert i["id"] in non_primary_market_ids + primary_market_ids url = url_for("apollo_api_blueprint.markets", extended=False) response = client.get(url) assert len(response.json) == 2 for i in response.json: assert i["id"] in primary_market_ids url = url_for("apollo_api_blueprint.markets") response = client.get(url) assert len(response.json) == 2 for i in response.json: assert i["id"] in primary_market_ids def test_get_starred_tracks(db_session, mocker, user_id, patch_auth_user, client): patch_auth_user(user_id) apollo_factories.StarredContentFactory.create(user_id=user_id, isrc="USRC11901901") apollo_factories.StarredContentFactory.create(user_id=user_id, isrc="USRC11901902") apollo_factories.StarredContentFactory.create_batch(2, user_id=user_id, isrc="USRC11901903") url = url_for("apollo_api_blueprint.starred_tracks", offset=0, limit=10) response = client.get(url) assert response.status_code == 200 assert len(response.json) == 3 def test_get_isrc_relation(db_session, user_id, patch_auth_user, client): patch_auth_user(user_id) main_to_related = {"USRC11901901": ["USRC11901911", "USRC11901921"], "USRC11901902": ["USRC11901912"]} for main, related in main_to_related.items(): for r in related: apollo_factories.IsrcRelationFactory.create(main_isrc=main, isrc=r) url = url_for("apollo_api_blueprint.isrc_relation_get_post_v1") response = client.get(url) assert response.status_code == 200 response_data = response.json assert len(response_data) == len(main_to_related) for main, related in response_data.items(): assert set(main_to_related[main]) == set(related) for main, related in main_to_related.items(): url = url_for("apollo_api_blueprint.isrc_relation_retrieve_get_delete_v1", isrc=main) response = client.get(url) assert response.status_code == 200 response_data = response.json assert len(response_data) == 1 for _main, _related in response_data.items(): assert main == _main assert set(related) == set(_related) def test_check_is_starred(db_session, user_id, patch_auth_user, client): patch_auth_user(user_id) isrc1 = "USRC11901901" isrc2 = "USRC11901902" isrc3 = "USRC11901903" apollo_factories.StarredContentFactory.create_batch(2, user_id=user_id) apollo_factories.StarredContentFactory.create(user_id=user_id, isrc=isrc1) apollo_factories.StarredContentFactory.create(user_id=user_id, isrc=isrc2) assert db_session.query(StarredContent).filter_by(user_id=user_id).count() == 4 isrc_list = ",".join([isrc1, isrc2, isrc3]) url = url_for("apollo_api_blueprint.starred_tracks_check", isrc=isrc_list) response = client.get(url) assert response.status_code == 200 response_data = response.json assert response_data[isrc1] is True assert response_data[isrc2] is True assert response_data[isrc3] is False @pytest.mark.parametrize( "isrc,status,code,uri", ( ("grum71029604", 400, "starring.broken_uri", "potify:track:0RhnShasURThgSiYEyfhxd"), ("GBUM71029604", 400, "starring.uri_absence", None), (None, 400, "starring.data_absence", None), ("GBUM71029604", 201, "starring.success", "1234567890"), ("grum71029604", 201, "starring.success", "1234567890"), ("GBUM71029604", 400, "starring.broken_uri", "12345678901"), ("grum71029604", 201, "starring.success", "345678901"), ("GBUM71029604", 400, "starring.broken_uri", "spotify:track:0RhnShasURThgSiYEyfhx"), ("GBUM71029604", 400, "starring.broken_uri", "spotify:track:0RhnShasURThgSiYEyfhxfj"), ("grum71029604", 400, "starring.broken_uri", "spotify:track:RRhnShasURThgSiYEyfhxd"), ("GBUM71029604", 201, "starring.success", "spotify:track:0RhnShasURThgSiYEyfhxd"), ("grum71029604", 201, "starring.success", "spotify:track:0RhnShasURThgSiYEyfhxd"), ("USUM71902679", 400, "starring.already_starred", "spotify:track:0RhnShasURThgSiYEyfhxd"), ("GBUM710296041", 400, "starring.broken_isrc", "spotify:track:0RhnShasURThgSiYEyfhxd"), ("GBU-71029604", 400, "starring.broken_isrc", "0RhnShasURThgSiYEyfhxd"), (None, 400, "starring.data_absence", "0RhnShasURThgSiYEyfhxd"), ), ) def test_starring_tracks(mocker, db_session, user_id, patch_auth_user, client, isrc, status, code, uri): patch_auth_user(user_id) url = url_for("apollo_api_blueprint.starred_tracks") apollo_factories.StarredContentFactory.create(isrc="USUM71902679", user_id=user_id) response = client.post(url, data=json.dumps(dict(uri=uri, isrc=isrc)), content_type="application/json") assert response.status_code == status assert response.json["code"] == code if status == 201: assert db_session.query(StarredContent).filter_by(user_id=user_id, uri=uri, isrc=isrc).all() @pytest.mark.parametrize( "isrc,status,code", ( ("USUM71902679", 200, "starring.success"), ("USUM7190267", 400, "starring.broken_isrc"), ("usum71902679", 200, "starring.success"), ("USU-71902679", 400, "starring.broken_isrc"), ("GBUM71029604", 400, "starring.already_unstarred"), (None, 400, "starring.data_absence"), ), ) def test_unstarring_tracks(db_session, user_id, patch_auth_user, client, isrc, status, code): patch_auth_user(user_id) apollo_factories.StarredContentFactory.create(isrc="USUM71902679", user_id=user_id) apollo_factories.StarredContentFactory.create(isrc="NOG841907010", user_id=user_id) url = url_for("apollo_api_blueprint.starred_tracks", isrc=isrc) response = client.delete(url) assert response.status_code == status assert response.json["code"] == code def get_app_markets_data() -> List[dict]: return [ dict( market_id=1, market_name="-", market_fullname="All (legacy)", language_code="sv-SE", active=True, ga_country_name="All (legacy)", default_service=None, services="", workout=0, include_other_playlists=0, fallback=1, ), dict( market_id=2, market_name="no", market_fullname="Norway", language_code="nb-NO", active=True, ga_country_name="Norway", default_service="spotify", services="spotify,apple", workout=0, include_other_playlists=0, fallback=0, ), ] def get_app_markets_responses() -> List[dict]: return [ dict( active=True, cultureInfo="sv-SE", defaultService=None, gaCountryName="All (legacy)", id=1, includeOtherPlaylists=False, name="All (legacy)", services=[], spotifyRegionCode="-", workoutMarket=False, ), dict( active=True, cultureInfo="nb-NO", defaultService="spotify", gaCountryName="Norway", id=2, includeOtherPlaylists=False, name="Norway", services=["spotify", "apple"], spotifyRegionCode="no", workoutMarket=False, ), ] def test_app_markets(db_session, user_id, patch_auth_user, client): patch_auth_user(user_id) url = url_for("apollo_api_blueprint.application_markets_get_post") items = get_app_markets_data() responses = get_app_markets_responses() for item in items: apollo_factories.MarketFactory.create(**item) response = client.get(url) assert response.status_code == 200 response_data = response.json assert len(response_data) == len(responses) for ind, data in enumerate(response_data): for k, v in data.items(): assert responses[ind][k] == v @pytest.mark.parametrize("os", (IOS, ANDROID)) def test_latest_go_versions_view(db_session, user_id, patch_auth_user, client, os): patch_auth_user(user_id) yesterday = datetime.now() - timedelta(days=1) today = datetime.now() ios_changelog = {"changelog": "test_ios"} android_changelog = {"changelog": "test_android"} # older_version_ios apollo_factories.MobileVersionFactory.create(id=1, created_at=yesterday, os=IOS) # older_version_android apollo_factories.MobileVersionFactory.create(id=2, created_at=yesterday, os=ANDROID) newer_version_ios = apollo_factories.MobileVersionFactory.create(id=3, created_at=today, os=IOS) newer_version_android = apollo_factories.MobileVersionFactory.create(id=4, created_at=today, os=ANDROID) apollo_factories.VersionCopiesFactory.create( os=IOS, version=newer_version_ios.version, data=json.dumps(ios_changelog) ) apollo_factories.VersionCopiesFactory.create( os=ANDROID, version=newer_version_android.version, data=json.dumps(android_changelog) ) url = url_for("apollo_api_blueprint.mobile_versions", os=os) response = client.get(url) assert response.status_code == 200 response_data = response.json assert os == response_data["os"] if os == ANDROID: assert newer_version_android.version == response_data["version"] assert response_data["changelog"] == android_changelog["changelog"] elif os == IOS: assert newer_version_ios.version == response_data["version"] assert response_data["changelog"] == ios_changelog["changelog"] def test_twilio_messaging_status_callback(db_session, user_id, patch_auth_user, client): url = url_for("apollo_api_blueprint.messaging_status_callback") data = { "ApiVersion": "2010-04-01", "MessagingServiceSid": "MGf42c458be7652b111856493a645983b7", "MessageStatus": TwilioMessageStatus.FAILED, "SmsSid": "SM72111b3c0d874816afb433508aa0b9bd", "SmsStatus": TwilioMessageStatus.FAILED, "From": "+19852601833", "To": "+380631435565", "ErrorCode": "21704", "MessageSid": "SM72111b3c0d874816afb433508aa0b9bd", "AccountSid": "ABCDEFGHIJKLM", } data = "&".join([f"{k}={v}" for k, v in data.items()]) response = client.post(url, data=data, content_type="application/x-www-form-urlencoded") assert response.status_code == 200 data = { "ApiVersion": "2010-04-01", "MessagingServiceSid": "MGf42c458be7652b111856493a645983b7", "MessageStatus": TwilioMessageStatus.DELIVERED, "SmsSid": "SM72111b3c0d874816afb433508aa0b9bd", "SmsStatus": TwilioMessageStatus.DELIVERED, "From": "+19852601833", "To": "+380631435565", "MessageSid": "SM72111b3c0d874816afb433508aa0b9bd", "AccountSid": "ABCDEFGHIJKLM", } data = "&".join([f"{k}={v}" for k, v in data.items()]) response = client.post(url, data=data, content_type="application/x-www-form-urlencoded") assert response.status_code == 200 data = { "ApiVersion": "2010-04-01", "MessagingServiceSid": "MGf42c458be7652b111856493a645983b7", "MessageStatus": TwilioMessageStatus.FAILED, "SmsSid": "SM72111b3c0d874816afb433508aa0b9bd", "SmsStatus": TwilioMessageStatus.FAILED, "From": "+19852601833", "To": "+380631435565", "ErrorCode": "21704", "MessageSid": "SM72111b3c0d874816afb433508aa0b9bd", "AccountSid": "ABCDEFGHIJKLMNPQ", # invalid account } data = "&".join([f"{k}={v}" for k, v in data.items()]) response = client.post(url, data=data, content_type="application/x-www-form-urlencoded") assert response.status_code == 400 # Lack of args data = { "AccountSid": "ABCDEFGHIJKLM", } data = "&".join([f"{k}={v}" for k, v in data.items()]) response = client.post(url, data=data, content_type="application/x-www-form-urlencoded") assert response.status_code == 400 def compare_response(response_data: dict, validate_date: dict or list): if isinstance(validate_date, list): validate_date = next(d for d in validate_date if d["id"] == response_data["id"]) for k, v in response_data.items(): assert validate_date[k] == v def check_get(client, responses: List[dict]): url = url_for("apollo_api_blueprint.application_markets_get_post") response = client.get(url) assert response.status_code == 200 assert len(response.json) == len(responses) for i in range(len(responses)): compare_response(response.json[i], responses) def test_app_markets_editing(db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) items1, items2 = get_app_markets_data() responses = get_app_markets_responses() response1, response2 = responses apollo_factories.MarketFactory.create(**items1) url = url_for("apollo_api_blueprint.application_markets_get_post") # check basic get of 1 item response = client.get(url) assert response.status_code == 200 assert len(response.json) == 1 compare_response(response.json[0], response1) # check post, cache reset, genres and globalization coping genres1 = [ apollo_factories.GenreFactory.create(market_id=items1["market_id"], name="name1"), apollo_factories.GenreFactory.create(market_id=items1["market_id"], name="name2"), ] globalization1 = [ apollo_factories.GlobalizationFactory(language_code=items1["language_code"], key="key1"), apollo_factories.GlobalizationFactory(language_code=items1["language_code"], key="key2"), ] del response2["id"] data = copy.copy(response2) data["services"] = ",".join(data["services"]) response = client.post(url, json=data) assert response.status_code == 200 response2["id"] = response.json["id"] compare_response(response.json, response2) genres2 = {i.name: i for i in db_session.query(Genre).filter(Genre.market_id == response2["id"]).all()} for g1 in genres1: assert g1.name in genres2 g2 = genres2[g1.name] assert g1.last_fm_tag == g2.last_fm_tag assert g1.active == g2.active assert g1.category_id == g2.category_id assert g1.id != g2.id assert g1.market_id != g2.market_id globalization2 = { i.key: i for i in (db_session.query(Globalization).filter(Globalization.language_code == response2["cultureInfo"]).all()) } for g1 in globalization1: assert g1.key in globalization2 g2 = globalization2[g1.key] assert g1.value == g2.value assert g1.language_code != g2.language_code check_get(client, responses) url = url_for("apollo_api_blueprint.application_markets_put_delete", market_id=response1["id"]) # check put response3 = dict(response1) response3["active"] = False response3["defaultService"] = "apple" response3["gaCountryName"] = "TestCountry" response3["includeOtherPlaylists"] = True response3["name"] = "TestName" response3["services"] = ["spotify", "apple"] response3["spotifyRegionCode"] = "te-St" response3["workoutMarket"] = True data = copy.copy(response3) data["services"] = ",".join(data["services"]) response = client.put(url, json=data) assert response.status_code == 200 compare_response(response.json, response3) check_get(client, [response3, response2]) # check delete response = client.delete(url) assert response.status_code == 200 check_get(client, [response2]) genres_count = db_session.query(Genre).count() assert genres_count == 2 @pytest.mark.parametrize( "isrc_count,playlist_count,items_count,current_date,dates_count", ( (2, 10, (7, 3), date(2020, 8, 18), 2), (3, 12, (6, 4, 1), date(2020, 8, 20), 4), (4, 6, (0, 0), date(2020, 8, 20), 2), (4, 6, (2, 4), date(2020, 8, 20), 2), (4, 6, (0, 0), date(2020, 8, 20), 2), ), ) def test_tracks_hot_hits_view( isrc_count: int, playlist_count: int, items_count: tuple, current_date: date, dates_count: int, db_session, client, patch_auth_key, ): """Test for tracks_hot_hits view.""" def filter_playlist_data(data: dict) -> dict: return {k: v for k, v in data.items() if k != "latest_date"} apollo_factories.ApolloKeyValueStorageFactory.create( key="hot_hits_latest_ts", value=current_date.strftime("%Y-%m-%dT%H:%M:%S") ) isrc_list = [f"ISRC_{i}" for i in range(isrc_count)] playlist_ids = [f"PL_ID_{i}" for i in range(playlist_count)] playlist_data = { pl_id: { "playlist_id": pl_id, "market": f"m{i}", "rank": i, "latest_date": current_date - timedelta(days=i if i % 3 == 0 else 0), "region": f"region{i}", "playlist_name": f"pl{i}", "playlist_image": SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=pl_id), } for i, pl_id in enumerate(playlist_ids) } expected_outdated = { i["playlist_id"]: filter_playlist_data(i) for i in playlist_data.values() if i["latest_date"] != current_date } actual_playlist_ids = [i for i in playlist_ids if i not in expected_outdated] expected_current = defaultdict(dict) expected_previous = defaultdict(dict) for isrc_index, pl_count in enumerate(items_count): isrc = isrc_list[isrc_index] for pl_index in range(pl_count): playlist_id = playlist_ids[pl_index] entry_date = current_date - timedelta(days=(isrc_index + pl_index) * 3) latest_date = current_date - timedelta(days=isrc_index if (isrc_index + pl_index) % 3 == 0 else 0) latest_position = isrc_index * 2 previous_position = ( None if (isrc_index + pl_index) % 4 == 0 else ( (isrc_index * 2) if (isrc_index + pl_index) % 3 == 0 and (isrc_index < 3 or isrc_index % 2 != 0) else (isrc_index + 3 * (-1 if isrc_index % 2 == 0 else 1)) ) ) peak_position = isrc_index * 2 - 10 if (pl_index + isrc_index) == 0 else 0 peak_date = current_date - timedelta(days=(pl_index + isrc_index) * 2) SpotifyHotHitsPlaylistTrackStatsFactory.create( id=((isrc_index + 1) * 10 + pl_index + 1), isrc=isrc, playlist_id=playlist_id, entry_date=entry_date, latest_date=latest_date, latest_position=latest_position, previous_date=( None if (isrc_index + pl_index) % 4 == 0 else current_date - timedelta(days=1 + isrc_index) ), previous_position=previous_position, peak_position=peak_position, peak_date=peak_date, ) if playlist_id in actual_playlist_ids: data = { "playlist_id": playlist_id, "market": playlist_data[playlist_id]["market"], "rank": playlist_data[playlist_id]["rank"], "region": playlist_data[playlist_id]["region"], "entry_date": str(entry_date), "peak_position": peak_position + 1, "peak_date": str(peak_date), "playlist_name": playlist_data[playlist_id]["playlist_name"], "playlist_image": playlist_data[playlist_id]["playlist_image"], } if latest_date == current_date: data.update( { "position": latest_position + 1, "is_entry": entry_date == latest_date, "trend": None if previous_position is None else (latest_position - previous_position), } ) expected_current[isrc][playlist_id] = data else: data["exit_date"] = str(latest_date) expected_previous[isrc][playlist_id] = data generate_view_playlists( [{"playlist_id": playlist_id, **playlist_data[playlist_id]} for playlist_id in playlist_ids], record_type=ViewPlaylistTypeEnum.HH, last_date_field="latest_date", playlist_name_field="playlist_name", add_extra=False, region_code_template="rc{}", ) expected_non_featured = defaultdict(list) for isrc in isrc_list: for playlist_id in actual_playlist_ids: if playlist_id not in expected_current[isrc] and playlist_id not in expected_previous[isrc]: expected_non_featured[isrc].append(filter_playlist_data(playlist_data[playlist_id])) url = url_for("apollo_api_blueprint.gtp_tracks_hot_hits", isrc=",".join(isrc_list)) response = client.get(url) assert response.status_code == HTTPStatus.OK for item in playlist_data.values(): item["latest_date"] = str(item["latest_date"]) expected_result = { "items": [ { "isrc": isrc, "current": list(expected_current[isrc].values()), "previous": list(expected_previous[isrc].values()), "non_featured": expected_non_featured[isrc], "current_count": len(expected_current[isrc]), "avg_position": ( round(sum(i["position"] for i in expected_current[isrc].values()) / len(expected_current[isrc])) if expected_current[isrc] else None ), } for index, isrc in enumerate(isrc_list) ], "hot_hits_date": str(current_date), "playlists_by_markets": {i["market"]: i["playlist_id"] for i in playlist_data.values()}, "playlists": list(playlist_data.values()), "outdated": list(expected_outdated.values()), "regions": [i["region"] for i in playlist_data.values()], } def sort_data(data: dict) -> dict: def sort_items(item: dict, sort_key: str = "playlist_id"): for key, value in item.items(): if isinstance(value, list): item[key] = list(sorted(value, key=lambda x: x[sort_key])) data["items"] = list(sorted(data["items"], key=lambda x: x["isrc"])) data["playlists"] = list(sorted(data["playlists"], key=lambda x: x["playlist_id"])) data["outdated"] = list(sorted(data["outdated"], key=lambda x: x["playlist_id"])) data["regions"] = list(sorted(data["regions"])) for item in data["items"]: sort_items(item) return data assert sort_data(response.json) == sort_data(expected_result) @pytest.mark.parametrize( "user_id,status,viewed,version,os", ( ("user_id_1", HTTP_400_BAD_REQUEST, False, None, None), ("user_id_1", HTTP_400_BAD_REQUEST, False, "1.1.1", "windows"), ("user_id_1", HTTP_400_BAD_REQUEST, False, "1.1.1", "ios"), ("user_id_1", HTTP_400_BAD_REQUEST, False, "2.2.2", "android"), ("user_id_1", HTTP_200_OK, False, "3.3.3", "ios"), ("user_id_2", HTTP_200_OK, True, "3.3.3", "ios"), ), ) def test_version_carousel_retrieving(db_session, client, patch_auth_user, user_id, version, os, viewed, status): patch_auth_user(user_id) carousel_version_1 = apollo_factories.VersionCarouselFactory.create(os="android", version="4.4.4") carousel_version_2 = apollo_factories.VersionCarouselFactory.create(os=os, version=version) apollo_factories.UserVersionCarouselFactory.create(user_id="user_id_2", carousel_id=carousel_version_2.id) carousel_slides = [] for slide_num in range(1, 5): carousel_slides.append( apollo_factories.VersionCarouselSlideFactory.create(carousel_id=carousel_version_2.id, slide_num=slide_num) ) apollo_factories.VersionCarouselSlideFactory.create(carousel_id=carousel_version_1.id, slide_num=slide_num) url = url_for("apollo_api_blueprint.version_carousel", os="ios", version="3.3.3") response = client.get(url) assert response.status_code == status if status == HTTP_200_OK: json_data = response.json assert json_data["viewed"] == viewed assert len(json_data["slides"]) == len(carousel_slides) @pytest.mark.parametrize( "user_id,status,version,os", ( ("user_id_1", HTTP_400_BAD_REQUEST, None, None), ("user_id_1", HTTP_400_BAD_REQUEST, "1.1.1", "windows"), ("user_id_1", HTTP_400_BAD_REQUEST, "1.1.1", "ios"), ("user_id_1", HTTP_400_BAD_REQUEST, "2.2.2", "android"), ("user_id_1", HTTP_200_OK, "3.3.3", "ios"), ("user_id_2", HTTP_409_CONFLICT, "3.3.3", "android"), ), ) def test_version_carousel_mark_viewed(db_session, client, patch_auth_user, user_id, version, os, status): patch_auth_user(user_id) carousel_version_1 = apollo_factories.VersionCarouselFactory.create(os="android", version="4.4.4") carousel_version_2 = apollo_factories.VersionCarouselFactory.create(os="android", version="3.3.3") apollo_factories.VersionCarouselFactory.create(os="ios", version="3.3.3") apollo_factories.UserVersionCarouselFactory.create(user_id="user_id_2", carousel_id=carousel_version_1.id) apollo_factories.UserVersionCarouselFactory.create(user_id="user_id_2", carousel_id=carousel_version_2.id) carousel_slides = [] for slide_num in range(1, 5): carousel_slides.append( apollo_factories.VersionCarouselSlideFactory.create(carousel_id=carousel_version_2.id, slide_num=slide_num) ) apollo_factories.VersionCarouselSlideFactory.create(carousel_id=carousel_version_1.id, slide_num=slide_num) url = url_for("apollo_api_blueprint.version_carousel") response = client.post(url, data=json.dumps(dict(os=os, version=version)), content_type="application/json") assert response.status_code == status @pytest.mark.parametrize( "user_id,status,_cache_mode,header_cache_mode", ( ("user_id_1", HTTP_200_OK, "regular", None), ("user_id_1", HTTP_200_OK, "update", "ignore"), ("user_id_1", HTTP_200_OK, "ignore", "ignore"), ("user_id_1", HTTP_200_OK, "ignore", "regular"), ("user_id_1", HTTP_200_OK, "ignore", "update"), ("user_id_1", HTTP_200_OK, None, "regular"), ("user_id_1", HTTP_200_OK, None, "update"), ), ) def test_cache_mode_version_carousel( mocker, db_session, client, patch_auth_user, user_id, status, _cache_mode, header_cache_mode ): patch_auth_user(user_id) mocker.patch("src.legacy.apollo_api.views.get_version_carousel", return_value={"id": 1, "slides": "test"}) mock_get_redis_value = mocker.patch("src.legacy.redis_db.decorators.get_redis_value", return_value=(None, "key")) mock_set_redis_value = mocker.patch("src.legacy.redis_db.decorators.set_redis_value") url = url_for("apollo_api_blueprint.version_carousel", os="ios", version="1.1.1", _cache_mode=_cache_mode) response = client.get(url, headers=({"Cache-Mode": header_cache_mode} if header_cache_mode else None)) assert response.status_code == status if not _cache_mode: if header_cache_mode == "ignore": assert mock_get_redis_value.call_count == 0 assert mock_set_redis_value.call_count == 0 else: assert mock_get_redis_value.call_count == 1 assert mock_set_redis_value.call_count == 1 if _cache_mode == "ignore": assert mock_get_redis_value.call_count == 0 assert mock_set_redis_value.call_count == 0 if _cache_mode in ("update", "regular"): assert mock_get_redis_value.call_count == 1 assert mock_set_redis_value.call_count == 1 @pytest.mark.parametrize( "vendor,response_items", ( ( "spotify", [ { "country_code": "nl", "followers": 766598, "id": "5aHawERps0AMmMLU1KHvv6", "image_url": "https://dhfsk7jl2g0km.cloudfront.net/" + "playlist-5aHawERps0AMmMLU1KHvv6.jpeg", "name": "Spinnin' Records Top 100", "owner": "Spinnin\u2019 Records", "personalized": True, "vendor": "spotify", }, { "country_code": "nl", "followers": 274669, "id": "3PCZEuLpGKpsEHeaQ6cKEN", "image_url": "https://dhfsk7jl2g0km.cloudfront.net/" + "playlist-3PCZEuLpGKpsEHeaQ6cKEN.jpeg", "name": "Armada Music Top 100", "owner": "armadamusicofficial", "personalized": False, "vendor": "spotify", }, ], ), ( "apple", [ { "country_code": "us", "id": "pl.043a2c9876114d95a4659988497567be", "image_url": "https://is2-ssl.mzstatic.com/image/thumb/Features128/v4/96/58/30/" + "96583032-0093-639c-f350-4e71e6979c1b/source/640x640cc.jpeg", "name": "Top 100: Japan", "owner": None, "vendor": "apple", }, { "country_code": "us", "id": "pl.d25f5d1181894928af76c85c967f8f31", "image_url": "https://is5-ssl.mzstatic.com/image/thumb/Features128/v4/af/00/27/" + "af00278c-3e4b-85ae-b7bb-2432b7d48cd7/source/640x640cc.jpeg", "name": "Top 100: Global", "owner": None, "vendor": "apple", }, ], ), ), ) def test_playlists_search_view(mocker, db_session, client, patch_auth_user, user_id, vendor, response_items): patch_auth_user(user_id) mocker_search = mocker.patch.object(elasticsearch.Elasticsearch, "search") mocked_ping = mocker.patch.object(elasticsearch.Elasticsearch, "ping") mocked_ping.return_value = True SpotifyPersonalizedPlaylistFactory.create(playlist_id="5aHawERps0AMmMLU1KHvv6") if vendor == VENDOR_SPOTIFY: mocker_search.return_value = { "hits": { "hits": [ { "_source": { "buzzCategoryId": 4, "countryCode": "nl", "description": "The best of dance music in one chart by Spinnin\u2019 Records", "duration": 17377, "followers": 766598, "image": "playlist-5aHawERps0AMmMLU1KHvv6.jpeg", "name": "Spinnin' Records Top 100", "playlistUri": "spotify:playlist:5aHawERps0AMmMLU1KHvv6", "public": True, "trackCount": 100, "trackLatestAdded": "2020-06-05T13:51:55Z", "updateDate": "2020-06-10T21:06:44Z", "user": "spinninrecordsofficial", "userDisplayName": "Spinnin\u2019 Records", } }, { "_source": { "buzzCategoryId": 4, "countryCode": "nl", "description": "The world's biggest and chart topping dance", "duration": 20485, "followers": 274669, "image": "playlist-3PCZEuLpGKpsEHeaQ6cKEN.jpeg", "name": "Armada Music Top 100", "playlistUri": "spotify:playlist:3PCZEuLpGKpsEHeaQ6cKEN", "public": True, "trackCount": 100, "trackLatestAdded": "2020-06-15T06:44:30Z", "updateDate": "2020-06-16T07:25:22Z", "user": "armadamusicofficial", } }, ] } } elif vendor == VENDOR_APPLE: mocker_search.return_value = { "hits": { "hits": [ { "_source": { "artworkUrl": "https://is2-ssl.mzstatic.com/image/thumb/Features128/v4/96/58/30/" + "96583032-0093-639c-f350-4e71e6979c1b/source/{w}x{h}cc.jpeg", "countryCode": "us", "name": "Top 100: Japan", "playlistId": "pl.043a2c9876114d95a4659988497567be", "streams56Days": 18285123, "updateDate": "2020-06-11T05:17:34Z", } }, { "_source": { "artworkUrl": "https://is5-ssl.mzstatic.com/image/thumb/Features128/v4/af/00/27/" + "af00278c-3e4b-85ae-b7bb-2432b7d48cd7/source/{w}x{h}cc.jpeg", "countryCode": "us", "name": "Top 100: Global", "playlistId": "pl.d25f5d1181894928af76c85c967f8f31", "streams56Days": 6380439, "updateDate": "2020-06-11T07:21:49Z", } }, ] } } else: mocker_search.return_value = {"hits": {"hits": []}} url = url_for("apollo_api_blueprint.playlists_search_elastic", vendor=vendor, query="test") response = client.get(url) assert response.status_code == HTTP_200_OK assert response.json["items"] == response_items @pytest.mark.parametrize( "type,vendor,response_items", ( ( "track", "spotify", [ { "id": "aaa", "uri": "spotify:track:aaa", "timestamp": "2021-02-09T10:00:00", "type": "track", "vendor": "spotify", } ], ), ( "track", None, [ {"id": "123", "uri": "123", "timestamp": "2021-02-09T11:00:00", "type": "track", "vendor": "apple"}, { "id": "aaa", "uri": "spotify:track:aaa", "timestamp": "2021-02-09T10:00:00", "type": "track", "vendor": "spotify", }, ], ), ( "track", "apple", [{"id": "123", "uri": "123", "timestamp": "2021-02-09T11:00:00", "type": "track", "vendor": "apple"}], ), ( "playlist", "spotify", [ { "id": "www", "uri": "spotify:playlist:www", "timestamp": "2021-02-09T10:02:00", "type": "playlist", "vendor": "spotify", }, { "id": "qqq", "uri": "spotify:playlist:qqq", "timestamp": "2021-02-09T10:01:00", "type": "playlist", "vendor": "spotify", }, ], ), ( "playlist", None, [ { "id": "pl.ttt", "uri": "pl.ttt", "timestamp": "2021-02-09T12:00:00", "type": "playlist", "vendor": "apple", }, { "id": "www", "uri": "spotify:playlist:www", "timestamp": "2021-02-09T10:02:00", "type": "playlist", "vendor": "spotify", }, { "id": "qqq", "uri": "spotify:playlist:qqq", "timestamp": "2021-02-09T10:01:00", "type": "playlist", "vendor": "spotify", }, ], ), ( "playlist", "apple", [ { "id": "pl.ttt", "uri": "pl.ttt", "timestamp": "2021-02-09T12:00:00", "type": "playlist", "vendor": "apple", } ], ), ), ) def test_search_history(db_session, client, patch_auth_user, user_id, type, vendor, response_items): patch_auth_user(user_id) apollo_factories.ApolloRecentSearchFactory.create( user_id="6784fa8b-d468-4237-a181-1b9ce2759561", created_at=datetime(2021, 2, 9, 10, 0, 3), uri="789", source="applemusic", search_type=2, ) apollo_factories.ApolloRecentSearchFactory.create( user_id="6784fa8b-d468-4237-a181-1b9ce2759515", created_at=datetime(2021, 2, 9, 10, 0, 2), uri="spotify:playlist:hhh", source="spotify", search_type=3, ) apollo_factories.ApolloRecentSearchFactory.create( user_id="6784fa8b-d468-4237-a181-1b9ce2759561", created_at=datetime(2021, 2, 9, 10, 0, 1), uri="1234", source="applemusic", search_type=2, ) apollo_factories.ApolloRecentSearchFactory.create( user_id=user_id, uri="123", search_type=2, created_at=datetime(2021, 2, 9, 11, 0, 0), source="applemusic" ) apollo_factories.ApolloRecentSearchFactory.create( user_id=user_id, uri="pl.ttt", search_type=0, created_at=datetime(2021, 2, 9, 12, 0, 0), source="applemusic" ) apollo_factories.ApolloRecentSearchFactory.create( user_id=user_id, uri="spotify:playlist:qqq", search_type=3, created_at=datetime(2021, 2, 9, 10, 1, 0), source="spotify", ) apollo_factories.ApolloRecentSearchFactory.create( user_id=user_id, uri="spotify:playlist:www", search_type=3, created_at=datetime(2021, 2, 9, 10, 2, 0), source="spotify", ) apollo_factories.ApolloRecentSearchFactory.create( user_id=user_id, uri="spotify:track:aaa", search_type=2, created_at=datetime(2021, 2, 9, 10, 0, 0), source="spotify", ) url = url_for("apollo_api_blueprint.search_history", type=type, vendor=vendor) response = client.get(url) assert response.status_code == HTTP_200_OK assert response.json["items"] == response_items @pytest.mark.parametrize( "params,status,result", ( ({"type_id": 1}, HTTPStatus.BAD_REQUEST, None), ({"date": "2021-02-01"}, HTTPStatus.BAD_REQUEST, None), ({"type_id": 1, "date": "2021-02-01"}, HTTPStatus.NOT_FOUND, None), ( {"type_id": 1, "date": "2021-02-02"}, HTTPStatus.OK, {"tracks": [{"id": "id11", "isrc": "isrc11"}, {"id": "id12", "isrc": "isrc12"}], "gtp_date": "2021-02-02"}, ), ({"type_id": 3, "date": "2021-02-03"}, HTTPStatus.NOT_FOUND, None), ( {"type_id": 3, "date": "2021-02-06"}, HTTPStatus.OK, {"tracks": [{"id": "id31", "isrc": "isrc31"}, {"id": "id32", "isrc": "isrc32"}], "gtp_date": "2021-02-04"}, ), ), ) def test_gtp_tracks_history_view(params, status, result, db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) start_date = date(2021, 2, 1) for i in range(1, 4): apollo_factories.ApolloGlobalTrackPrioritiesHistoryFactory.create( id=i, type_id=i, date=(start_date + timedelta(days=i)) ) for j in range(1, 3): apollo_factories.ApolloGlobalTrackPrioritiesHistoryTrackFactory.create( id=(10 * i + j), history_id=i, track_id=f"id{i}{j}", isrc=f"isrc{i}{j}" ) url = url_for("apollo_api_blueprint.gtp_tracks_history_view", **params) response = client.get(url) assert response.status_code == status if status == HTTPStatus.OK: response_data = response.json response_data["tracks"] = sorted(response_data["tracks"], key=lambda x: x["id"]) assert response_data == result def create_gtp_test_data(dates: Iterable[Iterable[int]]): for isrc_index, isrc_dates in enumerate(dates): for index, day in enumerate(isrc_dates): obj_id = 10 * (isrc_index + 1) + index + 1 apollo_factories.ApolloGlobalTrackPrioritiesHistoryFactory.create(id=obj_id, date=date(2021, 2, day)) apollo_factories.ApolloGlobalTrackPrioritiesHistoryTrackFactory.create( id=obj_id, history_id=obj_id, isrc=f"isrc{isrc_index + 1}" ) @pytest.mark.parametrize( "params,status,dates,result", ( ({"date": "2021-02-01"}, HTTPStatus.BAD_REQUEST, ((), ()), None), ({"isrc": "isrc1,isrc2"}, HTTPStatus.BAD_REQUEST, ((), ()), None), ({"isrc": "isrc1,isrc2", "date": "2021-02-01"}, HTTPStatus.OK, ((1,), ()), {"isrc1": 1}), ({"isrc": "isrc1", "date": "2021-02-10"}, HTTPStatus.OK, ((1, 7, 8), ()), {"isrc1": 2}), ({"isrc": "isrc1", "date": "2021-02-02"}, HTTPStatus.OK, ((1, 7, 8), ()), {"isrc1": 1}), ({"isrc": "isrc1", "date": "2021-02-10"}, HTTPStatus.OK, ((1, 7), ()), {"isrc1": 2}), ( {"isrc": "isrc1,isrc2", "date": "2021-02-28"}, HTTPStatus.OK, ((1, 15, 21), (14, 22)), {"isrc1": 2, "isrc2": 2}, ), ), ) def test_gtp_tracks_weeks_view(params, status, dates, result, db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) create_gtp_test_data(dates) url = url_for("apollo_api_blueprint.gtp_tracks_weeks_view", **params) response = client.get(url) assert response.status_code == status if status == HTTPStatus.OK: assert response.json == result @pytest.mark.parametrize( "params,status,dates,result", ( ({"date": "2021-02-01"}, HTTPStatus.BAD_REQUEST, ((),), None), ({"isrc": "isrc1,isrc2"}, HTTPStatus.BAD_REQUEST, ((),), None), ({"isrc": "isrc1,isrc2", "date": "2021-02-22"}, HTTPStatus.OK, ((1, 8, 22),), ["isrc1"]), ({"isrc": "isrc1,isrc2", "date": "2021-02-08"}, HTTPStatus.OK, ((1, 8, 22),), []), ({"isrc": "isrc1,isrc2", "date": "2021-02-01"}, HTTPStatus.OK, ((1, 8, 22),), []), ({"isrc": "isrc1,isrc2", "date": "2021-02-23"}, HTTPStatus.OK, ((1, 8, 22), (15, 22)), ["isrc1"]), ({"isrc": "isrc1,isrc2", "date": "2021-02-24"}, HTTPStatus.OK, ((1, 8, 22), (8, 22)), ["isrc1", "isrc2"]), ({"isrc": "isrc1,isrc2", "date": "2021-02-24"}, HTTPStatus.OK, ((1, 7), (7, 21)), ["isrc1", "isrc2"]), ), ) def test_gtp_tracks_reentry_view(params, status, dates, result, db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) create_gtp_test_data(dates) url = url_for("apollo_api_blueprint.gtp_tracks_reentry_view", **params) response = client.get(url) assert response.status_code == status if status == HTTPStatus.OK: assert response.json == result @pytest.mark.parametrize( "has_value,value,result,include", ( (True, None, None, None), (True, "", None, None), (True, "2021-02-01T03:06:07", "2021-02-01", None), (True, "2021-02-01T03:06:07", "2021-02-01T03:06:07", "time"), ), ) def test_gtp_hot_hits_latest_date_view(has_value, value, result, include, db_session, client, user_id, patch_auth_user): patch_auth_user(user_id) if has_value: apollo_factories.ApolloKeyValueStorageFactory.create(key="hot_hits_latest_ts", value=value) if include == "time": apollo_factories.ApolloKeyValueStorageFactory.create(key="hh_job_last_run_ts", value=value + ".974750") response = client.get(url_for("apollo_api_blueprint.gtp_hot_hits_lastest_date_view", include=include)) assert response.status_code == HTTPStatus.OK assert response.json == result @pytest.mark.parametrize( "user_id,status", ( ("user1", HTTP_404_NOT_FOUND), ("user2", HTTP_404_NOT_FOUND), ("user3", HTTP_200_OK), ("user4", HTTP_404_NOT_FOUND), ("user5", HTTP_404_NOT_FOUND), ), ) def test_nps_survey_retrieving(db_session, client, patch_auth_user, patch_date_today, user_id, status): patch_auth_user(user_id) survey_1 = apollo_factories.NPSSurveyFactory.create( name="Survey 1", id=1, active=True, start_date=date(2020, 5, 1), end_date=date(2020, 5, 31) ) survey_2 = apollo_factories.NPSSurveyFactory.create( name="Survey 2", id=2, active=False, start_date=date(2021, 5, 1), end_date=date(2021, 5, 31) ) survey_3 = apollo_factories.NPSSurveyFactory.create( name="Survey 3", id=3, active=True, start_date=date(2021, 5, 1), end_date=date(2021, 5, 31) ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_3.id, question_num=1, message="question #1", type=NPSQuestionType.RATING, labels=None ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_3.id, question_num=3, message="question #3", type=NPSQuestionType.TEXT, labels=None ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_3.id, question_num=2, message="question #2", type=NPSQuestionType.RATING, labels='{"key": "value"}', ) apollo_factories.NPSSurveyUserFactory.create(user_id="user1", survey_id=survey_1.id) apollo_factories.NPSSurveyUserFactory.create(user_id="user2", survey_id=survey_2.id) apollo_factories.NPSSurveyUserFactory.create(user_id="user3", survey_id=survey_3.id) apollo_factories.NPSSurveyUserFactory.create(user_id="user4", survey_id=survey_3.id) apollo_factories.NPSSurveyUserFactory.create(user_id="user5", survey_id=survey_3.id) active_survey_updated_at = datetime(2021, 5, 10, 12, 0, 0) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_3.id, user_id="user3", skips_num=2, status=NPSResultStatus.SKIPPED, updated_at=active_survey_updated_at, ) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_3.id, user_id="user4", status=NPSResultStatus.FINISHED, updated_at=datetime(2021, 5, 10, 12, 0, 0), ) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_3.id, user_id="user5", skips_num=4, status=NPSResultStatus.SKIPPED ) url = url_for("apollo_api_blueprint.nps_survey_view") response = client.get(url) assert response.status_code == status if response.status_code != HTTP_200_OK: return response_dict = { "questions": [ {"labels": None, "message": "question #1", "step": 1, "type": "rating"}, {"labels": {"key": "value"}, "message": "question #2", "step": 2, "type": "rating"}, {"labels": None, "message": "question #3", "step": 3, "type": "text"}, ], "survey": {"active": True, "end_date": "2021-05-31", "id": "3", "start_date": "2021-05-01"}, "user_response": {"skips_num": 2, "status": "skipped", "updated_at": active_survey_updated_at.timestamp()}, } assert response.json == response_dict @pytest.mark.parametrize("survey_id,status", ((9, HTTP_404_NOT_FOUND), (2, HTTP_404_NOT_FOUND), (1, HTTP_200_OK))) def test_nps_survey_results_retrieving(db_session, client, patch_auth_user, patch_date_today, survey_id, status): patch_auth_user("user1") survey_1 = apollo_factories.NPSSurveyFactory.create( name="Survey 1", id=1, active=True, start_date=date(2020, 5, 1), end_date=date(2020, 5, 31) ) survey_2 = apollo_factories.NPSSurveyFactory.create( name="Survey 2", id=2, active=False, start_date=date(2021, 5, 1), end_date=date(2021, 5, 31) ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_1.id, question_num=1, message="question #1", type=NPSQuestionType.RATING, labels=None ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_1.id, question_num=3, message="question #3", type=NPSQuestionType.TEXT, labels=None ) apollo_factories.NPSSurveyQuestionFactory.create( survey_id=survey_1.id, question_num=2, message="question #2", type=NPSQuestionType.RATING, labels='{"key": "value"}', ) apollo_factories.NPSSurveyUserFactory.create(user_id="user1", survey_id=survey_1.id) apollo_factories.NPSSurveyUserFactory.create(user_id="user2", survey_id=survey_2.id) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_1.id, user_id="user3", skips_num=2, status=NPSResultStatus.SKIPPED, updated_at=datetime(2021, 5, 10, 12, 0, 0), answers='[{"step": 1, "value": null}, {"step": 2, "value": null}, {"step": 3, "value": null}]', ) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_1.id, user_id="user4", status=NPSResultStatus.FINISHED, updated_at=datetime(2021, 5, 10, 12, 0, 0), answers='[{"step": 1, "value": 5}, {"step": 2, "value": "test"}, {"step": 3, "value": 10}]', ) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_1.id, user_id="user4", status=NPSResultStatus.PARTIAL, updated_at=datetime(2021, 5, 10, 12, 0, 0), answers='[{"step": 1, "value": 5}, {"step": 2, "value": None}, {"step": 3, "value": 10}]', ) apollo_factories.NPSSurveyResultFactory.create( survey_id=survey_1.id, user_id="user5", skips_num=4, status=NPSResultStatus.SKIPPED, updated_at=datetime(2021, 5, 10, 13, 0, 0), answers='[{"step": 1, "value": null}, {"step": 2, "value": null}, {"step": 3, "value": null}]', ) url = url_for("apollo_api_blueprint.nps_survey_results_view", survey_id=survey_id) response = client.get(url) assert response.status_code == status @pytest.mark.parametrize( "user_id, spotify_id, apple_id, isrc, status, market", ( ("user_id_1", "67BtfxlNbhBmCDR2L2l8qd", 1556766751, "USSM12100531", HTTP_200_OK, "US"), ("user_id_2", "57BtfxlNbhBmCDR2L2l8qd", 2556766751, "OSSM12100531", HTTP_200_OK, None), ("user_id_1", "67BtfxlNbhBmCDR2L2l8qd", None, "USSM12100531", HTTP_200_OK, "US"), ("user_id_2", "57BtfxlNbhBmCDR2L2l8qd", None, "OSSM12100531", HTTP_200_OK, None), ("user_id_1", "67BtfxlNbhBmCDR2L2l8qd", None, None, HTTP_200_OK, "US"), ("user_id_2", "57BtfxlNbhBmCDR2L2l8qd", None, None, HTTP_200_OK, None), ("user_id_1", "67BtfxlNbhBmCDR2L2l8qd", 1556766751, None, HTTP_200_OK, "US"), ("user_id_2", "57BtfxlNbhBmCDR2L2l8qd", 2556766751, None, HTTP_200_OK, None), (None, "67BtfxlNbhBmCDR2L2l8qd", 1556766751, "USSM12100531", HTTP_401_UNAUTHORIZED, None), ), ) def test_get_compared_tracks( mocker, client, patch_auth_user, spotify_track_response, compared_tracks_response, status, user_id: (str, None), spotify_id: str, apple_id: int, isrc: str, market: (str, None), ): patch_auth_user(user_id) track_list = [dict(spotify_id=spotify_id, apple_id=apple_id, isrc=isrc)] key = keys.get_key(keys.COMPARED_TRACKS, user_id) redis_client.setex(key, keys.COMPARED_TRACKS_TTL, json.dumps(track_list)) mocker_check_is_sony = mocker.patch("src.legacy.apollo_api.util.common.check_is_sony") mocker_check_is_sony.return_value = [] mocker_vendor_get_tracks = mocker.patch.object(VendorAPI, "get_tracks") mocker_vendor_get_tracks.return_value = [ spotify_track_response(track_id=spotify_id, add_markets=market is not None) ] expected_response = [ compared_tracks_response(spotify_id=spotify_id, apple_id=apple_id, isrc=isrc, add_markets=market is not None) ] url = url_for("apollo_api_blueprint.compared_tracks", market=market) response = client.get(url) assert response.status_code == status if response.status_code == HTTP_200_OK: res = response.get_json() assert res == expected_response @pytest.mark.parametrize( "user_id, spotify_id, apple_id, isrc, status, market", ( ("user_id_1", "67BtfxlNbhBmCDR2L2l8qd", 1556766751, "USSM12100531", HTTP_400_BAD_REQUEST, "US"), (None, "67BtfxlNbhBmCDR2L2l8qd", None, None, HTTP_401_UNAUTHORIZED, "US"), ("user_1", "67BtfxlNbhBmCDR2L2l8qd", None, None, HTTP_201_CREATED, "US"), ("user_1", None, 1556766751, None, HTTP_201_CREATED, None), ("user_1", "HTTP_409_CONFLICT", None, None, HTTP_409_CONFLICT, "US"), ), ) def test_post_compared_tracks( db_session, client, patch_auth_user, status, mocker, spotify_track_response, user_id: (str, None), spotify_id: (str, None), apple_id: (int, None), isrc: (str, None), market: (str, None), ): patch_auth_user(user_id) mocker_check_is_sony = mocker.patch("src.legacy.apollo_api.util.common.check_is_sony") mocker_check_is_sony.return_value = [] mocker_vendor_get_tracks = mocker.patch.object(VendorAPI, "get_track") mocker_vendor_get_tracks.return_value = spotify_track_response( track_id="HTTP_409_CONFLICT", add_markets=market is not None ) factories.AppleMusicSongFactory.create(id=1556766751, isrc="USSM12100531") SpotifyTrack2Factory(id="67BtfxlNbhBmCDR2L2l8qd", name="song1", isrc="USSM12100531") track_list = [dict(spotify_id="HTTP_409_CONFLICT", apple_id=409, isrc="409")] key = keys.get_key(keys.COMPARED_TRACKS, user_id) redis_client.setex(key, keys.COMPARED_TRACKS_TTL, json.dumps(track_list)) request_data = json.dumps( {k: v for k, v in dict(spotify_id=spotify_id, apple_id=apple_id).items() if v is not None} ) url = url_for("apollo_api_blueprint.compared_tracks", market=market) response = client.post(url, data=request_data, content_type="application/json") assert response.status_code == status @pytest.mark.parametrize( "user_id, spotify_id, apple_id, status", ( ("user_1", None, None, HTTP_400_BAD_REQUEST), ("user_1", "67BtfxlNbhBmCDR2L2l8qd", 1556766751, HTTP_400_BAD_REQUEST), ("user_1", "67BtfxlNbhBmCDR2L2l8qd", None, HTTP_200_OK), ("user_1", None, 1556766751, HTTP_200_OK), (None, None, 1556766751, HTTP_401_UNAUTHORIZED), ), ) def test_delete_compared_tracks( client, patch_auth_user, user_id: (str, None), spotify_id: (str, None), apple_id: (int, None), status ): patch_auth_user(user_id) url = url_for("apollo_api_blueprint.compared_tracks") response = client.delete( url, json={**({"spotify_id": spotify_id} if spotify_id else {}), **({"apple_id": apple_id} if apple_id else {})}, content_type="application/json", ) assert response.status_code == status @pytest.mark.parametrize( "storage_key,expected_status,expected_result", ( (None, HTTPStatus.BAD_REQUEST, None), ("abcd", HTTPStatus.NOT_FOUND, None), ("key1", HTTPStatus.OK, ""), ("key2", HTTPStatus.OK, "test_text"), ("date_key", HTTPStatus.OK, "2021-02-01"), ), ) def test_key_value_storage_view( storage_key, expected_status, expected_result, db_session, client, user_id, patch_auth_user ): patch_auth_user(user_id) key_value_list = {"key1": "", "key2": "test_text", "date_key": "2021-02-01"} for key, value in key_value_list.items(): apollo_factories.ApolloKeyValueStorageFactory.create(key=key, value=value) response = client.get(url_for("apollo_api_blueprint.key_value_storage_view", key=storage_key)) assert response.status_code == expected_status if expected_status == HTTPStatus.OK: assert response.json == expected_result @pytest.mark.parametrize( "isrc_list,expected_result", ( ( ["GBARL2000346", "USQX92004148"], {"GBARL2000346": "1var6aw8gXxihraYsvJEA8", "USQX92004148": "6AGOKlMZWLCaEJGnaROtF9"}, ), ( ["USRC12100544", "USSM11901898"], {"USRC12100544": "0IakguBBeTUsAILmugkEam", "USSM11901898": "0GBOCuNj99GKUS9ylDP8c1"}, ), (["USSM12101148"], {"USSM12101148": "0xqHkO5570zlGpLpCnd4MG"}), ), ) def test_isrc_to_id_relation(db_session, user_id, patch_auth_user, client, isrc_list, expected_result): patch_auth_user(user_id) isrc_to_id = { "GBARL2000346": "1var6aw8gXxihraYsvJEA8", "USQX92004148": "6AGOKlMZWLCaEJGnaROtF9", "USRC12100544": "0IakguBBeTUsAILmugkEam", "USSM11901898": "0GBOCuNj99GKUS9ylDP8c1", "USSM12101148": "0xqHkO5570zlGpLpCnd4MG", } for track_isrc, track_id in isrc_to_id.items(): SpotifyTrack2Factory(isrc=track_isrc, id=track_id) url = url_for("apollo_api_blueprint.isrc_to_id_relation", isrc_list=isrc_list) response = client.get(url) assert response.status_code == 200 response_data = response.json assert response_data == expected_result @pytest.mark.parametrize( "track_ids, status, expected_response", ( ( ["spotify_track_id_1"], HTTP_200_OK, { "spotify_track_id_1": { "artist_name": "Panic! At The Disco", "id": "spotify_track_id_1", "isrc": "isrc_1", "name": "Emperor's New Clothes", "sony_track": None, } }, ), (["spotify_track_id_2"], HTTP_200_OK, {}), ), ) def test_track_id_to_track_info(client, db_session, patch_auth_user, track_ids, status, expected_response): track_1 = SpotifyTrack2Factory(id="spotify_track_id_1", name="Emperor's New Clothes", isrc="isrc_1") artist_1 = SpotifyArtistFactory(id="artist_id_1", name="Panic! At The Disco") SpotifyTrackArtistFactory(track_id=track_1.id, artist_id=artist_1.id) url = url_for("apollo_api_blueprint.track_id_to_track_info", vendor="spotify", track_ids=track_ids) response = client.get(url) assert response.status_code == HTTPStatus.OK assert response.json == expected_response @pytest.mark.parametrize( "params,track_data,expected_status,expected_result", ( ( {"isrc": "QZK6P2113983", "region": "as", "include": ["regions"]}, get_hh_track(isrc="QZK6P2113983", limit=3, with_name=True), HTTPStatus.OK, { "count": 3, "previous": None, "next": None, "items": [ { "playlist_id": "pl_2", "market": "tw", "top_market": 7, "isrc": "QZK6P2113983", "entry_date": (date.today() - timedelta(days=5)).strftime("%Y-%m-%d"), "region": "Asia", "position": 5, "previous_position": 8, "latest_date": (date.today() - timedelta(days=1)).strftime("%Y-%m-%d"), "peak_position": 5, "playlist_name": "plname_2", "image_url": "https://dhfsk7jl2g0km.cloudfront.net/playlist-pl_2.jpeg", "code": "tw", "name": "TW", "position_change": 3, "added_days": 4, "is_new": False, }, { "playlist_id": "pl_3", "market": "id", "top_market": 8, "isrc": "QZK6P2113983", "entry_date": (date.today() - timedelta(days=5)).strftime("%Y-%m-%d"), "region": "Asia", "position": 7, "previous_position": 10, "latest_date": (date.today() - timedelta(days=1)).strftime("%Y-%m-%d"), "peak_position": 7, "playlist_name": "plname_3", "image_url": "https://dhfsk7jl2g0km.cloudfront.net/playlist-pl_3.jpeg", "code": "id", "name": "ID", "position_change": 3, "added_days": 4, "is_new": False, }, { "playlist_id": "pl_4", "market": "my", "top_market": 9, "isrc": "QZK6P2113983", "entry_date": (date.today() - timedelta(days=5)).strftime("%Y-%m-%d"), "region": "Asia", "position": 9, "previous_position": 12, "latest_date": (date.today() - timedelta(days=1)).strftime("%Y-%m-%d"), "peak_position": 9, "playlist_name": "plname_4", "image_url": "https://dhfsk7jl2g0km.cloudfront.net/playlist-pl_4.jpeg", "code": "my", "name": "MY", "position_change": 3, "added_days": 4, "is_new": False, }, ], "positions_avg": 7, "positions_top_total": 3, "regions": [ {"name": "All markets", "code": "all", "total": 9, "placements": 3}, {"name": "Asia", "code": "as", "total": 6, "placements": 3}, ], "total": 6, }, ), ), ) def test_hh_tracks_view( params, track_data, expected_status, expected_result, db_session, user_id, client, patch_auth_user ): patch_auth_user(user_id) entry_date = datetime.now() - timedelta(days=5) latest_date = datetime.now() - timedelta(days=1) apollo_factories.ApolloKeyValueStorageFactory.create( key="hot_hits_latest_ts", value=latest_date.strftime("%Y-%m-%dT%H:%M:%S") ) generate_view_playlists(track_data, ViewPlaylistTypeEnum.HH) for index, item in enumerate(track_data): SpotifyHotHitsPlaylistTrackStatsFactory.create( isrc=item["isrc"], playlist_id=item["playlist_id"], latest_position=item["position"], previous_position=item["previous_position"], peak_position=item["position"], latest_date=latest_date, previous_date=latest_date, entry_date=entry_date, peak_date=latest_date, ) response = client.get(url_for("apollo_api_blueprint.hot-hits-tracks", **params)) assert response.status_code == expected_status if expected_status == HTTPStatus.OK: assert response.json == expected_result