import math from http import HTTPStatus as http_status import pytest from apollo_utils.core.constants.dsp import DSP from server.constants.core import ApolloEntityType from server.db.models.users import Favorites from server.db.session import db_session as session_scope from tests.factories import FavoritesFactory from tests.utils import create_accounts, check_object, set_test_db def get_dsp(i): rnd_map = { 1: DSP.APPLE.value, 2: DSP.SPOTIFY.value, 0: DSP.AMAZON.value, } return rnd_map[i % 3] def get_country_code(i): dsp = get_dsp(i) if dsp == DSP.APPLE.value: return f"c{i}" return None def get_favorites_record_data(i: int) -> dict: dsp = get_dsp(i) return { "id": f"pl_id_{i}", "vendor": dsp, **({"country_code": get_country_code(i)} if dsp == DSP.APPLE.value else {}), } def get_entity_id(i: int) -> str: return f"{get_dsp(i)}_{str(get_country_code(i))}_pl_id_{i}" def get_favorite_data(i: int, force_id: int or None = None, force_account_id: int or None = None) -> dict: return dict( id=(force_id if force_id else i), entity_id=get_entity_id(i), entity_type=(ApolloEntityType.PLAYLIST if bool(i % 4) else ApolloEntityType.TRACK), data=get_favorites_record_data(i), account_id=(force_account_id if force_account_id else math.ceil(i / 2)), ) def get_favorites(*args) -> list: return [get_favorite_data(i) for i in args] async def create_favorites(session, count: int = 4, accounts_count: int or None = None, apps_count: int or None = None): await create_accounts(session, accounts_count or math.ceil(count / 2), apps_count=apps_count or math.ceil(count / 4)) for item in get_favorites(*range(1, count + 1)): session.add(FavoritesFactory.create(**item)) await session.commit() def get_entity_params(*args) -> list: return [ { "entity_id": get_entity_id(i), "entity_type": (ApolloEntityType.PLAYLIST if bool(i % 3) else ApolloEntityType.TRACK).value } for i in args ] @pytest.mark.parametrize( "data,params,headers,status,expected_result", ( ({}, {}, {}, http_status.BAD_REQUEST, None), ({"entity_type_list": ["playlist", "track"]}, {"X-App-Slug": "app1"}, {}, http_status.BAD_REQUEST, None), ( {"entity_type_list": ["playlist"]}, {}, {"X-App-Slug": "app1"}, http_status.OK, { "count": 2, "items": [ { "entity_type": "playlist", "entity_id": "spotify_None_pl_id_2", "user_id": "user1", "account_id": 1, "data": {"id": "pl_id_2", "vendor": "spotify"} }, { "entity_type": "playlist", "entity_id": "apple_c1_pl_id_1", "user_id": "user1", "account_id": 1, "data": {"country_code": "c1", "id": "pl_id_1", "vendor": "apple"} }, ], "next": None, "previous": None } ), ( {"user_id_list": ["user1"], "entity_type_list": ["playlist"], "vendor_list": ["apple"]}, {"offset": 0, "limit": 1}, {"X-App-Slug": "app1"}, http_status.OK, {'count': 1, 'items': [{'account_id': 1, 'data': {'country_code': 'c1', 'id': 'pl_id_1', 'vendor': 'apple'}, 'entity_id': 'apple_c1_pl_id_1', 'entity_type': 'playlist', 'user_id': 'user1'}], 'next': None, 'previous': None} ), ( {"user_id_list": ["user1"], "entity_type_list": ["playlist"], "vendor_list": ["amazon"]}, {"offset": 0, "limit": 1}, {"X-App-Slug": "app1"}, http_status.OK, {'count': 1, 'items': [{'account_id': 2, 'data': {'id': 'pl_id_3', 'vendor': 'amazon'}, 'entity_id': 'amazon_None_pl_id_3', 'entity_type': 'playlist', 'user_id': 'user2'}], 'next': None, 'previous': None} ), ), ) async def test_get_favorite_playlists_by_params( data, params, headers, status, expected_result, db_session, client, auth): async with session_scope() as session: await create_favorites(session, 4) headers.update(auth) response = await client.post( "/api/v1/favorites/by-params/bulk/", headers=headers, params=params, json=data) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result @pytest.mark.parametrize( "params,headers,status,expected_result", ( ({}, {}, http_status.BAD_REQUEST, None), ( {}, {"X-App-Slug": "app1", "X-User-Id": "user2"}, http_status.OK, {'count': 1, 'items': [ { "favorites_id": 3, 'country_code': None, 'id': 'pl_id_3', 'vendor': 'amazon', "tag_id": None } ], 'next': None, 'previous': None} ), ( {"offset": 1, "limit": 1}, {"X-App-Slug": "app1", "X-User-Id": "user1"}, http_status.OK, { "count": 2, "items": [ { "favorites_id": 1, "country_code": "c1", "id": "pl_id_1", "vendor": "apple", "tag_id": None } ], "next": None, "previous": "/user-data-api/v1/favorites/playlists/?limit=1&offset=0", } ), ( {"limit": 1}, {"X-App-Slug": "app1", "X-User-Id": "user1"}, http_status.OK, { "count": 2, "items": [ { "favorites_id": 2, "country_code": None, "id": "pl_id_2", "vendor": "spotify", "tag_id": None } ], "next": "/user-data-api/v1/favorites/playlists/?limit=1&offset=1", "previous": None, } ), ), ) async def test_get_favorite_playlists_by_account(params, headers, status, expected_result, db_session, client, auth): async with session_scope() as session: await create_favorites(session, 4) headers.update(auth) response = await client.get("/api/v1/favorites/playlists/", headers=headers, params=params) assert response.status == status if status == http_status.OK: result = await response.json() result_items = result.pop("items", []) expected_items = expected_result.pop("items", []) assert result == expected_result for i, result_item in enumerate(result_items): expected_item = expected_items[i] for k in ("created_at", "updated_at"): assert result_item.pop(k) is not None assert result_item == expected_item @pytest.mark.parametrize( "headers,body,status,expected_result,expected_data", ( ({}, {}, http_status.BAD_REQUEST, None, None), ( {"X-App-Slug": "app1", "X-User-Id": "user2"}, {"id": "pl_id_1", "vendor": "apple"}, http_status.BAD_REQUEST, None, None, ), ( {"X-App-Slug": "app1", "X-User-Id": "user2"}, {"id": "pl_id_1", "vendor": "spotify", "country_code": "us"}, http_status.BAD_REQUEST, None, None, ), ( {"X-App-Slug": "app1", "X-User-Id": "user2"}, get_favorites_record_data(1), http_status.CREATED, { "data": {"country_code": "c1", "id": "pl_id_1", "vendor": "apple"}, "entity_id": "apple_c1_pl_id_1", "entity_type": "playlist", "favorites_id": 1, "count": None, "tag_id": None, }, get_favorite_data(1, force_account_id=2), ), ( {"X-App-Slug": "app1", "X-User-Id": "user1"}, get_favorites_record_data(2), http_status.CREATED, { "data": {"id": "pl_id_2", "vendor": "spotify"}, "entity_id": "spotify_None_pl_id_2", "entity_type": "playlist", "favorites_id": 1, "count": None, "tag_id": None, }, get_favorite_data(2, force_id=1, force_account_id=1) ), ( {"X-App-Slug": "app1", "X-User-Id": "user1"}, get_favorites_record_data(3), http_status.CREATED, { 'data': {'id': 'pl_id_3', 'vendor': 'amazon'}, 'entity_id': 'amazon_None_pl_id_3', "entity_type": "playlist", "favorites_id": 1, "count": None, "tag_id": None, }, get_favorite_data(3, force_id=1, force_account_id=1) ), ), ) async def test_add_playlist(headers, body, status, expected_result, expected_data, db_session, client, auth): async with session_scope() as session: await create_accounts(session, 2) headers.update(auth) response = await client.post("/api/v1/favorites/playlists/", headers=headers, json=body) assert response.status == status async with session_scope(): items, _ = await Favorites.list() assert len(items) == int(status == http_status.CREATED) if status == http_status.CREATED: result = await response.json() assert result == expected_result check_object(items[0], expected_data) @pytest.mark.parametrize( "headers,body,status", ( ({}, {}, http_status.BAD_REQUEST), ( {"X-App-Slug": "app1", "X-User-Id": "user2"}, get_favorites_record_data(1), http_status.NOT_FOUND, ), ( {"X-App-Slug": "app1", "X-User-Id": "user1"}, get_favorites_record_data(1), http_status.NO_CONTENT, ), ), ) async def test_delete_playlist_by_data(headers, body, status, db_session, client, auth): async with session_scope() as session: await create_favorites(session, 2) headers.update(auth) response = await client.delete("/api/v1/favorites/playlists/", headers=headers, json=body) assert response.status == status async with session_scope(): count = await Favorites.count() assert count == (1 if status == http_status.NO_CONTENT else 2) @pytest.mark.parametrize( "headers,entity_id,status", ( ({}, get_entity_id(1), http_status.BAD_REQUEST), ( {"X-App-Slug": "app1", "X-User-Id": "user2"}, get_entity_id(1), http_status.NOT_FOUND, ), ( {"X-App-Slug": "app1", "X-User-Id": "user1"}, get_entity_id(1), http_status.NO_CONTENT, ), ), ) async def test_delete_playlist_by_entity_id(headers, entity_id, status, db_session, client, auth): async with session_scope() as session: await create_favorites(session, 2) headers.update(auth) response = await client.delete(f"/api/v1/favorites/playlists/{entity_id}/", headers=headers) assert response.status == status async with session_scope(): count = await Favorites.count() assert count == (1 if status == http_status.NO_CONTENT else 2) @pytest.mark.parametrize( "headers,params,status,expected_result", ( ({}, {}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"user": {"id": [2, 3]}, "account": {"id": [2, 5]}}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"device": {"os": "fridge OS"}}, http_status.BAD_REQUEST, None), ({"X-App-Slug": "app_1"}, {"include": "something"}, http_status.BAD_REQUEST, None), ( {"X-App-Slug": "app_1"}, {}, http_status.OK, { "data": [ {"accounts": [{"account_id": 1, "user_id": "u_1"}], "entity_id": "pl_1", "entity_type": "playlist"}, {"accounts": [{"account_id": 1, "user_id": "u_1"}], "entity_id": "pl_2", "entity_type": "playlist"}, {"accounts": [{"account_id": 1, "user_id": "u_1"}], "entity_id": "tr_1", "entity_type": "track"}, ], }, ), ( {"X-App-Slug": "app_2"}, {"user": {"id": ["u_2", "u_3", "u_4"]}}, http_status.OK, { "data": [ { "accounts": [{"account_id": 2, "user_id": "u_2"}, {"account_id": 3, "user_id": "u_3"}], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [{"account_id": 2, "user_id": "u_2"}, {"account_id": 4, "user_id": "u_4"}], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"account": {"id": [2, 5]}}, http_status.OK, { "data": [ {"accounts": [{"account_id": 2, "user_id": "u_2"}], "entity_id": "pl_1", "entity_type": "playlist"}, {"accounts": [{"account_id": 2, "user_id": "u_2"}], "entity_id": "pl_3", "entity_type": "playlist"}, ], }, ), ( {"X-App-Slug": "app_2"}, {"device": {"is_active": False}}, http_status.OK, { "data": [ { "accounts": [{"account_id": 2, "user_id": "u_2"}], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [{"account_id": 2, "user_id": "u_2"}, {"account_id": 4, "user_id": "u_4"}], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"device": {"os": ["android"]}}, http_status.OK, { "data": [ { "accounts": [{"account_id": 2, "user_id": "u_2"}, {"account_id": 3, "user_id": "u_3"}], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [{"account_id": 2, "user_id": "u_2"}, {"account_id": 4, "user_id": "u_4"}], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"settings": {"type": "mobile", "version": "1"}, "include": ["devices"]}, http_status.OK, { "data": [ { "accounts": [ { "account_id": 4, "devices": [ {"expo_token": "token_9", "id": 9, "is_active": False, "os": "android"}], "user_id": "u_4", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "entity": {"type": "playlist", "id": ["pl_1", "pl_3"]}, "include": ["settings"], "settings": {"type": "portal"}, }, http_status.OK, { "data": [ { "accounts": [ { "account_id": 4, "settings": {"data": {"s_2": "True"}, "id": 6, "type": "portal", "version": "2"}, "user_id": "u_4", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"entity": {"type": "playlist", "id": ["pl_1", "pl_3"]}, "include": ["settings"]}, http_status.OK, { "data": [ { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, { "account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3", }, ], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, {"entity": { "type": "playlist", "id": ["pl_1", "pl_3"], "fake_field": "to be excluded" }, "include": ["settings"]}, http_status.OK, { "data": [ { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, { "account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3", }, ], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "entity": { "type": "playlist", "id": ["pl_1", "pl_3"] }, "include": ["settings"], "settings": { "versions": { "min": "1", }, }, }, http_status.OK, { "data": [ { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, { "account_id": 3, "settings": {"data": {"f_3": "v_3"}, "id": 3, "type": None, "version": "1"}, "user_id": "u_3", }, ], "entity_id": "pl_1", "entity_type": "playlist", }, { "accounts": [ { "account_id": 2, "settings": {"data": {"f_2": "v_2"}, "id": 2, "type": None, "version": "1"}, "user_id": "u_2", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ( {"X-App-Slug": "app_2"}, { "entity": { "type": "playlist", "id": ["pl_1", "pl_3"] }, "include": ["settings"], "settings": { "versions": { "min": "2", }, }, }, http_status.OK, { "data": [], }, ), ( {"X-App-Slug": "app_2"}, { "entity": {"type": "playlist", "id": ["pl_1", "pl_3"]}, "include": ["settings"], "settings": { "type": "portal", "versions": { "eq": "1", }, }, }, http_status.OK, { "data": [], }, ), ( {"X-App-Slug": "app_2"}, { "entity": {"type": "playlist", "id": ["pl_1", "pl_3"]}, "include": ["settings"], "settings": { "type": "portal", "versions": { "eq": "2", }, }, }, http_status.OK, { "data": [ { "accounts": [ { "account_id": 4, "settings": {"data": {"s_2": "True"}, "id": 6, "type": "portal", "version": "2"}, "user_id": "u_4", }, ], "entity_id": "pl_3", "entity_type": "playlist", }, ], }, ), ), ) async def test_post_service_users_favorites_list(headers, params, status, expected_result, client, auth, db_session): async with session_scope() as session: await set_test_db(session, with_favorites=True) headers.update(auth) response = await client.post("/api/service/users/favorites/list/", headers=headers, json=params) assert response.status == status if status == http_status.OK: result = await response.json() assert result == expected_result