from datetime import datetime, timedelta from itertools import product import pytest from apollo_utils.job.clients.base.client import BaseApiClient from charts_feed_messages import handler from charts_feed_messages.client.utils import get_basic_messages_data, post_feed_messages from charts_feed_messages.processing.check_incoming_events import check_incoming_events from charts_feed_messages.processing.get_feed_messages import create_feed_message, get_feed_message_schema, \ get_feed_messages from charts_feed_messages.schemas import IncomingEvent, MessageEventWithData from charts_feed_messages.utils.common import check_received_event, handle_check_received_events, deserialize_event, \ handle_events_deserialization from charts_feed_messages.utils.feed_message import FeedMessageFieldsFabricator from charts_feed_messages.utils.logger_messages import get_incoming_event_schema_logger_message, \ get_event_with_data_schema_logger_message, get_event_schema_logger_message_builder from tests.helpers import create_incoming_event, get_broken_data, create_full_message_with_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(), IncomingEvent(), create_incoming_event), (True, None, IncomingEvent(), create_incoming_event), (False, create_incoming_event(), MessageEventWithData(), create_full_message_with_data), (True, None, MessageEventWithData(), create_full_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, IncomingEvent()), ([create_incoming_event(), get_broken_data()], 1, IncomingEvent()), ([get_broken_data(), get_broken_data()], 0, IncomingEvent()), ([create_full_message_with_data(), create_full_message_with_data()], 2, MessageEventWithData()), ([create_full_message_with_data(), get_broken_data()], 1, MessageEventWithData()), ([get_broken_data(), get_broken_data()], 0, MessageEventWithData()), ), ) 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", ( (IncomingEvent(), get_incoming_event_schema_logger_message, create_incoming_event), (MessageEventWithData(), get_event_with_data_schema_logger_message, create_full_message_with_data) ), ) def test_get_event_schema_logger_message_builder(schema, expected_result, create_data): result = get_event_schema_logger_message_builder(schema) assert result == expected_result data = create_data() err = {"key": "i am an error"} assert result(data, err) @pytest.mark.parametrize( "dsp,type", ( ("spotify", "starred_track_top_chart_entry"), ("apple", "starred_track_top_chart_entry"), ("spotify", "starred_track_top_chart_exit"), ("apple", "starred_track_top_chart_exit"), ("spotify", "starred_track_top_chart_move"), ("apple", "starred_track_top_chart_move"), ), ) def test_create_feed_message(test_logger, dsp, type): deserialized_message = deserialize_event( logger=test_logger, schema=MessageEventWithData(), event=create_full_message_with_data(meta_dsp=dsp, meta_type=type) ) expected = FeedMessageFieldsFabricator(message=deserialized_message) expected_chart = expected.get_chart() expected_body = expected.get_body() not_expected_meta = deserialized_message["meta"] result = create_feed_message(schema=get_feed_message_schema(), message=deserialized_message) result_chart = result["data"]["content"]["chart"] result_body = result["data"]["content"]["body"] result_meta = result["meta"] assert result_chart == expected_chart assert result_body == expected_body assert result_meta != not_expected_meta @pytest.mark.parametrize( "dsp_list,meta_type", ( ( ["apple", "spotify"], ["starred_track_top_chart_entry", "starred_track_top_chart_exit", "starred_track_top_chart_move"] ), ) ) def test_get_feed_messages(test_logger, dsp_list, meta_type): deserialized_messages = [deserialize_event( logger=test_logger, schema=MessageEventWithData(), event=message ) for message in [create_full_message_with_data( meta_dsp=dsp, meta_type=event_type ) for dsp, event_type in product(dsp_list, meta_type) ]] fabricators = [ FeedMessageFieldsFabricator(message=message) for message in deserialized_messages ] result = get_feed_messages(deserialized_messages) for fabricator, feed_message in zip(fabricators, result): assert feed_message["data"]["content"]["chart"] == fabricator.get_chart() assert feed_message["data"]["content"]["body"] == fabricator.get_body() 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/list/", False, get_basic_messages_data), (102, 2, "api/service/messages/list/", False, get_basic_messages_data), (1, 1, "api/service/messages/feed/", True, post_feed_messages), (102, 2, "api/service/messages/feed/", True, post_feed_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(feed_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_full_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/list/": return {"data": [full_message]} if url == "api/service/messages/feed/": return {"ok": [{}], "failed": [{}]} mocker.patch.object(BaseApiClient, "_send_request", side_effect=client_mock) handler(logger=test_logger, event=events)