# pylint: disable=redefined-outer-name,unused-argument,invalid-name,too-many-locals # pylint: disable=protected-access import json import typing as t from copy import deepcopy from datetime import datetime, timezone from http import HTTPStatus from os.path import join as p_join from unittest import mock from unittest.mock import MagicMock, Mock, PropertyMock, patch import boto3 import pytest from am_http_client.client import Client as AppleMusicClient from dapd_db_schema.schemas.workflow import Playlist from jsonschema.exceptions import ValidationError from moto import mock_secretsmanager from moto.secretsmanager.models import SecretsManager from requests import Response from requests.exceptions import HTTPError from smelog.factory import SmeBoundLogger from spotipy import SpotifyException from dapd_public_api_scraper.config.entity import DataSourceCredentials from dapd_public_api_scraper.const import INSERT_BATCH_SIZE_DEFAULT_VALUE from dapd_public_api_scraper.entity import base as base_entity from dapd_public_api_scraper.entity import spotify as spotify_entity from dapd_public_api_scraper.repository.postgres import Postgres from dapd_public_api_scraper.repository.validator import Validator from dapd_public_api_scraper.service.scraper import apple_music as apple_music_scraper from dapd_public_api_scraper.service.scraper import spotify as spotify_scraper from dapd_public_api_scraper.service.workflowdb import apple_music as apple_music_workflowdb from dapd_public_api_scraper.service.workflowdb import spotify as spotify_workflowdb class RecursiveNextCallMock: def __init__(self, return_values_list: t.List[t.Dict[str, t.Any]]): self.call_count = 0 self.return_values_list = return_values_list def __call__(self, item: t.Dict[str, t.Any]) -> t.Optional[t.Dict[str, t.Any]]: if item.get('next'): self.call_count += 1 return self.return_values_list[self.call_count - 1] return None @pytest.fixture(scope='function') def secrets_manager(aws_credentials): with mock_secretsmanager(): yield boto3.client('secretsmanager') @pytest.fixture(scope='function') def apple_music_secrets_manager(secrets_manager): secrets_manager.create_secret( Name='delphi/dev/apd/apple-music', SecretString=json.dumps({ 'APPLE_MUSIC_TEAM_ID': 'sme_team', 'APPLE_MUSIC_KEY_ID': 'sme_key', 'APPLE_MUSIC_SECRET_KEY': 'aaa\\nbbb\\n' }) ) yield secrets_manager @pytest.fixture(scope='function') def apple_music_get_playlist_response(s3_backup_apple_music_dir): with open( p_join(s3_backup_apple_music_dir, 'playlist/pl.3343f35d630744efa7e01543e415ffeb.json'), 'r', encoding='utf-8', ) as f: raw_report = json.load(f) response = raw_report['item'] yield response @pytest.fixture(scope='function') def spotify_secrets_manager(secrets_manager): secrets_manager.create_secret( Name='delphi/dev/apd/spotify', SecretString=json.dumps({ 'client_id': 'sme_client_id', 'client_secret': 'sme_client_secret', }) ) secrets_manager.create_secret( Name='delphi/dev/apd/spotify_2', SecretString=json.dumps({ 'client_id': 'random_user_client_id', 'client_secret': 'random_user_client_secret', }) ) yield secrets_manager @pytest.fixture(scope='function') def spotify_get_playlist_response(s3_backup_spotify_dir): with open( p_join(s3_backup_spotify_dir, 'playlist/034aPYRs3TW4tEpsDyokB6.json'), 'r', encoding='utf-8', ) as f: raw_report = json.load(f) response = raw_report['item'] yield response @patch('dapd_public_api_scraper.service.scraper.apple_music.AppleMusicScraper.get_client') def test_apple_music_playlist( apple_music_client_getter: MagicMock, logger: SmeBoundLogger, apple_music_secrets_manager: SecretsManager, apple_music_get_playlist_response: t.Dict[str, t.Any] ): # yapf: disable expected_response = apple_music_get_playlist_response apple_music_client_mock = MagicMock(spec=AppleMusicClient) apple_music_client_mock.get_playlist.return_value = expected_response apple_music_client_getter.return_value = apple_music_client_mock db_item = MagicMock(Playlist()) db_item.snapshot_id = '12345' pg = MagicMock(Postgres) workflowdb = MagicMock( apple_music_workflowdb.AppleMusicPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) workflowdb.get.return_value = db_item credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='', ) validator = MagicMock(Validator) scraper = apple_music_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, apple_music_secrets_manager, ) task = base_entity.Task('playlist', 'us') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item.item == expected_response assert is_item_valid is True # Validator switched off # validator.validate.assert_called_once_with(expected_response) workflowdb.get.assert_called_once_with(task.id, task.storefront) apple_music_client_mock.get_playlist.assert_called_with( 'playlist', 'us', include=['tracks', 'curator'], raw=True, ) @patch('dapd_public_api_scraper.service.scraper.apple_music.AppleMusicScraper.get_client') def test_apple_music_playlist_404( apple_music_client_getter: MagicMock, logger: SmeBoundLogger, apple_music_secrets_manager: SecretsManager, ): error = HTTPError() error_response = Response() error_response.status_code = HTTPStatus.NOT_FOUND error.response = error_response apple_music_client_mock = MagicMock(spec=AppleMusicClient) apple_music_client_mock.get_playlist.side_effect = error apple_music_client_getter.return_value = apple_music_client_mock pg = MagicMock(Postgres) workflowdb = MagicMock( apple_music_workflowdb.AppleMusicPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='', ) scraper = apple_music_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, apple_music_secrets_manager, ) task = base_entity.Task('playlist', 'us') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item is None assert is_item_valid is False validator.validate.assert_not_called() workflowdb.get.assert_not_called() workflowdb.set_removed.assert_called_once_with(task) apple_music_client_mock.get_playlist.assert_called_with( 'playlist', 'us', include=['tracks', 'curator'], raw=True, ) @patch('dapd_public_api_scraper.service.scraper.apple_music.AppleMusicScraper.get_client') def test_apple_music_playlist_not_changed( applemusic_client_getter: MagicMock, logger: SmeBoundLogger, apple_music_secrets_manager: SecretsManager, apple_music_get_playlist_response: t.Dict[str, t.Any], ): expected_response = apple_music_get_playlist_response apple_music_client = Mock() apple_music_client.get_playlist.return_value = expected_response applemusic_client_getter.return_value = apple_music_client db_item = MagicMock(Playlist()) db_item.snapshot_id = '4362cf460096cb81ac80caf9b4770bcf3787656195d5e2dc09b27620fed52df9' pg = MagicMock(Postgres) workflowdb = MagicMock( apple_music_workflowdb.AppleMusicPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) workflowdb.get.return_value = db_item validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='', ) scraper = apple_music_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, apple_music_secrets_manager, ) task = base_entity.Task('playlist', 'us') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item is None assert is_item_valid is False # Validator switched off # validator.validate.assert_called_once_with(expected_response) workflowdb.get.assert_called_once_with(task.id, task.storefront) apple_music_client.get_playlist.assert_called_with( 'playlist', 'us', include=['tracks', 'curator'], raw=True, ) @patch('dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.user_api_client') @patch('dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.api_client') def test_spotify_playlist( spotify_client_mock: MagicMock, spotify_user_client_mock: MagicMock, logger: SmeBoundLogger, spotify_secrets_manager: SecretsManager, spotify_get_playlist_response: t.Dict[str, t.Any], ): expected_response = spotify_get_playlist_response spotify_client_mock.playlist.return_value = expected_response spotify_user_client_mock.playlist.return_value = expected_response is_personalized = True db_item = MagicMock(Playlist()) db_item.snapshot_id = '1234' db_item.is_personalized = is_personalized pg = MagicMock(Postgres) workflowdb = MagicMock( spotify_workflowdb.SpotifyPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) workflowdb.get.return_value = db_item validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='user_secret_key', ) scraper = spotify_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, spotify_secrets_manager, ) task = base_entity.Task('playlist', 'global') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item.item == expected_response assert is_item_valid is True assert result_item.meta.is_personalized is is_personalized # Validator switched off # validator.validate.assert_called_once_with(expected_response) workflowdb.get.assert_called_once_with(task.id, task.storefront) spotify_client_mock.playlist.assert_called_with('playlist') spotify_user_client_mock.playlist.assert_not_called() @patch('dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.api_client') def test_spotify_playlist_404( spotify_client_mock: MagicMock, logger: SmeBoundLogger, spotify_secrets_manager: SecretsManager, ): spotify_client_mock.playlist.side_effect = SpotifyException( http_status=HTTPStatus.NOT_FOUND, code=-1, msg='dummy message', reason=None, headers=None, ) pg = MagicMock(Postgres) workflowdb = MagicMock( spotify_workflowdb.SpotifyPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='user_secret_key', ) scraper = spotify_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, spotify_secrets_manager, ) task = base_entity.Task('playlist', 'us') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item is None assert is_item_valid is False validator.validate.assert_not_called() workflowdb.get.assert_not_called() workflowdb.set_removed.assert_called_once_with(task) spotify_client_mock.playlist.assert_called_with('playlist') @patch('dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.user_api_client') @patch('dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.api_client') def test_spotify_playlist_not_changed( spotify_client_mock: MagicMock, spotify_user_client_mock: MagicMock, logger: SmeBoundLogger, spotify_secrets_manager: SecretsManager, spotify_get_playlist_response: t.Dict[str, t.Any] ): # yapf: disable expected_response = spotify_get_playlist_response spotify_client_mock.playlist.return_value = expected_response spotify_user_client_mock.playlist.return_value = expected_response db_item = MagicMock(Playlist()) db_item.snapshot_id = 'OSw1NmI5ZmI1ODBhNWIwMTExMTNiN2U2MjQ3NGIzNTk1NDM4ODI5NGM4' pg = MagicMock(Postgres) workflowdb = MagicMock( spotify_workflowdb.SpotifyPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) workflowdb.get.return_value = db_item validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='user_secret_key', ) scraper = spotify_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, spotify_secrets_manager, ) task = base_entity.Task('playlist_id', 'global') result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item is None assert is_item_valid is False # Validator switched off # validator.validate.assert_called_once_with(expected_response) workflowdb.get.assert_called_once_with(task.id, task.storefront) spotify_client_mock.playlist.assert_called_with('playlist_id') spotify_user_client_mock.playlist.assert_not_called() @pytest.mark.skip('Temporary switched off personalized logic') def test_spotify_playlist_personalized( logger, spotify_secrets_manager, spotify_get_playlist_response ): spotify_get_playlist_response['owner']['id'] = 'spotify' spotify_get_playlist_response[ 'name'] = 'playlist which name is not started with This is:, Inspired by or Radio' response = spotify_get_playlist_response user_response = deepcopy(response) # change list of tracks in user playlist user_response['tracks']['items'].pop(0) user_response['snapshot_id'] += 'v2' with patch( 'dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.api_client', new_callable=PropertyMock ) as api_client_getter: api_client = Mock() api_client.playlist.return_value = response type(api_client_getter).return_value = api_client with patch( 'dapd_public_api_scraper.service.scraper.spotify.SpotifyScraper.user_api_client', new_callable=PropertyMock ) as user_api_client_getter: user_api_client = Mock() user_api_client.playlist.return_value = user_response type(user_api_client_getter).return_value = user_api_client db_item = MagicMock(Playlist()) db_item.snapshot_id = '1234' db_item.is_personalized_expired_at = None pg = MagicMock(Postgres) workflowdb = MagicMock( spotify_workflowdb.SpotifyPlaylist( logger, pg, INSERT_BATCH_SIZE_DEFAULT_VALUE, None, ) ) workflowdb.get.return_value = db_item validator = MagicMock(Validator) credentials = DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='user_secret_key', ) scraper = spotify_scraper.PlaylistScraper( logger, workflowdb, validator, credentials, spotify_secrets_manager, ) task = base_entity.Task('playlist', 'global', save_tracklist=True) result_task, result_item, is_item_valid = scraper.get_item(task) assert result_task == task assert result_item.item == response assert is_item_valid is True assert result_item.meta.is_personalized # Validator switched off # assert validator.validate.call_count == 2 # validator.validate.assert_has_calls([mock.call(response), mock.call(user_response)]) workflowdb.get.assert_called_once_with(task.id, task.storefront) api_client.playlist.assert_called_with('playlist') user_api_client.playlist.assert_called_with('playlist') @pytest.mark.parametrize( 'pl1,pl2,expected', [([ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], [ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], False), ([ { 'added_at': '2020-05-28T18:31:53Z', 'track': { 'id': '1' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], [ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], True), ([ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, ], [ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], True), ([ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '1' } }, ], [ { 'added_at': '2020-05-28T18:30:53Z', 'track': { 'id': '2' } }, ], True)] ) def test_spotify_check_playlist_changed(pl1, pl2, expected): assert spotify_scraper.PlaylistScraper.check_playlist_changed(pl1, pl2) == expected @pytest.mark.freeze_time('2021-01-21') @pytest.mark.parametrize( 'owner_id,name,is_personalized_expired_at,expected', [ ('apple_music', 'Somename', None, False), ('spotify', 'This is:', datetime(2021, 1, 2, 0), False), ('spotify', 'This is ', datetime(2021, 1, 2, 0), False), ('spotify', 'Inspired by ', datetime(2021, 1, 2, 0), False), ('spotify', ' Radio', datetime(2021, 1, 2, 0), False), ('spotify', 'Smooth jazz', datetime(2021, 1, 20, 0, tzinfo=timezone.utc), True), ] ) def test_spotify_can_be_personalized( owner_id: str, name: str, is_personalized_expired_at, expected ): assert expected == spotify_scraper.PlaylistScraper.can_be_personalized( owner_id, name, is_personalized_expired_at ) @pytest.mark.parametrize( 'playlist, provider_response, expected, expected_call_count', [ ( {'tracks': {'items': []}}, [{'items': []}], {'tracks': {'items': []}}, 0 ), ( {'tracks': {'items': [], 'next': ''}}, [{'items': [2]}], {'tracks': {'items': [], 'next': ''}}, 0 ), ( {'tracks': {'items': [], 'next': 'url'}}, [{'items': [2]}], {'tracks': {'items': [2], 'next': 'url'}}, 1 ), ( {'tracks': {'items': [1], 'next': 'url'}}, [{'items': [2], 'next': ''}], {'tracks': {'items': [1, 2], 'next': 'url'}}, 1 ), ( {'tracks': {'items': [1], 'next': 'url'}}, [{'items': [2], 'next': 'another_url'}, {'items': [3], 'next': ''}], {'tracks': {'items': [1, 2, 3], 'next': 'url'}}, 2 ), ] ) # yapf: disable def test_spotify__extend_playlist_tracks( playlist: t.Dict[str, t.Any], provider_response: t.List[t.Dict[str, t.Any]], expected: t.Dict[str, t.Any], expected_call_count: int, ): api_client = mock.Mock() api_client.next.side_effect = RecursiveNextCallMock(provider_response) result = spotify_scraper.PlaylistScraper._extend_playlist_tracks(api_client, playlist) assert playlist == expected assert playlist is result assert api_client.next.call_count == expected_call_count assert api_client.next.has_calls([expected_call_count * mock.call('another_url')]) @pytest.mark.skip('TODO: fix validation fails') @pytest.mark.parametrize( 'can_be_personalized, is_db_item_personalized, has_playlist_changed,' 'call_returns, validate_returns, extend_returns,' 'expected', [ (False, False, False, lambda *x: {'tracks': {'items': []}}, None, None, False), (False, False, True, lambda *x: {'tracks': {'items': []}}, None, None, False), (False, True, False, lambda *x: {'tracks': {'items': []}}, None, None, True), (False, True, True, lambda *x: {'tracks': {'items': []}}, None, None, True), (True, True, True, lambda *x: {'tracks': {'items': []}}, None, None, True), (True, True, False, lambda *x: {'tracks': {'items': []}}, None, None, False), (True, False, True, lambda *x: {'tracks': {'items': []}}, None, None, True), (True, False, False, lambda *x: {'tracks': {'items': []}}, None, None, False), (True, False, False, SpotifyException(500, 500, ''), None, None, None), (True, False, False, SpotifyException(404, 404, ''), None, None, None), (True, False, False, lambda *x: {'tracks': {'items': []}}, ValidationError(''), None, None), ( True, False, False, lambda *x: {'tracks': {'items': []}}, None, SpotifyException(404, 404, ''), None ), ] ) # yapf: disable @patch('dapd_public_api_scraper.service.scraper.spotify.PlaylistScraper.check_playlist_changed') @patch('dapd_public_api_scraper.service.scraper.spotify.PlaylistScraper._extend_playlist_tracks') @patch('dapd_public_api_scraper.service.scraper.spotify.PlaylistScraper.can_be_personalized') @patch('dapd_public_api_scraper.service.scraper.spotify.PlaylistScraper._call') @patch('dapd_public_api_scraper.service.scraper.spotify.PlaylistScraper.user_api_client') def test_get_personalized_flag( user_api_client_mock: MagicMock, call_method_mock: MagicMock, can_be_personalized_mock: MagicMock, extend_playlist_tracks_mock: MagicMock, check_playlist_changed_mock: MagicMock, can_be_personalized: bool, is_db_item_personalized: bool, has_playlist_changed: bool, call_returns, validate_returns, extend_returns, expected: t.Optional[bool], ): # pylint: disable=too-many-arguments call_method_mock.side_effect = call_returns can_be_personalized_mock.return_value = can_be_personalized extend_playlist_tracks_mock.side_effect = extend_returns check_playlist_changed_mock.return_value = has_playlist_changed validator_mock = mock.MagicMock(spec=Validator) validator_mock.validate.side_effect = validate_returns scraper = spotify_scraper.PlaylistScraper( logger=mock.MagicMock(spec=SmeBoundLogger), workflowdb=mock.MagicMock(spec=spotify_workflowdb.SpotifyPlaylist), validator=validator_mock, credentials=DataSourceCredentials( sme_secret_key='sme_secret_key', user_secret_key='user_secret_key', ), secrets_manager=mock.MagicMock(spec=SecretsManager), ) task = base_entity.Task(id='playlist_id', storefront='global') new_personalized_flag = scraper.get_personalized_flag( task=task, item=spotify_entity.Playlist( item={'tracks': {'items': []}}, meta=base_entity.PlaylistMeta( item_id=task.id, item_type='playlist', item_ts='2020-05-28T18:30:53', item_storefront=task.storefront, data_source='spotify', app_version='test_version', ), ), raw_data={'owner': {'id': '123'}, 'name': 'playlist_name'}, db_item=Playlist(is_personalized=is_db_item_personalized), ) # yapf: disable if new_personalized_flag is not None: assert new_personalized_flag == expected else: assert new_personalized_flag is expected