from datetime import datetime, timedelta import pytest from apollo_utils.job.clients.base.client import BaseApiClient from charts_push_messages import handler from charts_push_messages.client.utils import get_feed_messages_data, post_push_messages from charts_push_messages.processing.check_incoming_events import check_incoming_events from charts_push_messages.processing.get_push_messages import PushMessageFactory, get_push_messages from charts_push_messages.schemas import FeedMessageCreatedSchema, FeedMessageCreatedWithDataSchema from charts_push_messages.utils.common import check_received_event, handle_check_received_events, deserialize_event, \ handle_events_deserialization from charts_push_messages.utils.logger_messages import get_feed_message_created, \ get_feed_message_with_data, get_deserializing_logger_message from tests.helpers import create_incoming_event, create_feed_message_with_data, get_broken_data @pytest.mark.parametrize( "event,expected_result", ( ({"created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 500}, True), ({"created_at": datetime.utcnow() - timedelta(seconds=10), "ttl": 1}, False), ), ) def test_check_recieved_event(event, expected_result): result = check_received_event(event) assert result == expected_result @pytest.mark.parametrize( "events,expected_result", ( ( [ {"id": 1, "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 500}, {"id": 2, "created_at": datetime.utcnow() - timedelta(seconds=10), "ttl": 1} ], [1] ), ( [ {"id": 1, "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 500}, {"id": 2, "created_at": datetime.utcnow() - timedelta(seconds=10), "ttl": 500} ], [1, 2] ), ( [ {"id": 1, "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 1}, {"id": 2, "created_at": datetime.utcnow() - timedelta(seconds=10), "ttl": 1} ], [] ), ), ) def test_handle_check_received_events(test_logger, events, expected_result): result = handle_check_received_events(logger=test_logger, events=events) assert result == expected_result @pytest.mark.parametrize( "broken,expected_result,schema,create_data_to_check", ( # (False, create_incoming_event(), FeedMessageCreatedSchema(), create_incoming_event), # (True, None, FeedMessageCreatedSchema(), create_incoming_event), (False, create_incoming_event(), FeedMessageCreatedWithDataSchema(), create_feed_message_with_data), (True, None, FeedMessageCreatedWithDataSchema(), create_feed_message_with_data), ), ) def test_deserialize_event(test_logger, broken, expected_result, schema, create_data_to_check): data_to_check = create_data_to_check() if not broken else get_broken_data() result = deserialize_event( logger=test_logger, schema=schema, event=data_to_check ) assert result if not broken else result is expected_result @pytest.mark.parametrize( "incoming_events,expected_result_length,schema", ( ([create_incoming_event(), create_incoming_event()], 2, FeedMessageCreatedSchema()), ([create_incoming_event(), get_broken_data()], 1, FeedMessageCreatedSchema()), ([get_broken_data(), get_broken_data()], 0, FeedMessageCreatedSchema()), ([create_feed_message_with_data(), create_feed_message_with_data()], 2, FeedMessageCreatedWithDataSchema()), ([create_feed_message_with_data(), get_broken_data()], 1, FeedMessageCreatedWithDataSchema()), ([get_broken_data(), get_broken_data()], 0, FeedMessageCreatedWithDataSchema()), ), ) def test_handle_events_deserialization(test_logger, incoming_events, expected_result_length, schema): result = handle_events_deserialization(logger=test_logger, schema=schema, events=incoming_events) assert len(result) == expected_result_length @pytest.mark.parametrize( "events, add_broken, expected_result_length", ( ( [ {"id": "1", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 500}, {"id": "2", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 1}, {"id": "3", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 500} ], True, 2 ), ( [ {"id": "1", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 1}, {"id": "2", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 1}, {"id": "3", "created_at": datetime.utcnow() - timedelta(seconds=5), "ttl": 1} ], False, 0 ), ), ) def test_check_incoming_events(test_logger, events, add_broken, expected_result_length): events_to_check = [ create_incoming_event( event_id=i["id"], event_created_at=i["created_at"].isoformat(), event_ttl=i["ttl"] ) for i in events] if add_broken: events_to_check.append({"id": "broken_event_id"}) result = check_incoming_events(logger=test_logger, events=events_to_check) assert len(result) == expected_result_length @pytest.mark.parametrize( "schema,expected_result,create_data", ( (FeedMessageCreatedSchema(), get_feed_message_created, create_incoming_event), (FeedMessageCreatedWithDataSchema(), get_feed_message_with_data, create_feed_message_with_data) ), ) def test_get_deserializing_logger_message(schema, expected_result, create_data): result = get_deserializing_logger_message(schema) assert result == expected_result data = create_data() err = {"key": "i am an error"} assert result(data, err) @pytest.mark.parametrize( "dsp,type,devices", ( ("spotify", "starred_track_top_chart_entry", 1), ("apple", "starred_track_top_chart_entry", 1), ("spotify", "starred_track_top_chart_exit", 1), ("apple", "starred_track_top_chart_exit", 1), ("spotify", "starred_track_top_chart_move", 1), ("apple", "starred_track_top_chart_move", 1), ("spotify", "starred_track_top_chart_entry", 2), ("apple", "starred_track_top_chart_entry", 3), ("spotify", "starred_track_top_chart_exit", 2), ("apple", "starred_track_top_chart_exit", 3), ("spotify", "starred_track_top_chart_move", 2), ("apple", "starred_track_top_chart_move", 3), ), ) def test_get_push_message(test_logger, dsp, type, devices): deserialized_feed_message = deserialize_event( logger=test_logger, schema=FeedMessageCreatedWithDataSchema(), event=create_feed_message_with_data(meta_dsp=dsp, meta_type=type, number_of_devices=devices) ) expected = PushMessageFactory(**deserialized_feed_message).get_push_messages() result = get_push_messages([deserialized_feed_message]) assert result == expected def get_url(request): url = request.url index = url.find("api/") return url[index:] @pytest.mark.parametrize( "ids_count,expected_call_count,url_to_call,feed,callable_to_check", ( (1, 1, "api/service/messages/feed/list/", False, get_feed_messages_data), (102, 2, "api/service/messages/feed/list/", False, get_feed_messages_data), (1, 1, "api/service/messages/push/", True, post_push_messages), (102, 2, "api/service/messages/push/", True, post_push_messages), ), ) def test_user_data_client(mocker, ids_count, expected_call_count, url_to_call, feed, callable_to_check): ids_list = list(range(ids_count)) dicts_list = [{} for _ in ids_list] return_result = {"ok": list().append({}) for _ in ids_list} if feed else {"data": ids_list} def client_mock(*args, **kwargs): url = get_url(args[0]) if url == url_to_call: return return_result client_mocker = mocker.patch.object(BaseApiClient, "_send_request", side_effect=client_mock) callable_to_check(message_ids=ids_list) if not feed else callable_to_check(push_messages=dicts_list) assert client_mocker.call_count == expected_call_count @pytest.mark.parametrize( "dsp,type,broken_event", ( ("spotify", "starred_track_top_chart_entry", None), ("apple", "starred_track_top_chart_entry", None), ("spotify", "starred_track_top_chart_exit", None), ("apple", "starred_track_top_chart_exit", None), ("spotify", "starred_track_top_chart_move", None), ("apple", "starred_track_top_chart_move", None), ("spotify", "starred_track_top_chart_entry", get_broken_data()), ("apple", "starred_track_top_chart_entry", get_broken_data()), ("spotify", "starred_track_top_chart_exit", get_broken_data()), ("apple", "starred_track_top_chart_exit", get_broken_data()), ("spotify", "starred_track_top_chart_move", get_broken_data()), ("apple", "starred_track_top_chart_move", get_broken_data()), ), ) def test_handler(test_logger, mocker, dsp, type, broken_event): incoming_event = create_incoming_event(meta_dsp=dsp, meta_type=type) events = [incoming_event] if broken_event: events.append(broken_event) full_message = create_feed_message_with_data( message_id=incoming_event["id"], meta_type=incoming_event["meta"]["type"], meta_dsp=incoming_event["meta"]["dsp"] ) def client_mock(*args, **kwargs): url = get_url(args[0]) if url == "api/service/messages/feed/list/": return {"data": [full_message]} if url == "api/service/messages/push/": return {"ok": [{}], "failed": [{}]} mocker.patch.object(BaseApiClient, "_send_request", side_effect=client_mock) handler(logger=test_logger, event=events)