import pytest from apollo_delphi_events.constants import EventType from tests.data.events import pack_events, event, output_event from tests.data.utils import dt_now @pytest.mark.parametrize( "input_event,expected_result_len,expected_result", ( ( pack_events([ event(1, event_type=EventType.ADDITION, country_code=["us", "fr", "au"], top_country_code=["us"], dsp="apple_music"), # outdated event(2, event_type=EventType.ADDITION, extra={"ttl": 5}, dsp="spotify") ]), 1, None ), ( pack_events([ event(1, event_type=EventType.UPDATE, country_code=["fr", "au"], top_country_code=["au"], dsp="apple_music"), event(2, event_type=EventType.ADDITION, extra={"meta.event_type": "unknown"}, dsp="spotify"), event(3, event_type=EventType.ADDITION, dsp="unknown"), event(4, event_type=EventType.ADDITION, dsp="spotify", extra={"meta.domain": "unknown"}), event(5, event_type=EventType.ADDITION, dsp="apple_music", exclude=("meta.domain",)), ]), 1, None ), ( pack_events([ event(1, event_type=EventType.UPDATE, country_code=["fr", "au"], top_country_code=["au"], dsp="apple_music", exclude=("id",)), event(2, event_type=EventType.ADDITION, exclude=("meta.event_type",)), event(3, event_type=EventType.ADDITION, exclude=("meta.dsp",)), event(5, event_type=EventType.ADDITION, exclude=("data.content.track",), dsp="apple_music",), event(6, event_type=EventType.UPDATE, exclude=("data.content.playlist.last_updated_at",)), ]), 0, None ), ( pack_events([ event(1, j=3, event_type=EventType.UPDATE, country_code=["au", "mx"], dsp="apple_music", exclude=("meta.top_country_code", "ttl")), event(2, j=4, k=6, event_type=EventType.ADDITION, country_code=["global"], top_country_code=["us", "fr", "global"]), event(3, j=5, k=9, event_type=EventType.ADDITION, country_code=["global"], dsp="apple_music", exclude=("meta.top_country_code",)), event(4, j=6, k=8, event_type=EventType.ADDITION, country_code=[], dsp="spotify", top_country_code=["us"]) ]), 4, [ event(1, j=3, event_type=EventType.UPDATE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), event(2, j=4, k=6, event_type=EventType.ADDITION, country_code=["global"], top_country_code=["us", "fr", "global"], exclude=( "app", "publisher", "meta.type", "data.content.playlist.image_url", "data.content.previous_position" ), event_type_as_value=False, serialize_dt=False), event(3, j=5, k=9, event_type=EventType.ADDITION, country_code=["global"], top_country_code=[], exclude=( "app", "publisher", "meta.type", "data.content.playlist.image_url", "data.content.previous_position" ), event_type_as_value=False, serialize_dt=False), event(4, j=6, k=8, event_type=EventType.ADDITION, country_code=[], top_country_code=["us"], exclude=( "app", "publisher", "meta.type", "data.content.playlist.image_url", "data.content.previous_position" ), event_type_as_value=False, serialize_dt=False) ] ), ( pack_events([ event(1, j=3, event_type=EventType.DROP, country_code=["au", "mx"], dsp="apple_music", exclude=("meta.top_country_code", "ttl")), ]), 1, [ event(1, j=3, event_type=EventType.DROP, country_code=["au", "mx"], top_country_code=[], exclude=( "app", "publisher", "meta.type", "data.content.playlist.image_url", "data.content.current_position" ), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), ] ), ( pack_events([ event(1, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], dsp="apple_music", exclude=("meta.top_country_code", "ttl")), ]), 1, [ event(1, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), ] ), ( pack_events([ event(1, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], dsp="apple_music", exclude=("meta.top_country_code", "ttl")), event(2, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], dsp="spotify", exclude=("meta.top_country_code", "ttl")), event(3, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], dsp="apple_music", exclude=("meta.top_country_code", "ttl")), event(4, j=3, event_type=EventType.DROP, country_code=["au", "mx"], exclude=("meta.top_country_code", "ttl")), event(1, event_type=EventType.DROP, country_code=["us", "fr", "au"], top_country_code=["us"], dsp="apple_music"), # outdated event(2, event_type=EventType.MAJOR_MOVE, extra={"ttl": 5}, dsp="spotify"), event( 2, j=1, k=10, event_type=EventType.MAJOR_MOVE, dsp="spotify", top_country_code=["us", "fr", "global"] ) ]), 4, [ event(1, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), event(2, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), event(3, j=1, k=10, event_type=EventType.MAJOR_MOVE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), event(4, j=3, event_type=EventType.DROP, country_code=["au", "mx"], top_country_code=[], exclude=( "app", "publisher", "meta.type", "data.content.playlist.image_url", "data.content.current_position" ), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), ] ), ( pack_events([ event(1, event_type=EventType.DROP, country_code=["fr", "au"], top_country_code=["au"], dsp="apple_music", exclude=("id",)), event(2, event_type=EventType.MAJOR_MOVE, exclude=("meta.event_type",)), event(3, event_type=EventType.DROP, exclude=("meta.dsp",)), event(5, event_type=EventType.MAJOR_MOVE, exclude=("data.content.track",), dsp="apple_music", ), event(6, event_type=EventType.DROP, exclude=("data.content.playlist.last_updated_at",)), ]), 0, None ), ), ) def test_parse_raw_input(test_logger, input_event, expected_result_len, expected_result): from apollo_delphi_events.utils.input import parse_input result = parse_input(test_logger, input_event, current_dt=dt_now()) assert len(result) == expected_result_len if expected_result is not None: for i, item in enumerate(expected_result): assert result[i] == item @pytest.mark.parametrize( "input_events,expected_result_len,expected_result", ( ( [ event(1, j=3, event_type=EventType.UPDATE, country_code=["au", "mx"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), extra={"ttl": 7200}, event_type_as_value=False, serialize_dt=False), event(2, j=4, k=6, event_type=EventType.ADDITION, country_code=["global"], top_country_code=["us", "fr", "global"], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), event_type_as_value=False, serialize_dt=False), event(3, j=5, k=9, event_type=EventType.ADDITION, country_code=["global"], top_country_code=[], exclude=("app", "publisher", "meta.type", "data.content.playlist.image_url"), event_type_as_value=False, serialize_dt=False) ], 3, [ output_event(1, j=3, event_type=EventType.UPDATE, country_code=["au", "mx"], top_country_code=[]), output_event(2, j=4, k=6, event_type=EventType.ADDITION, country_code=["global"], top_country_code=["us", "fr", "global"]), output_event(3, j=5, k=9, event_type=EventType.ADDITION, country_code=["global"], top_country_code=[]) ] ), ), ) def test_output(test_logger, input_events, expected_result_len, expected_result): from apollo_delphi_events.utils.output import format_output result = format_output(input_events) result = result["data"] assert len(result) == expected_result_len if expected_result is not None: for i, item in enumerate(expected_result): assert result[i] == item