import pytest from apollo_playlists_messages.constants import EventCode from tests.data.base import parsed_device from tests.data.events import create_parsed_full_event from tests.data.favorites import favorite_playlists_accounts, favorite_tracks_accounts from tests.data.messages import starred_exit_message, top_exit_message from tests.data.utils import set_os_event def parsed_exit_events(): return [ create_parsed_full_event( 1, EventCode.EXIT, playlist_seed=1, track_seed=1, country_code=["us", "gb", "au"], top_country_code=["us", "au"], with_date=True ), create_parsed_full_event( 2, EventCode.EXIT, playlist_seed=2, track_seed=2, country_code=["us"], top_country_code=["us", "global", "gb"], with_date=True ), create_parsed_full_event( 3, EventCode.EXIT, playlist_seed=3, track_seed=3, country_code=["fr", "us"], top_country_code=["us"], with_date=True, extra={"previous_position": 0} ), create_parsed_full_event( 4, EventCode.EXIT, playlist_seed=4, track_seed=2, country_code=["mx"], top_country_code=[], with_date=True ), create_parsed_full_event( 5, EventCode.EXIT, playlist_seed=5, track_seed=5, country_code=["us", "ca"], top_country_code=["us", "ca"], with_date=True, extra={"previous_position": 0} ), create_parsed_full_event( 6, EventCode.EXIT, playlist_seed=6, track_seed=5, country_code=[""], top_country_code=["us"], with_date=True, extra={"track.id": "id_6"} ), ] @pytest.mark.parametrize( "input_events, ignored_ids, expected_result_len, expected_result", ( (parsed_exit_events(), (), 6, [ starred_exit_message( 1, 1, 1, dsp="apple", country_code="us", account_id=1, devices=[parsed_device(11), parsed_device(12)] ), starred_exit_message( 2, 2, 2, dsp="spotify", country_code="us", account_id=4, devices=[parsed_device(41)] ), top_exit_message( 3, 3, 3, dsp="apple", country_code="global", account_id=1, position=1, devices=[parsed_device(11), parsed_device(12)] ), top_exit_message( 5, 5, 5, dsp="apple", country_code="ca", account_id=5, position=1, devices=[parsed_device(51)] ), top_exit_message( 5, 5, 5, dsp="apple", country_code="us", account_id=5, position=1, devices=[parsed_device(51)] ), top_exit_message( 6, 6, 5, dsp="spotify", country_code="us", account_id=5, extra={"data.content.track.id": "id_6", "data.content.playlist.country_code": ""}, devices=[parsed_device(51)] ) ]), (parsed_exit_events(), ("1", "3", "5", "6"), 1, [ starred_exit_message( 2, 2, 2, dsp="spotify", country_code="us", account_id=4, devices=[parsed_device(41)] ), ]), ([], (), 0, None), ), ) def test_exit_handler(mocker, test_logger, input_events, ignored_ids, expected_result_len, expected_result): from apollo_playlists_messages.handlers.common.handler import handler def get_favorites(ids, entity_type): if entity_type == "track": return favorite_tracks_accounts() elif entity_type == "playlist": return favorite_playlists_accounts() gate_api_mock = mocker.MagicMock() gate_api_mock.get_favorites_with_accounts.side_effect = get_favorites result = handler(input_events, set(ignored_ids), gate_api_mock, test_logger, EventCode.EXIT) assert len(result) == expected_result_len result.sort(key=lambda i: (i["event_id"], i["account_id"], i["meta"]["country_code"])) if expected_result is not None: for i, item in enumerate(expected_result): result_item = result[i] result_item["data"]["recipient"]["devices"].sort(key=lambda i: i["id"]) assert result[i] == item assert gate_api_mock.get_favorites_with_accounts.call_count == 2 @pytest.mark.parametrize( "event, input_events, ignored_ids, expected_result_len, expected_result", ( (set_os_event("playlists_exit"), parsed_exit_events(), (), 6, [ starred_exit_message( 1, 1, 1, dsp="apple", country_code="us", account_id=1, devices=[parsed_device(11), parsed_device(12)] ), starred_exit_message( 2, 2, 2, dsp="spotify", country_code="us", account_id=4, devices=[parsed_device(41)] ), top_exit_message( 3, 3, 3, dsp="apple", country_code="global", account_id=1, position=1, devices=[parsed_device(11), parsed_device(12)] ), top_exit_message( 5, 5, 5, dsp="apple", country_code="ca", account_id=5, position=1, devices=[parsed_device(51)] ), top_exit_message( 5, 5, 5, dsp="apple", country_code="us", account_id=5, position=1, devices=[parsed_device(51)] ), top_exit_message( 6, 6, 5, dsp="spotify", country_code="us", account_id=5, extra={"data.content.track.id": "id_6", "data.content.playlist.country_code": ""}, devices=[parsed_device(51)] ) ]), (set_os_event("playlists_exit"), parsed_exit_events(), ("1", "3", "5", "6"), 1, [ starred_exit_message( 2, 2, 2, dsp="spotify", country_code="us", account_id=4, devices=[parsed_device(41)] ), ]), (set_os_event("playlists_exit"), [], (), 0, None), ), ) def test_exit_handler_settings_v2( event, mocker, test_logger_playlists_exit, input_events, ignored_ids, expected_result_len, expected_result ): from apollo_playlists_messages.handlers.common.handler import handler def get_favorites(ids, entity_type): if entity_type == "track": return favorite_tracks_accounts(settings_version="2") elif entity_type == "playlist": return favorite_playlists_accounts(settings_version="2") gate_api_mock = mocker.MagicMock() gate_api_mock.get_favorites_with_accounts.side_effect = get_favorites result = handler(input_events, set(ignored_ids), gate_api_mock, test_logger_playlists_exit, EventCode.EXIT) assert len(result) == expected_result_len result.sort(key=lambda i: (i["event_id"], i["account_id"], i["meta"]["country_code"])) if expected_result is not None: for i, item in enumerate(expected_result): result_item = result[i] result_item["data"]["recipient"]["devices"].sort(key=lambda i: i["id"]) assert result[i] == item assert gate_api_mock.get_favorites_with_accounts.call_count == 2