import json import logging import re import time from urllib.parse import unquote import allure from assertpy import assert_that from ui_automation_framework.core import BasePage, TestStatus from ui_automation_framework.utils import CoreConfig from ui_automation_framework.utils import logger as cl from app.src.pages.login_page import LoginPage class AmplitudeHelper(BasePage): log = cl.Logger(logging.DEBUG) CHARTS_DIGEST_PAGE = {"event_type": "charts digest page viewed", "link_destination": "Charts Digest", "gtp_view_button": "gtp charts digest header view cta clicked"} TOP_CHARTS_PAGE = {"event_type": "top charts page viewed", "youtube_event_type": "YT trending charts page viewed", "youtube_filtered_event_type": "YT trending charts page filtered", "youtube_search": "YT trending charts page search", "youtube_export": "YT trending track page export"} HAMBURGER_MENU = {"event_type_icon": "hamburger icon clicked", "event_type_link": "hamburger link clicked", "contact_us_link_destination": "Contact Us"} TOP_HEADER_MENU = {"event_type": "header link clicked"} MOST_VISITED = {"event_type": "most visited track clicked"} COMPARISON_PAGE = { "event_type": "track comparison page viewed", "event_type_track_added": "track comparison track added", "event_type_track_removed": "track comparison track removed", "event_type_page_exited": "track comparison page exited", "link_destination": "Track Comparison", "event_type_page_filtered": "track comparison page filtered", "event_type_tab_changed": "track comparison tab changed", "event_type_table_sorted": "track comparison table sorted", "event_type_back_to_top": "track comparison fab clicked", } MY_STARRED_TRACKS_PAGE = {"event_type": "starred tracks page viewed", "link_destination": "My Starred Tracks", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "My Starred Tracks Page"} GTP_PAGE = { "event_type": "gtp page viewed", "link_destination": "Global Track Priorities", "banner_arrow_event_type": "gtp charts digest header arrow clicked", "interacted_card_event_type": "gtp charts digest header cards interacted", "hot_hits_opened": "gtp hot hits modal opened", "chart_position_opened": "gtp chart position modal opened", "change_date": "gtp date changed", "change_market": "gtp market changed", "change_tab": "gtp tab changed", "hot_hits_change_tab": "gtp hot hits modal interacted", } NEW_MUSIC_FRIDAY_PAGE = {"event_type": "new music friday page viewed", "filtered_event_type": "new music friday page filtered", "link_destination": "New Music Friday", "search_event_type": "new music friday page search", "export_event_type": "new music friday page export"} NEW_MUSIC_FRIDAY_POPUP = {"event_type": "new music friday page popup opened", "filtered_event_type": "new music friday page popup filtered"} MARKET_OVERVIEW_PAGE = {"event_type": "market overview page viewed", "link_destination": "Market Overview"} BUZZ_PAGE = {"event_type": "buzz page viewed", "link_destination": "Buzz"} ACCOUNT_BROWSER_PAGE = {"event_type": "account browser page viewed", "link_destination": "Account Browser"} SPOTIFY_BROWSE_PAGE = {"event_type": "spotify browse page viewed", "link_destination": "Spotify Browse"} PLAYLIST_DUPLICATION_PAGE = {"event_type": "playlist duplication page viewed", "link_destination": "Playlist Duplication"} MOST_ADDED_TRACK_PAGE = {"event_type": "most added track page viewed", "link_destination": "Most Added Track", "track_favorited_event_type": "track favorited", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "Most Added Track Page"} SONY_MUSIC_STREAM_CHART_PAGE = {"event_type": "sony music stream chart page viewed", "link_destination": "Sony Music Stream Chart"} ABOUT_PAGE = {"event_type": "filtr about page viewed", "link_destination": "About"} CAROUSEL_IMAGES_PAGE = {"event_type": "filtr carousel images page viewed", "link_destination": "Carousel Images"} LOCALIZATIONS_PAGE = {"event_type": "filtr localizations page viewed", "link_destination": "Localizations"} PLAYLISTS_CATEGORIES_PAGE = {"event_type": "filtr playlist categories page viewed", "link_destination": "Playlist Categories"} PLAYLISTS_PAGE = {"event_type": "filtr playlists page viewed", "link_destination": "Playlists"} RELEASES_PUSH_TRACKS_PAGE = {"event_type": "filtr releases and push tracks page viewed", "link_destination": "Releases & Push Tracks"} TOP_FILTR_PLAYLISTS_PAGE = {"event_type": "top playlists page viewed", "link_destination": "Top Filtr Playlists", "owner_account": "Filtr"} TOP_SPOTIFY_PLAYLISTS_PAGE = {"event_type": "top playlists page viewed", "link_destination": "Top Spotify Playlists", "owner_account": "Spotify"} TRACK_PAGE = { "event_type": "track page viewed", "dsp_type_spotify": "Spotify", "dsp_type_apple": "Apple Music", "track_favorited_event_type": "track favorited", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "Track Page", "filtered_event_type": "track page filtered", "current_playlists_tab_opened": "Current Playlists & Stations", "previous_playlists_tab_opened": "Past Playlists", "charts_tab_opened": "Charts", "stations_tab_opened": "Apple Stations", "videos_tab_opened": "YouTube", "tiktok_tab_opened": "TikTok", "by_markets_tab_opened": "by Markets", "video_performance_tab_opened": "YouTube", "streams_and_sources_tab_opened": "Streams and Sources", "apple_dsp_selected": "Apple Music", "all_streams_dsp_selected": "All Streams", "by_dsp_dsp_selected": "By DSP", "spotify_dsp_selected": "Spotify", "track_page_interacted": "Track Page Details interacted", "td_export_clicked": "Track Page Details export clicked", "tp_popup_opened": "Track Page Playlist details pop-up opened", "amazon_dsp_selected": "Amazon", "amazon_by_voice_selected": "By Voice", "amazon_by_tier_selected": "By Tier", "amazon_all_streams_selected": "All Streams", "amazon_by_source_selected": "By Source", "amazon_lean_fwd_back": "Lean Fwd / Back", "amazon_ad_supported_tier": ["Ad-Supported"], "amazon_prime_tier": ["Prime"], "amazon_unlimited_tier": ["Unlimited"], "amazon_all_tiers": "All tiers", "amazon_prime_unlimited_tier": ["Prime", "Unlimited"], "amazon_by_source_album": ["Album"], "amazon_by_source_artist": ["Artist"], "amazon_by_source_all_sources": "All sources", "amazon_by_source_with_out_playlist": ["Album", "Artist", "Search", "Songs", "Station", "User Playlist"], "date_range_changed_event_type": "date range changed", "yt_views_and_sources_all_views": "All Views", "yt_views_and_sources_by_video": "By Video", "yt_views_and_sources_by_source": "By Source", "yt_views_and_sources_lean_fwd_back": "Lean Fwd / Back", "yt_other_metrics_by_likes": "By Likes", "yt_other_metrics_by_dislikes": "By Dislikes", "yt_other_metrics_by_comments": "By Comments", "yt_other_metrics_by_shares": "By Shares", "yt_views_and_sources_by_source_playlists": ["Playlists"], "yt_views_and_sources_by_source_direct_or_unkn": ["Direct or Unkn"], "yt_views_and_sources_by_source_video_cards_and_annotations": ["Video Cards & Annotations"], "yt_views_and_sources_by_source_several_sources": ["Video Cards & Annotations", "Suggested Videos", "Shorts"], "yt_views_and_sources_by_source_key_sources": ["Key Sources"], "yt_views_and_sources_by_source_other_sources": ["Other Sources"], "yt_views_and_sources_by_source_all_sources": ["All sources"], "yt_views_and_sources_by_source_key_sources_shorts": ["Key Sources", "Shorts"], "yt_views_and_sources_by_source_other_sources_notifications": ["Other Sources", "Notifications"], "graph_type_line_graph": "Line Graph", "graph_type_area_graph": "Area Graph", "graph_type_area_graph_percentage": "Area Graph Percentage", "tiktok_by_markets_btn_event": "tiktok by markets button clicked", } DSP_COMPARISON_PAGE = { "event_type": "comparison page viewed", "filtered_event_type": "comparison page filtered", "button_clicked": "comparison page button clicked", "graph_filtered": "comparison page graph filtered", "graph_deleted": "comparison page graph deleted", "graph_added": "comparison page add graph", "date_range_changed": "date range changed", } PLAYLIST_PAGE = {"event_type": "playlist page viewed", "dsp_type_spotify": "Spotify", "dsp_type_apple": "Apple Music", "track_favorited_event_type": "track favorited", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "Playlist Page"} ARTIST_PAGE = {"event_type": "artist page viewed", "track_favorited_event_type": "track favorited", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "Artist Page"} ALBUM_PAGE = {"event_type": "album page viewed", "track_favorited_event_type": "track favorited", "track_unfavorited_event_type": "track unfavorited", "track_favorited_source": "Album Page"} ACCOUNT_PAGE = {"event_type": "account page viewed"} VIDEO_PAGE = {"event_type": "video page viewed", "filtered_event_type": "video page filtered", "date_range_changed_event_type": "date range changed", "trending_event_type": "YT trending track page filtered", "trending_export": "YT trending track page export"} TOP_HEADER_MARKET = {"event_type": "global market changed"} SEARCH_RESULTS = {"event_type": "search result clicked", "search_result_type_track": "Track", "search_result_type_playlist": "Playlist", "search_result_type_artist": "Artist", "search_result_type_album": "Album", "search_result_type_account": "Account"} LOGO = {"event_type": "apollo logo clicked"} SIGN_OUT = {"event_type": "Sign Out"} CONTACT_US = {"link_destination": "Contact Us"} TT_DASHBOARD_PAGE = { "event_type": "Sony TikTok Toplists page viewed", "event_page_filtered": "Sony TikTok Toplists page filtered", "event_page_search": "Sony TikTok Toplists page search", "event_page_export": "Sony TikTok Toplists page export", "event_potp_opened": "Sony TikTok Toplists Position popup viewed", "event_top10_opened": "Sony TikTok Toplists Top Markets popup viewed", } def __init__(self, driver): super().__init__(driver) self.driver = driver self.ts = TestStatus(self.driver) self.loginPage = LoginPage(self.driver) @allure.step("Getting Amplitude events from network") def get_amplitude_events(self): network = self.driver.get_log("performance") result_list = [] for entry in network: log = json.loads(entry["message"])["message"] if log["method"] == "Network.requestWillBeSent" and ("amplitude" in log["params"]["request"]["url"]): postData = unquote(log["params"]["request"]["postData"]) self.log.info("PostData: " + str(postData)) found = re.search(r"e=(.*|\s|\S)&upload_time", postData).group(1) result_list.append(json.loads(found)) final_results_list = [] self.log.info("Result list: " + str(result_list)) for t in result_list: final_results_list.extend(t) self.log.info("Final result list: " + str(final_results_list)) self.log.info(f"""amplitude events: {[r['event_type'] for r in final_results_list]}""") self.log.info("All amplitude events: " + str(final_results_list)) return final_results_list @allure.step("Getting Amplitude events from network and wait") def wait_and_get_amplitude_events(self, count=1, strict=True): events_all = [] timeout = int(CoreConfig.SELENIUM_TIMEOUT) for t in list(range(1, timeout)): time.sleep(1) events_raw = self.get_amplitude_events() events_all = events_all + events_raw self.log.info(f"event: {events_all}, len: {len(events_all)}, t: {t}") if strict: if len(events_all) == count: break else: if len(events_all) >= count: break assert_that(t).is_less_than(timeout) if strict: assert_that(len(events_all)).is_equal_to(count) else: assert_that(len(events_all)).is_greater_than_or_equal_to(count) self.log.info(f"events all: {events_all}, len: {len(events_all)}, t: {t}") return events_all @allure.step("Checking first event as Amplitude user initialization") def is_initializer_event(self, event): user_email = self.loginPage.get_user_email_with_worker() self.ts.markFinal(event["event_type"] == "$identify", "First event is initializer") self.ts.markFinal(event["user_properties"]["$set"]["email"] == user_email, "Setted up correct user email") @allure.step("Checking custom Amplitude event") def verify_for_event_type(self, event, event_type): assert_that(event["event_type"]).is_equal_to(event_type) @allure.step("Checking event user properties: number of stared tracks, name of the user and email") def verify_event_user_properties(self, event): user_email = self.loginPage.get_user_email_with_worker() event_email = event["user_properties"]["$set"]["email"] event_user = event["user_properties"]["$set"]["name"] assert_that(event_email).is_equal_to(event_user) assert_that(event_user).is_equal_to(user_email)