import json from datetime import date, timedelta from typing import Callable, Dict, List from urllib.parse import quote import pytest import requests_mock from juno_email_messages import config from juno_email_messages.clients import service from juno_email_messages.clients.apollo_api import ApolloClientConfig from juno_email_messages.clients.atlas_api import AtlasClientConfig from juno_email_messages.clients.dsp_api import DspClientConfig from juno_email_messages.clients.gate_api import GateClientConfig from juno_email_messages.clients.notifications_api import NotificationsClientConfig from juno_email_messages.clients.user_data_api import UserDataClientConfig from juno_email_messages.handler import handler from juno_email_messages.logger import get_logger def get_history_info(history, is_json: bool = True, data_modifier: Callable = None) -> tuple: data = history.json() if is_json else "&".join(sorted(history.query.split("&"))) if data_modifier: data = data_modifier(data) return history.method, history.path, data def get_event(count: int = 5) -> dict: return { "Records": [{"body": json.dumps({"Message": json.dumps({"id": i})})} for i in range(1, count + 1)] } def get_message_list(index_map: Dict[int, List[int]]): event_base_list = [0] + [len(i) for i in index_map.values()] return [ { "event_id": event_base_list[index] + j, "account_id": 10 + i, "meta": { "topic": "juno_digest", "views": ["juno_email"], "user_id": f"user_id_{i}", "settings_id": 40 + i, "filter_id": f"filter_id_{i}_{j}", }, "data": { "name": f"Filter_{i}_{j}", "country_code": [f"m{i + j}"], **({"isrc_country_code": [f"m{i + j + 1}"]} if (i + j) % 2 else {}), **({"percent_change": 20 + i + j} if (i + j) % 2 else {}), **({"min_product_sale_days": 200 + i + j} if (i + j) % 3 else {}), **({"max_product_sale_days": 100 + i + j} if (i + j) % 2 == 0 else {}), }, } for index, (i, sub_index_list) in enumerate(index_map.items()) for j in sub_index_list ] def get_juno_data(count: int = 10) -> List[dict]: distributed_by_options = ("awal", "theorchard", "sme", None) return [ { "isrc": f"isrc_{i}", "latest_entry_date": f"2022-08-1{i}" if i % 4 else "2022-08-20", "spotify_top_track_meta": { "spotify_track_id": f"track_id_{i}", "track_name": f"Track{i}", "artist_names": [f"Artist{i}{j}" for j in range(1, min(i, 3))], "track_uri": f"spotify:track:track_id_{i}" if i % 4 else f"non_track_id_{i}", }, "metrics": { "totals": {"streams_days_7": 1020 * i, "streams_days_8_14": 920 * i}, "change": { "streams_change_7_to_14_days": 205 * i, "streams_change_percent_7_to_14_days": 400 * i }, "lean_back": { "streams_days_7": (810 if i > 3 else 210) * i, "streams_days_8_14": (560 if i > 4 else 110) * i, }, "lean_forward": { "streams_days_7": (210 if i > 4 else 810) * i, "streams_days_8_14": (110 if i > 3 else 560) * i, }, "lean_back_change": { "streams_change_7_to_14_days": 350 * i * (-1 if i % 4 == 0 else 1), "streams_change_percent_7_to_14_days": 41 * i * (-1 if i % 3 == 0 else 1), }, "lean_forward_change": { "streams_change_7_to_14_days": 281 * i * (-1 if i % 4 == 0 else 1), "streams_change_percent_7_to_14_days": 26 * i * (-1 if i % 3 == 0 else 1), }, "demographics": { "male": { "streams_days_7": (840 if i > 2 else 180) * i, "streams_days_8_14": (570 if i > 3 else 130) * i, }, "female": { "streams_days_7": (180 if i > 4 else 840) * i, "streams_days_8_14": (130 if i > 2 else 570) * i, }, }, }, "distributed_by": distributed_by_options[i % 4], } for i in range(1, count + 1) ] def get_filter_str(i: int, j: int) -> str: juno_date = date(2022, 8, 20) result = [ f"country_code=m{i + j}", f"limit={config.JUNO_TRACK_COUNT}", "sort_by=streams_days_7", "sort_order=desc", "distributors=awal%2ctheorchard%2csme", ] if (i + j) % 2: result.append(f"isrc_country_code=m{i + j + 1}") result.append(f"percent_change={20 + i + j}") if (i + j) % 3: result.append(f"min_product_sale_date={(juno_date - timedelta(days=200 + i + j)).isoformat()}") if (i + j) % 2 == 0: result.append(f"max_product_sale_date={(juno_date - timedelta(days=101 + i + j)).isoformat()}") return "&".join(sorted(result)) def mocked_render_template(filter_id: str, event_data: dict, juno_date: date, juno_track_list: List[dict]) -> str: return json.dumps( { "filter_id": filter_id, "event_data": event_data, "juno_date": juno_date.isoformat(), "juno_track_list": juno_track_list, } ) def format_number(value: int) -> str: if value < 1000: return str(value) if value < 1000000: return f"{round(value / 1000)}k" return f"{round(value / 1000000)}M" def get_email_result(index_map: Dict[int, List[int]], track_count: int) -> List[dict]: distributed_by_options = ("awal", "theorchard", "sme", None) return [ { "body": { "filter_id": f"filter_id_{i}_{j}", "event_data": { "name": f"Filter_{i}_{j}", "country_code": f"Ma{i + j}" if i + j < 5 else f"m{i + j}", "isrc_country_code": (f"Ma{i + j + 1}" if i + j < 4 else f"m{i + j + 1}") if (i + j) % 2 else "All", "product_sale_days": ( f"{(100 + i + j) if (i + j) % 2 == 0 else 0} Days—" + (f"{200 + i + j} Days" if (i + j) % 3 else "10 Years+") ), **({"percent_change": str(20 + i + j)} if (i + j) % 2 else {}), }, "juno_date": "2022-08-20", "juno_track_list": [ { "is_entry": k >= 4, "track_name": f"Track{k}", "artist_name": ", ".join(f"Artist{k}{l}" for l in range(1, min(k, 3))), "is_male": k > 4, "is_female": k < 3, "is_lean_back": k > 4, "is_lean_forward": k < 4, "streams": format_number(1020 * k), "is_change_negative": False, "change_value": format_number(205 * k), "change_percent": format_number(400 * k), "is_lf_change_negative": k % 4 == 0, "lf_change_value": format_number(281 * k), "lf_change_percent": format_number(26 * k), "track_id": f"track_id_{k}" if k % 4 else None, "track_page_url": quote( f"{config.APOLLO_PORTAL_URL}/spotify/track/track_id_{k}" ) if k % 4 else None, "distributed_by": distributed_by_options[k % 4], } for k in range(1, track_count + 1) ], }, "subject": f"[JUNO] Trending Tracks 08/20/22 - Filter_{i}_{j}", "to": [f"email_{i}@test.com"], } for i, sub_list in index_map.items() for j in sub_list ] def loads_body(data: List[dict]) -> List[dict]: for item in data: item["body"] = json.loads(item["body"]) return data @pytest.mark.parametrize( "event_count,message_map,email_index_list,juno_track_count,call_count,expected_map,amplitude_called", ( (0, {}, [], 3, 0, {}, 0), (3, {1: [1, 2], 2: [1]}, [1, 3], 3, 8, {1: [1, 2]}, 1), (6, {1: [1], 3: [1, 2, 3]}, [3], 5, 9, {3: [1, 2, 3]}, 1), ) ) @requests_mock.Mocker(kw="requests_mocker") def test_handler( event_count: int, message_map: Dict[int, List[int]], email_index_list: List[int], juno_track_count: int, call_count: int, expected_map: Dict[int, List[int]], amplitude_called: int, mocker, **kwargs, ): requests_mocker = kwargs["requests_mocker"] apollo_config = ApolloClientConfig() requests_mocker.register_uri( "GET", f"{apollo_config._schema}://{apollo_config._host}/api/markets/", json=[{"code": f"m{i}", "name": f"Ma{i}"} for i in range(1, 5)], ) requests_mocker.register_uri( "GET", f"{apollo_config._schema}://{apollo_config._host}/api/users/emails/", json={"items": [{"user_id": f"user_id_{i}", "email": f"email_{i}@test.com"} for i in email_index_list]}, ) atlas_config = AtlasClientConfig() requests_mocker.register_uri( "POST", f"{atlas_config._schema}://{atlas_config._host}/oauth/token", json={"access_token": "test_token"} ) dsp_config = DspClientConfig() requests_mocker.register_uri( "GET", f"{dsp_config._schema}://{dsp_config._host}/api/delphi/juno/spotify/status/daily", json={"dsp": {"spotify": {"report_date": "2022-08-20"}}}, ) gate_config = GateClientConfig() requests_mocker.register_uri( "GET", f"{gate_config._schema}://{gate_config._host}/api/charts/tracks/streams/spotify/", json=get_juno_data(juno_track_count), ) notifications_config = NotificationsClientConfig() requests_mocker.register_uri( "POST", f"{notifications_config._schema}://{notifications_config._host}/api/v1/email/send", json=[{f"email_{i}@test.com": f"task_id_{i}"} for i in range(1, 4)], ) user_data_config = UserDataClientConfig() requests_mocker.register_uri( "POST", f"{user_data_config._schema}://{user_data_config._host}/api/service/messages/list/", json={"data": get_message_list(message_map) if message_map else []}, ) mocker.patch("juno_email_messages.render.render_template", side_effect=mocked_render_template) mocked_amplitude = mocker.patch("juno_email_messages.amplitude_events.send_amplitude_events") handler(get_event(event_count), get_logger()) history = requests_mocker.request_history assert len(history) == call_count if not call_count: return assert get_history_info(history[0], is_json=True) == ( "POST", "/api/service/messages/list/", {"id": list(range(1, event_count + 1))} ) assert get_history_info(history[1], is_json=True) == ( "POST", "/oauth/token", { "audience": config.ATLAS_AUDIENCE, "grant_type": config.ATLAS_GRANT_TYPE, "client_id": atlas_config._client_id, "client_secret": atlas_config._client_secret, }, ) assert service.notifications._apikey == "Bearer test_token" assert get_history_info(history[2], is_json=False) == ( "GET", "/api/users/emails/", "&".join(f"id=user_id_{i}" for i in message_map.keys()) ) assert get_history_info(history[3], is_json=False) == ("GET", "/api/markets/", "extended=true") assert get_history_info(history[4], is_json=False) == ("GET", "/api/delphi/juno/spotify/status/daily", "") index = 5 for i, filter_list in expected_map.items(): for j in filter_list: assert get_history_info(history[index], is_json=False) == ( "GET", "/api/charts/tracks/streams/spotify/", get_filter_str(i, j) ) index += 1 assert get_history_info(history[index], is_json=True, data_modifier=loads_body) == ( "POST", "/api/v1/email/send", get_email_result(expected_map, juno_track_count) ) assert mocked_amplitude.call_count == amplitude_called