from dataclasses import dataclass, asdict, field from datetime import datetime from typing import List, Dict, Union, Tuple, Any from apollo_utils.core.utils.ext_enum import ExtEnum from faker import Faker faker = Faker() TEST_TTL = 10 class MetaType(ExtEnum): ENTRY = "starred_track_top_chart_entry" EXIT = "starred_track_top_chart_exit" MOVE = "starred_track_top_chart_move" @dataclass class IncomingEventNestedMeta: """Nested meta information for incoming event""" domain: str dsp: str country_code: str type: str subject: str = "starred_track_in_chart" @dataclass class IncomingEventNestedPublisher: """Nested publisher information for incoming event""" id: str = "user_data_api_ecs_web_test" owner: str = "test" type: str = "test" instance: str = "test" env: str = "testing" @dataclass class IncomingEvent: """Helper for incoming event creation id: should be an integer represented as a string created_at: isoformat datetime """ meta: IncomingEventNestedMeta publisher: IncomingEventNestedPublisher id: str = "1234567890" app: str = "apollo_test" code: str = "messages_created" ttl: int = TEST_TTL created_at: str = datetime.utcnow().isoformat() @dataclass class IcnomingEventWithData: meta: IncomingEventNestedMeta id: str = "1234567890" app: str = "apollo_test" code: str = "messages_created" ttl: int = TEST_TTL created_at: str = datetime.utcnow().isoformat() updated_at: str = datetime.utcnow().isoformat() event_id: str = "123456" account_id: str = "123456" message_id: str = "123456" def create_incoming_event( event_id: str = "123456789", event_ttl: int = TEST_TTL, event_created_at: str = datetime.utcnow().isoformat(), meta_dsp: str = "spotify", meta_country_code: str = "us", meta_domain: str = "message", meta_type: str = "starred_track_top_chart_entry", ): event = IncomingEvent( meta=IncomingEventNestedMeta( domain=meta_domain, dsp=meta_dsp, country_code=meta_country_code, type=meta_type, ), id=event_id, publisher=IncomingEventNestedPublisher(), ttl=event_ttl, created_at=event_created_at ) return asdict(event) def create_incoming_message( db_id: str = "1", account_id: str = "1", event_id: str = "1", message_id: str = "123456789", message_ttl: int = TEST_TTL, message_created_at: str = datetime.utcnow().isoformat(), message_updated_at: str = datetime.utcnow().isoformat(), meta_dsp: str = "spotify", meta_country_code: str = "us", meta_domain: str = "message", meta_type: str = "starred_track_top_chart_entry", ): event = IcnomingEventWithData( meta=IncomingEventNestedMeta( domain=meta_domain, dsp=meta_dsp, country_code=meta_country_code, type=meta_type, ), id=db_id, event_id=event_id, account_id=account_id, message_id=message_id, ttl=message_ttl, created_at=message_created_at, updated_at=message_updated_at ) return asdict(event) def get_string_fake_field(field_name: str = "fake_field", i: int = 0) -> str: """Get fake field Args: field_name: name of a field to return i: field index to add Returns: str """ return f"fake_{field_name}_{i}" def get_artists(i: int) -> List[Dict[str, str]]: """Get fake tracks artists Args: i: field index to add Returns: List of artists dict """ return [{ "id": str(faker.random_int(100, 999)), "name": faker.name() }] @dataclass class MessageTrack: """Nested track information for any event""" id: str isrc: str name: str image_url: str artists: list = field(default_factory=list) @dataclass class Content: """Dataclass to create events such as additions/removals/moves""" track: MessageTrack current_position: Union[int, None] = None previous_position: Union[int, None] = None body: str = "I am a body message describing the event of the track in a chart" title: str = "" def get_fake_track_positions( changelog_type: MetaType, min_random_position: int = 0, max_random_position: int = 20 ) -> Tuple[Union[int, None], Union[int, None]]: """Get fake tracks positions based on its changelog type such as entry/exit/move Args: changelog_type: ENTRY/EXIT/MOVE of a <> min_random_position: min of random integer for a position max_random_position: max ^^ --""--""-- ^^ Returns: (current_position, previous_position) """ changelog_specific_positions = { MetaType.ENTRY: ( faker.random_int(min_random_position, max_random_position), None ), MetaType.EXIT: ( None, faker.random_int(min_random_position, max_random_position) ), MetaType.MOVE: ( faker.random_int(min_random_position, max_random_position), faker.random_int(min_random_position, max_random_position) ) } return changelog_specific_positions[changelog_type] def get_content_with_track(i: int = 0, changelog_type: MetaType = MetaType.ENTRY) -> Dict[str, Any]: """Get single fake track""" current_position, previous_position = get_fake_track_positions(changelog_type) track = Content( track=MessageTrack( id=get_string_fake_field("id", i), isrc=get_string_fake_field("isrc", i), name=get_string_fake_field("name", i), image_url=get_string_fake_field("image_url", i), artists=get_artists(i) ), current_position=current_position, previous_position=previous_position, title="" ) return asdict(track) @dataclass class Device: expo_token: str id: str def get_nested_devices(device_count: int = 1): result = [] for i in range(device_count): result.append( asdict( Device( expo_token=get_string_fake_field("expo_token", i), id=str(i) ) ) ) return result ChartName = { "spotify": "Spotify Daily Top 200 Chart", "apple": "Apple Music Top 100 Chart" } @dataclass class DataContentNestedChart: name: str def get_content_nested_chart(dsp: str = "spotify"): return asdict( DataContentNestedChart( name=ChartName[dsp] ) ) def create_feed_message_with_data( message_db_id: str = "0", account_id: str = "0", message_id: str = "0", event_id: str = "123456789", message_ttl: int = TEST_TTL, message_created_at: str = datetime.utcnow().isoformat(), message_updated_at: str = datetime.utcnow().isoformat(), meta_dsp: str = "spotify", meta_country_code: str = "us", meta_domain: str = "message", meta_type: str = MetaType.ENTRY.value, number_of_devices: int = 1 ): message_without_data = create_incoming_message( db_id=message_db_id, account_id=account_id, event_id=event_id, message_id=message_id, message_ttl=message_ttl, message_created_at=message_created_at, message_updated_at=message_updated_at, meta_dsp=meta_dsp, meta_country_code=meta_country_code, meta_domain=meta_domain, meta_type=meta_type, ) devices = get_nested_devices(number_of_devices) content_nested_track = get_content_with_track(changelog_type=MetaType(meta_type)) content_nested_chart = {"chart": get_content_nested_chart(dsp=meta_dsp)} data = { "data": { "recipient": { "devices": devices, }, "content": { **content_nested_track, **content_nested_chart } } } return {**message_without_data, **data} def get_broken_data(): return { "id": "fake id", "event_id": "fake event id", "message_id": "fake_message_id", "fail": "reason to break data =)" }