import random import re from time import sleep, time from datetime import datetime import pytest from appium.webdriver.common.appiumby import AppiumBy as By from selenium.common.exceptions import TimeoutException, \ StaleElementReferenceException, NoSuchElementException from helpers.appium.gestures import tap, press_hold from helpers.platform.android.locators import ( content_desc_xpath, resource_id_xpath, text_view_xpath) from ..base_screen import BaseScreen from page_objects.mixins.header_info_mixin import HeaderInfoMixin from page_objects.mixins.playlists_mixin import PlaylistsMixin from page_objects.mixins.utils import ( retry_on_assertion, retry_on_stale_element, parse_custom_format) class SoundRecordingScreen(BaseScreen, HeaderInfoMixin, PlaylistsMixin): access = By.ACCESSIBILITY_ID # --- Header --- header = resource_id_xpath('swipeViewPage') cover_art = '//android.view.ViewGroup[@resource-id="coverArt"]' favorite_btn = 'favoriteButton' bullet_0 = resource_id_xpath('bullet0') bullet_1 = resource_id_xpath('bullet1') bullet_2 = resource_id_xpath('bullet2') # --- General Info --- all_time_streams = 'heroMetrics' all_time_streams_amount = \ ("//android.view.ViewGroup[@resource-id='allTimeStreams']/" "android.widget.TextView[1]") seven_days_tiktok_creations = text_view_xpath('7 DAYS TIKTOK CREATIONS') seven_days_tiktok_creations_amount = \ ("//android.view.ViewGroup[@resource-id='tiktokCreations']/" "android.widget.TextView[1]") playlists_total = text_view_xpath('PLAYLISTS TOTAL') playlists_total_amount = ( "//android.view.ViewGroup[@resource-id='playlistsTotal']/" "android.widget.TextView[1]") # --- Streaming Trend --- st_section = 'streamingTrend' st_title = text_view_xpath('Streaming Trend') st_store_selection_btn = resource_id_xpath('storeSelectionButtonTitle') st_by_stores_btn = resource_id_xpath('store') st_by_countries_btn = resource_id_xpath('country') st_by_sources_btn = resource_id_xpath('sos') st_graph = 'streamingTrend' st_recent_and_all_time_switcher = resource_id_xpath('switchSelector') st_recent_btn = ( st_recent_and_all_time_switcher + content_desc_xpath('RECENT')) st_all_time_btn = ( st_recent_and_all_time_switcher + content_desc_xpath('ALL TIME')) st_table = resource_id_xpath('table') st_store_row = resource_id_xpath('touchableTableRow') st_show_all_items_btn = 'SHOW ALL ITEMS' st_show_less_items_btn = 'SHOW LESS ITEMS' st_store_status_btn = resource_id_xpath('sourcesLink') st_store_status_title = text_view_xpath('Store Status') st_store_status_icon = (content_desc_xpath('Bottom Sheet') + resource_id_xpath('storeIcon')) st_store_status_date = resource_id_xpath('date') st_store_status_message = resource_id_xpath('message') # --- TikTok Creations --- tiktok_section = 'accessibility.soundRecordingScreenBody.tiktokChart' see_all_tiktok_data_btn = 'SEE ALL TIKTOK DATA' tiktok_pot_by_total_btn = 'Total' tiktok_pot_by_store_btn = 'UGC/PGC' tiktok_pot_by_country_btn = 'Country' # --- Top Playlists --- playlist_section = 'soundRecordingPlacementList' top_playlists_title = resource_id_xpath('topPlaylistTitle') top_playlists_i_icon = (top_playlists_title + '//*[@resource-id="informationIcon"]') bottom_sheet = content_desc_xpath('Bottom Sheet') top_playlists_content = [ "Top Playlists features current playlist placements " "and relevant metrics, including:", "• playlist type, curator market, and followers", "• playlist add date, song streams generated by the playlist, " "and playlist position", "The default view for Top Playlists filters to highly impactful " "stores and playlist types. All current playlist placements can be " "viewed by adjusting filters within \'see all\'."] one_day_streams_btn = '1d Streams' seven_days_streams_btn = '7d Streams' most_recent_btn = 'Most Recent' followers_btn = 'Followers' see_all_playlists_btn = ( '//android.view.ViewGroup[@resource-id="cardPlaylist"]/' 'following-sibling::android.view.ViewGroup[@content-desc="SEE ALL"]') # --- Top Markets --- top_markets_section = 'topMarkets' top_markets_title = text_view_xpath('Top Markets') top_markets_time_period = text_view_xpath('PAST 7 DAYS') top_markets_sources = 'Sources' top_markets_card_country_name = resource_id_xpath('countryName') top_markets_card_stream_count = resource_id_xpath('streamCount') top_markets_sort_btn = resource_id_xpath('sortButton') # --- Top Charts --- top_charts_section = resource_id_xpath('topCharts') top_charts_i_icon = resource_id_xpath('informationIcon') top_charts_i_title = ('//android.widget.TextView[@resource-id="headerText"' ' and @text="What are top charts?"]') top_charts_i_content = text_view_xpath( 'Top charts shows current track placements for Spotify and Apple ' 'Music charts across markets.') top_charts_i_close_btn = resource_id_xpath('closeButton') top_charts_bullets = \ ('//android.view.ViewGroup[starts-with(@resource-id, "bullet") and ' 'string-length(@resource-id) > 6]') spotify_top_charts_card = \ ('//android.view.ViewGroup[@resource-id="topCharts"]/' 'android.widget.HorizontalScrollView[1]') apple_top_charts_card = \ ('//android.view.ViewGroup[@resource-id="topCharts"]/' 'android.widget.HorizontalScrollView[2]') top_charts_card = \ ('//android.view.ViewGroup[@resource-id="topCharts"]/' 'android.widget.HorizontalScrollView') spotify_charts_i_icon = resource_id_xpath('informationIcon') apple_charts_i_icon = resource_id_xpath('informationIcon') # --- Top Charts Page --- spotify_charts_title = text_view_xpath('Spotify Charts') apple_charts_title = text_view_xpath('Apple Music Charts') top_charts_page_i_icon = resource_id_xpath('informationIcon') top_charts_page_sort_btn = resource_id_xpath('sortButton') # --- Other --- hit_a_snag_msg = text_view_xpath('Sorry, we hit a snag') data_failed_to_load_msg = \ '//android.widget.TextView[contains(@text, "Data failed to load")]' error_container = resource_id_xpath('placementsError.errorContainer') refresh_on_data_failed_btn = ( resource_id_xpath('placementsError.retryButton')) not_in_catalog_msg = resource_id_xpath('notInCatalog') def verify_header_elements_for_artist(self, artist_name): self.waiter(20, By.XPATH, self.cover_art) self.waiter(15, self.access, self.favorite_btn) self.waiter(3, By.XPATH, text_view_xpath(artist_name)) self.waiter(3, By.XPATH, self.bullet_0) self.waiter(3, By.XPATH, self.bullet_1) self.waiter(3, By.XPATH, self.bullet_2) def verify_header_info(self, expected_info): self.scroll_header("left") header_info = self.header_info() for key, expected_value in expected_info.items(): actual_value = header_info.get(key) assert actual_value == expected_value, ( f"Expected '{key}' to be '{expected_value}', " f"but got '{actual_value}'. Full header info: {header_info}" ) self.scroll_header("right") @retry_on_stale_element() def verify_general_info_section(self): self.waiter(30, self.access, self.all_time_streams) self.waiter(3, By.XPATH, self.seven_days_tiktok_creations) self.waiter(3, By.XPATH, self.playlists_total) self.get_all_time_streams_amount() self.get_seven_days_tiktok_creation_amount() self.get_playlists_total_amount() def tap_artist(self, artist_name): self.waiter(20, By.XPATH, content_desc_xpath(artist_name)).click() def verify_streaming_trend_main_elements(self): elements = [ self.st_title, self.st_by_stores_btn, self.st_by_countries_btn, self.st_by_sources_btn, self.st_recent_and_all_time_switcher, self.st_table, ] for locator in elements: self.scroll_and_assert(By.XPATH, locator, max_scrolls=5, small_scroll=True) def verify_recent_period_selected_by_default(self): self.waiter(10, By.XPATH, resource_id_xpath('1day')) self.wait_for_element_invisibility( By.XPATH, resource_id_xpath('allTime')) def verify_pot_rows_count(self, amount): self.scroll_and_assert( By.XPATH, self.st_by_stores_btn, direction='up', small_scroll=True, max_scrolls=10) expected = amount + 1 # +1 for "Total Streams" row max_scrolls = 8 for attempt in range(max_scrolls): rows = self.driver.find_elements(By.XPATH, self.st_store_row) current_count = len(rows) if current_count == expected: return self.scroll_page(direction="down", small_scroll=True) sleep(1) # short settle; keep small to reduce flakiness rows = self.driver.find_elements(By.XPATH, self.st_store_row) final_count = len(rows) if final_count == expected: return raise AssertionError( f"POT stores count not reached after scrolling:\n" f" Expected: {expected}\n" f" Actual: {len(rows)}\n" f" Max scrolls: {max_scrolls}" ) @retry_on_assertion() def verify_pot_graph_has_expected_number_of_lines(self, num_lines=1): self.scroll_and_assert(By.XPATH, self.st_by_stores_btn, direction='up') lines_loc = \ ('//android.view.ViewGroup[@resource-id="chartContainer"]/' 'com.horcrux.svg.SvgView/com.horcrux.svg.GroupView/' 'com.horcrux.svg.GroupView/com.horcrux.svg.PathView') graph_lines = self.waiter(10, By.XPATH, lines_loc, multiple=True) assert len(graph_lines) == num_lines, ( f"Expected {num_lines} graph lines, but found {len(graph_lines)}" ) @retry_on_assertion() @retry_on_stale_element() def verify_pot_graph_last_day_matches_table_1d_value( self, context, value_index=1): legend_values_loc = \ ('//android.view.ViewGroup[@content-desc="streamingTrend"]//' '*[@resource-id="labelContainer"]//android.widget.EditText') try: legend_values = self.waiter(10, By.XPATH, legend_values_loc, multiple=True) except TimeoutException: if context.table_value_for_1day in ['-', '0']: self.verify_streaming_data_delayed_message() return else: raise if context.active_screen == 'Song Screen': graph_amount = legend_values[value_index].text.strip() elif context.active_screen == 'Product Screen': graph_amount = legend_values[value_index + 1].text.strip() # Fallback for unstable Product Screen graph legend ordering if 'TOTAL' not in graph_amount: graph_amount = legend_values[value_index].text.strip() else: adjusted_index = max(0, value_index - 1) graph_amount = legend_values[adjusted_index].text.strip() if getattr(context, 'active_pot_tab', None) != 'By Stores': graph_amount = graph_amount.removeprefix('TOTAL: ') assert context.table_value_for_1day == graph_amount, ( f'POT graph last day value mismatch: ' f'expected "{context.table_value_for_1day}", ' f'got "{graph_amount}"') @retry_on_stale_element() def verify_graph_last_day_matches_table_1d_value(self, context, tab_name): table_value_for_1day_loc = self.st_table + self.st_store_row self.scroll_and_assert(By.XPATH, table_value_for_1day_loc, small_scroll=True) row = self.waiter(10, By.XPATH, table_value_for_1day_loc) row_data = row.get_attribute("content-desc") if tab_name != 'By Stores': value = row_data.split(',')[2].strip() # TODO: Conditional should be removed once bug GO-4463 is fixed if value == '-' and tab_name == 'By Countries': value = '0' context.table_value_for_1day = value else: if len(row_data.split(',')) <= 3: # Sometimes below values of the row is outside of the screen # That's why we need to make a small scroll to see the full row self.scroll_page(direction="down", small_scroll=True) row = self.waiter(10, By.XPATH, table_value_for_1day_loc) row_data = row.get_attribute("content-desc") value = row_data.split(',')[1].strip() context.table_value_for_1day = value self.verify_pot_graph_last_day_matches_table_1d_value(context) def verify_streaming_data_delayed_message(self): self.scroll_and_assert(By.XPATH, self.st_by_stores_btn, direction='up') message = self.waiter(3, By.XPATH, resource_id_xpath('emptyState') + '/android.widget.TextView') assert 'We receive streaming data on a ~2 day delay' in message.text def verify_pot_recent_total_value_consistent_across_tabs(self, context): views = ["By Stores", "By Countries", "By Sources"] totals_by_view = {} for view in views: self.switch_pot_view(context, view) total_streams_row = self.waiter( 3, By.XPATH, self.st_store_row, multiple=True)[0] content_desc = total_streams_row.get_attribute("content-desc") values = [value.strip() for value in content_desc.split(",")[-4:]] totals_by_view[view] = values assert ( totals_by_view["By Stores"] == totals_by_view["By Countries"] == totals_by_view["By Sources"] ), ( "POT recent total values differ across tabs:\n" f' By Stores: {totals_by_view["By Stores"]}\n' f' By Countries: {totals_by_view["By Countries"]}\n' f' By Sources: {totals_by_view["By Sources"]}' ) def verify_pot_all_time_total_value_consistent_across_tabs(self, context): views = ["By Stores", "By Countries", "By Sources"] totals_by_view = {} for view in views: self.switch_pot_view(context, view) total_streams_row = self.waiter( 3, By.XPATH, self.st_store_row, multiple=True)[0] content_desc = total_streams_row.get_attribute("content-desc") values = content_desc.split(",")[1].strip() totals_by_view[view] = values assert ( totals_by_view["By Stores"] == totals_by_view["By Countries"] == totals_by_view["By Sources"] ), ( "POT all time total values differ across tabs:\n" f' By Stores: {totals_by_view["By Stores"]}\n' f' By Countries: {totals_by_view["By Countries"]}\n' f' By Sources: {totals_by_view["By Sources"]}' ) def verify_show_more_button(self): self.scroll_and_assert( By.ACCESSIBILITY_ID, self.st_show_all_items_btn, small_scroll=True) def verify_show_less_button(self): self.scroll_and_assert( By.ACCESSIBILITY_ID, self.st_show_less_items_btn, small_scroll=True) @retry_on_assertion() def verify_store_status_screen(self): self.waiter(10, By.XPATH, self.st_store_status_title) store_icons = self.waiter( 3, By.XPATH, self.st_store_status_icon, multiple=True ) dates = self.waiter( 3, By.XPATH, self.st_store_status_date, multiple=True ) messages = self.waiter( 3, By.XPATH, self.st_store_status_message, multiple=True ) assert len(store_icons) >= 10, ( f"Expected at least 10 store icons, got {len(store_icons)}" ) assert len(store_icons) == len(messages), ( f"Store icons and messages count mismatch: " f"icons={len(store_icons)}, messages={len(messages)}" ) # Dates may not be fully rendered in the scrollable list, # so allow a +1 tolerance compared to icons/messages. assert len(store_icons) in (len(dates), len(dates) + 1), ( f"Store icons and dates count mismatch: " f"icons={len(store_icons)}, dates={len(dates)}" ) self.swipe_down_to_dismiss() self.wait_for_element_invisibility( By.XPATH, self.st_store_status_title) def verify_tiktok_creations_main_elements(self): elements = { "TikTok Creations title": text_view_xpath('TikTok Creations'), "PAST 28 DAYS label": resource_id_xpath('pastDaysPeriod'), "TikTok chart": resource_id_xpath('chartContainer'), "Data table": resource_id_xpath(self.tiktok_section) + resource_id_xpath('table') } for name, locator in elements.items(): self.scroll_and_assert(By.XPATH, locator, max_scrolls=5, small_scroll=True) def _collect_chart_values(self, chart_loc, popup_value_loc, days): """ Collects numeric values from the chart by performing a smooth, coordinate-based drag across the chart. """ sleep(2) # short delay in case scrolling still in progress chart = self.waiter(10, By.XPATH, chart_loc) location, size = chart.location, chart.size # Define dynamic geometry start_x = location['x'] + size['width'] - 10 # near right edge y = location['y'] + size['height'] * 0.25 # safe Y midline steps = 28 delta = size['width'] / (steps - 1) # dynamic delta total = 0 seen_values = set() def read_value(timeout=2, poll=0.1): end = time() + timeout last_err = None while time() < end: try: el = self.driver.find_element(By.XPATH, popup_value_loc) text = (el.text or "").strip() if text: return text except (NoSuchElementException, StaleElementReferenceException) as e: last_err = e sleep(poll) raise last_err or NoSuchElementException( f"Popup value not found: {popup_value_loc}") tap(self.driver, x=start_x, y=y) # Collect values by pressing each point for i in range(days): value_text = read_value() seen_values.add(value_text) total += parse_custom_format(value_text) print(f"Step {i + 1}: {value_text}") # Only advance to the next point if there is a next point if i < days - 1: next_x = start_x - (i + 1) * delta press_hold(self.driver, next_x, y, hold_s=0.2) # Dismiss tooltip tap(self.driver, 5, 5) print(f"Total chart sum: {total}") return total def _get_table_total(self, table_total_loc): self.scroll_and_assert(By.XPATH, table_total_loc, max_scrolls=5, small_scroll=True) total = self.waiter(10, By.XPATH, table_total_loc) try: return int(total.text.strip()) except ValueError: return parse_custom_format(total.text.strip()) def verify_chart_total_matches_table( self, chart_loc, popup_value_loc, table_total_loc, label, days): """ Collects all values displayed in the chart popup, sums them up, and compares the total to the 28 DAYS total in the table. """ chart_total = self._collect_chart_values( chart_loc, popup_value_loc, days) table_total = self._get_table_total(table_total_loc) view_labels = ( 'TikTok Views [Detailed Screen]', 'UGC TikTok Views [Detailed Screen]', 'PGC TikTok Views [Detailed Screen]', ) if label not in view_labels: assert chart_total == table_total, ( f"[{label}] Mismatch detected:\n" f" - Chart total: {chart_total}\n" f" - Table total: {table_total}" ) print(f"[{label}] Chart total matches table: {chart_total}") else: assert round(chart_total, -2) == round(table_total, -2), ( f"[{label}] Mismatch detected:\n" f" - Chart total: {chart_total}\n" f" - Table total: {table_total}" ) print(f"[{label}] Chart total matches table: {chart_total}") def verify_see_all_tiktok_data_button(self): self.scroll_and_assert(self.access, self.see_all_tiktok_data_btn, max_scrolls=10, small_scroll=True, wd_timeout=4) self.scroll_page(small_scroll=True) def tap_see_all_tiktok_data_button(self): self.click_with_retry(10, self.access, self.see_all_tiktok_data_btn) self.wait_for_element_invisibility( self.access, self.see_all_tiktok_data_btn) def verify_tiktok_data_page(self): self.waiter(10, self.access, self.tiktok_pot_by_total_btn) self.waiter(3, self.access, self.tiktok_pot_by_store_btn) self.waiter(3, self.access, self.tiktok_pot_by_country_btn) def _verify_tiktok_chart_matches_table( self, label: str, days: int, chart_index: int = 1, table_index: int = 1, row_name: str | None = None, value_index: int = 2): """ Generic verifier for TikTok chart-to-table consistency. Args: label: Descriptive label for logging and assertion. chart_index: Which chart to use (1 = Creations, 2 = Views). table_index: Which table to use (1 = Creations table, 2 = Views). row_name: Optional row resource ID ('ugcRowName', 'pgcRowName'). value_index: Which TextView to compare (default = 2). """ chart_elements = self.waiter( 10, By.XPATH, resource_id_xpath('chartContainer'), multiple=True ) # --- Chart locator --- if len(chart_elements) >= chart_index: chart_loc = \ f"({resource_id_xpath('chartContainer')})[{chart_index}]" if chart_index == 1: popup_value_loc = ( f"({resource_id_xpath('labelContainer')})[{chart_index}]" f"/android.view.ViewGroup/android.widget.EditText[2]" ) else: popup_value_loc = ( f"({resource_id_xpath('labelContainer')})[{chart_index}]" f"/android.view.ViewGroup/android.widget.EditText" ) else: chart_loc = resource_id_xpath('chartContainer') popup_value_loc = ( f"{resource_id_xpath('labelContainer')}" f"/android.view.ViewGroup/android.widget.EditText" ) # --- Table locator --- if table_index == 1: base_table = f"({resource_id_xpath('touchableTableRow')})" elif table_index == 2: base_table = f"{resource_id_xpath('touchableTableRow')}" elif table_index == 3: base_table = f"({resource_id_xpath('touchableTableRow')})[2]" if row_name: table_total_loc = ( f"({base_table}{resource_id_xpath(row_name)})[{table_index}]" f"/following-sibling::android.widget.TextView[{value_index}]" ) else: table_total_loc = ( f"{base_table}//android.widget.TextView[{value_index}]" ) # --- Run verification --- self.verify_chart_total_matches_table(chart_loc, popup_value_loc, table_total_loc, label, days) def verify_tiktok_creations_chart_matches_table( self, value_index, days): self._verify_tiktok_chart_matches_table( label="TikTok Creations [Song Screen]", chart_index=1, value_index=value_index, days=days ) def verify_detailed_tiktok_creations_chart_matches_table( self, value_index, days): self._verify_tiktok_chart_matches_table( label="TikTok Creations [Detailed Screen]", chart_index=2, value_index=value_index, days=days ) def verify_detailed_tiktok_views_chart_matches_table( self, value_index, days): self.scroll_page() self._verify_tiktok_chart_matches_table( label="TikTok Views [Detailed Screen]", chart_index=2, table_index=3, value_index=value_index, days=days ) def verify_detailed_ugc_tiktok_creations_chart_matches_table( self, value_index, days): self._verify_tiktok_chart_matches_table( label="UGC TikTok Creations [Detailed Screen]", row_name='ugcRowName', chart_index=2, value_index=value_index, days=days ) def verify_detailed_ugc_tiktok_views_chart_matches_table( self, value_index, days): try: self._verify_tiktok_chart_matches_table( label="UGC TikTok Views [Detailed Screen]", chart_index=2, table_index=2, row_name='ugcRowName', value_index=value_index, days=days ) except TimeoutException: self._verify_tiktok_chart_matches_table( label="UGC TikTok Views [Detailed Screen]", chart_index=2, table_index=1, row_name='ugcRowName', value_index=value_index, days=days ) def verify_detailed_pgc_tiktok_creations_chart_matches_table( self, value_index, days): self._verify_tiktok_chart_matches_table( label="PGC TikTok Creations [Detailed Screen]", chart_index=2, row_name='pgcRowName', value_index=value_index, days=days ) def verify_detailed_pgc_tiktok_views_chart_matches_table( self, value_index, days): try: self._verify_tiktok_chart_matches_table( label="PGC TikTok Views [Detailed Screen]", chart_index=2, table_index=2, row_name='pgcRowName', value_index=value_index, days=days ) except TimeoutException: self._verify_tiktok_chart_matches_table( label="PGC TikTok Views [Detailed Screen]", chart_index=2, table_index=1, row_name='pgcRowName', value_index=value_index, days=days ) def tap_ugc_pgc_tab(self): self.scroll_page(direction='up') self.click_with_retry(10, self.access, self.tiktok_pot_by_store_btn) def toggle_chart_to_ugc_only(self): self.scroll_and_assert(By.XPATH, resource_id_xpath('ugcRowName'), max_scrolls=3, small_scroll=True) ugc_rows = self.waiter( 10, By.XPATH, resource_id_xpath('ugcRowName'), multiple=True) if len(ugc_rows) == 1: self.click_with_retry( 10, By.XPATH, resource_id_xpath('ugcRowName')) else: self.click_with_retry(element=ugc_rows[-1]) def toggle_chart_to_pgc_only(self): self.scroll_and_assert(By.XPATH, resource_id_xpath('pgcRowName'), max_scrolls=3, small_scroll=True) pgc_rows = self.waiter(10, By.XPATH, resource_id_xpath('pgcRowName'), multiple=True) if len(pgc_rows) == 1: self.click_with_retry(10, By.XPATH, resource_id_xpath('pgcRowName')) else: self.click_with_retry(element=pgc_rows[-1]) def scroll_to_ugc_pgc_views_section(self): self.scroll_page(times=2) def tap_country_tab(self): self.scroll_page(direction='up') self.click_with_retry(10, self.access, self.tiktok_pot_by_country_btn) def verify_ugc_pgc_tab(self): self.waiter(10, By.ACCESSIBILITY_ID, 'UGC/PGC') def verify_country_tab(self): self.waiter(10, By.ACCESSIBILITY_ID, 'Country') def tap_top_playlists_i_icon(self): try: self.waiter(5, By.XPATH, self.error_container) self.click_with_retry(5, By.XPATH, self.refresh_on_data_failed_btn) except TimeoutException: pass self.scroll_and_assert(By.XPATH, self.top_playlists_i_icon, 10, 'down', True, 5) self.scroll_page(small_scroll=True) top_playlists_i_icon_elem = ( self.driver.find_element(By.XPATH, self.top_playlists_i_icon)) self.tap_after_overlap_check(top_playlists_i_icon_elem) def tap_top_charts_i_icon(self): self.scroll_and_assert(By.XPATH, self.top_charts_i_icon, 10, 'down') self.click_with_retry(10, By.XPATH, self.top_charts_i_icon) def tap_see_all_playlists_btn(self): self.waiter(10, By.XPATH, self.see_all_playlists_btn) sleep(1) see_all_playlists_btn_elem = ( self.driver.find_element(By.XPATH, self.see_all_playlists_btn)) self.tap_after_overlap_check(see_all_playlists_btn_elem) self.wait_for_element_invisibility(By.XPATH, self.see_all_playlists_btn) def assert_track_and_artist_names(self, track_name, artist_name): track_xpath = text_view_xpath(f'{track_name}0') artist_list = artist_name.split(', ') # artist_assertion_limit = 2 if len(artist_list) <= 2 else 1 def assert_elements(): self.waiter(10, By.XPATH, self.cover_art) self.assert_element(By.XPATH, track_xpath, 20) # TODO: Return check of all artists when GO-4053 will be fixed # for artist in artist_list[:artist_assertion_limit]: # Check if at least one artist is visible artist_found = False for artist in artist_list: artist_loc = \ (f'//android.widget.TextView[' f'starts-with(@text, "{artist}")]') try: self.assert_element(By.XPATH, artist_loc) artist_found = True break except Exception: continue assert artist_found, \ f"None of the expected artists were found: {artist_list}" try: assert_elements() except StaleElementReferenceException: sleep(3) assert_elements() def go_to_section(self, context, section_name): sections = { 'Streaming Trend': (self.access, self.st_section), 'TikTok Creations': (self.access, self.tiktok_section), 'Top Playlists': (self.access, self.playlist_section), 'Top Markets': (self.access, self.top_markets_section), 'Top Charts': (By.XPATH, self.top_charts_section) } strategy, loc = sections[section_name] self.scroll_and_assert(strategy, loc, 5) def inspect_pot_section(self): self.scroll_and_assert(By.XPATH, self.st_by_stores_btn) def inspect_pot_period_switcher(self): self.scroll_and_assert(By.XPATH, self.st_recent_and_all_time_switcher) def select_table_period(self, period_type): period_map = { 'Recent': self.st_recent_btn, 'All Time': self.st_all_time_btn, } if period_type not in period_map: raise ValueError('period_type must be "Recent" or "All Time"') period_elem = ( self.driver.find_element(By.XPATH, period_map[period_type])) self.tap_after_overlap_check(period_elem) def filter_dsos_by_store(self, store_name): sheet_title_loc = text_view_xpath('Store') random_store_loc = '(//*[@resource-id="storeIcon"]/..)[position() > 5]' self.scroll_and_assert( By.XPATH, self.st_store_selection_btn, direction='up', small_scroll=True, max_scrolls=5) self.click_with_retry(3, By.XPATH, self.st_store_selection_btn) self.waiter(5, By.XPATH, sheet_title_loc) if store_name == 'Random store': rows = self.waiter(5, By.XPATH, random_store_loc, multiple=True) random_store = random.choice(rows) self.click_with_retry(3, element=random_store) else: self.click_with_retry(5, By.XPATH, text_view_xpath(store_name)) def switch_pot_view(self, context, tab_name): self.scroll_and_assert(By.XPATH, self.st_title, direction='up') self.click_with_retry(10, By.ACCESSIBILITY_ID, tab_name) context.active_pot_tab = tab_name sleep(1) def tap_show_all_items_button(self): self.scroll_and_assert( By.ACCESSIBILITY_ID, self.st_show_all_items_btn, max_scrolls=1, small_scroll=True) button_elem = self.driver.find_element( By.ACCESSIBILITY_ID, self.st_show_all_items_btn) self.tap_after_overlap_check(button_elem) def tap_show_less_items_button(self): self.scroll_and_assert( By.ACCESSIBILITY_ID, self.st_show_less_items_btn, max_scrolls=1, small_scroll=True) button_elem = self.driver.find_element(By.ACCESSIBILITY_ID, self.st_show_less_items_btn) self.tap_after_overlap_check(button_elem) def select_random_pot_store(self, context): stores_with_values_loc = ( '//*[@resource-id="touchableTableRow" and ' 'not(contains(@content-desc, "%, -,")) and ' 'not(contains(@content-desc, "%, 0, -,"))]' ) # First element represents "Total Streams", skip it stores = self.waiter(10, By.XPATH, stores_with_values_loc, multiple=True)[1:] valid_stores = [] for store in stores: row_data = store.get_attribute("content-desc") or "" row_parts = [part.strip() for part in row_data.split(",")] if len(row_parts) > 2 and row_parts[2] not in ("", "-"): valid_stores.append((store, row_parts[2])) assert valid_stores, \ "No POT stores with a valid 1-day value were found." # Select a random store from all valid stores random_store, table_value_for_1day = random.choice(valid_stores) context.table_value_for_1day = table_value_for_1day self.click_with_retry(3, element=random_store) def select_random_pot_store_without_data(self, context): stores_with_no_values_loc = ( '//*[@resource-id="touchableTableRow" and ' '(contains(@content-desc, "-, -, 0, -"))]' ) stores = self.waiter(10, By.XPATH, stores_with_no_values_loc, multiple=True) if stores: random_store = random.choice(stores) self.tap_after_overlap_check(random_store) context.store_without_data_is_selected = True else: context.store_without_data_is_selected = False @retry_on_stale_element() def select_total_streams_in_pot_list(self, context): self.scroll_and_assert(By.XPATH, self.st_store_row, small_scroll=True) st_store_row_elem = ( self.driver.find_element(By.XPATH, self.st_store_row)) self.tap_after_overlap_check(st_store_row_elem) def get_row_and_parts(): row = self.waiter( 10, By.XPATH, self.st_store_row, multiple=True)[0] row_data = row.get_attribute("content-desc") parts = [part.strip() for part in row_data.split(",")] return row, parts total_streams_row, parts = get_row_and_parts() if context.active_screen == "Song Screen": index = 1 else: if len(parts) <= 3: self.scroll_page(small_scroll=True, times=1) total_streams_row, parts = get_row_and_parts() index = 2 if len(parts) <= index: raise ValueError(f"Unexpected content-desc format: {parts}.") context.table_value_for_1day = parts[index] self.click_with_retry(3, element=total_streams_row) @retry_on_assertion() def tap_store_status_button(self): self.scroll_and_assert(By.XPATH, self.st_store_status_btn, max_scrolls=8, small_scroll=True) button_elem = ( self.driver.find_element(By.XPATH, self.st_store_status_btn)) self.tap_after_overlap_check(button_elem) def assert_all_sections(self, context): strategy = By.ACCESSIBILITY_ID # TODO: Temporarily disabled behind flag GO-4730 # self.assert_all_time_streams() try: self.assert_element(strategy, self.st_graph, 30) self.scroll_and_assert(strategy, self.playlist_section, 8) if "sme" not in context.app_name: self.scroll_and_assert(strategy, self.top_markets_section, 5) except TimeoutException: try: self.assert_element(By.XPATH, self.hit_a_snag_msg, 1) except TimeoutException: self.assert_element(By.XPATH, self.data_failed_to_load_msg, 1) def assert_number_of_displayed_playlists(self, context, number): if "sme" not in context.app_name: actual_num_of_playlists = \ (len(self.waiter( 10, By.XPATH, self.playlist_element, multiple=True))) assert actual_num_of_playlists == int(number), \ (f"Expected {number} playlists, " f"but found {actual_num_of_playlists}") @retry_on_assertion(max_retries=2, delay=3) def assert_number_of_displayed_playlists_exceeds(self, number): actual_num_of_playlists = ( len(self.waiter(10, By.XPATH, self.playlist_element, multiple=True))) assert actual_num_of_playlists > int(number), \ (f"Expected more than {number} playlists, " f"but found {actual_num_of_playlists}") def assert_number_of_displayed_top_market_cards(self, number): actual_num_of_cards = ( len(self.waiter(10, By.XPATH, self.top_markets_card_country_name, multiple=True))) assert actual_num_of_cards == int(number), \ f"Expected {number} playlists, but found {actual_num_of_cards}" def assert_all_time_streams(self): try: self.assert_element(self.access, self.all_time_streams, 90) except TimeoutException: try: self.assert_element(By.XPATH, self.hit_a_snag_msg, 1) except TimeoutException: self.assert_element(By.XPATH, self.data_failed_to_load_msg, 1) @staticmethod def validate_number_format(value, field_name=""): num_format_regex = r'^\d+(\.\d+)?[KM]?$' if not re.match(num_format_regex, value): raise ValueError(f"Invalid format for {field_name}: {value}") def get_all_time_streams_amount(self): amount = self.driver.find_element( By.XPATH, self.all_time_streams_amount).text self.validate_number_format(amount, "All Time Streams") return amount def get_seven_days_tiktok_creation_amount(self): amount = self.driver.find_element( By.XPATH, self.seven_days_tiktok_creations_amount).text self.validate_number_format(amount, "7 Days Tiktok Creations") return amount def get_playlists_total_amount(self): amount = self.driver.find_element( By.XPATH, self.playlists_total_amount).text self.validate_number_format(amount, "Playlists Total") return amount def click_streaming_trend_tab(self, tab): streaming_trend_tab = text_view_xpath(tab) self.waiter(15, By.XPATH, streaming_trend_tab).click() def get_all_values_from_total_column_in_streaming_trend(self): """ The method extracts all values from the 'Total' column within the currently selected tab of the Streaming Trend section. """ values = [] for row in range(1, 10): value = \ (f'//android.view.ViewGroup[@content-desc="streamingTrend"]/' f'android.view.ViewGroup[7]/android.view.ViewGroup[{row}]/' f'android.widget.TextView[1]') try: self.scroll_and_assert(By.XPATH, value) values.append(self.driver.find_element(By.XPATH, value).text) except TimeoutException: assert len(values) > 0, ('Unusual behavior, ' 'there should be at least one value.') break return values def verify_top_playlists_section_content(self): self.scroll_and_assert(self.access, self.seven_days_streams_btn, small_scroll=True, max_scrolls=20) self.waiter(10, By.XPATH, self.top_playlists_title) self.waiter(3, By.XPATH, self.top_playlists_i_icon) self.waiter(3, self.access, self.seven_days_streams_btn) self.waiter(3, self.access, self.most_recent_btn) self.waiter(3, self.access, self.followers_btn) def verify_see_all_playlists_button(self): self.scroll_and_assert(By.XPATH, self.see_all_playlists_btn, max_scrolls=20, small_scroll=True, wd_timeout=4) def verify_top_playlists_popup_content(self): self.waiter(20, By.XPATH, self.bottom_sheet) for content in self.top_playlists_content: self.waiter(3, By.XPATH, text_view_xpath(content)) self.swipe_down_to_dismiss() self.wait_for_element_invisibility(By.XPATH, self.bottom_sheet) def verify_top_charts_popup_content(self): self.waiter(10, By.XPATH, self.top_charts_i_title) self.waiter(3, By.XPATH, self.top_charts_i_content) self.click_with_retry(3, By.XPATH, self.top_charts_i_close_btn) sleep(5) def select_top_playlists_filter(self, filter_by): filter_btns = \ {'1d Streams': self.one_day_streams_btn, '7d Streams': self.seven_days_streams_btn, 'Most Recent': self.most_recent_btn, 'Followers': self.followers_btn} sleep(1) try: self.waiter(3, self.access, filter_btns[filter_by]).click() except TimeoutException: self.scroll_and_assert(self.access, filter_btns[filter_by], direction='up', small_scroll=True) self.waiter(3, self.access, filter_btns[filter_by]).click() def verify_top_markets_section_content(self, context): if "sme" not in context.app_name: self.waiter(10, By.XPATH, self.top_markets_title) self.waiter(3, By.XPATH, self.top_markets_time_period) def verify_top_markets_cards_content(self, context): if "sme" not in context.app_name: self.scroll_page() self.waiter(10, By.XPATH, self.top_markets_card_country_name) self.waiter(3, By.XPATH, self.top_markets_card_stream_count) def assert_non_catalog_track_message_displayed(self): self.waiter(10, By.XPATH, self.not_in_catalog_msg) def assert_only_non_catalog_sections_visible(self): in_catalog_sections = [ # TODO: Temporarily disabled behind flag GO-4730 # self.all_time_streams, self.st_graph, self.top_markets_sources] for section in in_catalog_sections: with pytest.raises(TimeoutException): self.scroll_and_assert(self.access, section) self.scroll_page(direction='up', times=3) self.scroll_and_assert(self.access, self.playlist_section) self.verify_see_all_playlists_button() self.waiter(3, self.access, self.most_recent_btn) self.waiter(3, self.access, self.followers_btn) if self.driver.find_elements( By.XPATH, resource_id_xpath('spotifyIcon')): self.waiter(3, By.XPATH, self.followers) else: self.wait_for_element_invisibility(By.XPATH, self.followers, 1) self.wait_for_element_invisibility(self.access, self.one_day_streams_btn, 1) self.wait_for_element_invisibility(self.access, self.seven_days_streams_btn, 1) self.wait_for_element_invisibility(self.access, self.song_stream_quantity, 1) self.scroll_and_assert(By.XPATH, self.top_charts_section) def assert_spotify_top_charts_cards(self, context, expected_charts): right_arrow_loc = resource_id_xpath('chevRightIcon') updated_date_loc = \ '//android.widget.TextView[starts-with(@text, "Updated: ")]' chart_card_loc = f"{updated_date_loc}/.." try: visible_chart_bullets = len(self.waiter( 5, By.XPATH, self.top_charts_bullets, multiple=True)) except TimeoutException: visible_chart_bullets = 1 context.visible_chart_bullets = visible_chart_bullets charts_to_check = expected_charts[:visible_chart_bullets] for chart in charts_to_check: # Ensure chart title is present self.scroll_and_assert(By.XPATH, text_view_xpath(chart)) # Ensure presence of the right arrow icon self.waiter(3, By.XPATH, right_arrow_loc) # Validate updated date field format self.scroll_and_assert(By.XPATH, updated_date_loc) updated_date = ( self.waiter(3, By.XPATH, updated_date_loc).text.strip()) date_str = updated_date.replace("Updated: ", "") try: datetime.strptime(date_str, "%d %b %Y") except ValueError as e: raise AssertionError( f"Invalid date in 'Updated:' field: {date_str}") from e # Swipe to the next chart card chart_card = self.waiter(3, By.XPATH, chart_card_loc) self.swipe_element_left(chart_card) def assert_apple_top_charts_cards(self, context, expected_charts): right_arrow_loc = resource_id_xpath('chevRightIcon') updated_date_loc = \ '//android.widget.TextView[starts-with(@text, "Updated: ")]' # Ensure chart title is present self.scroll_and_assert(By.XPATH, text_view_xpath(expected_charts[0])) # Ensure presence of the right arrow icon self.waiter(3, By.XPATH, right_arrow_loc) # Validate updated date field format self.scroll_and_assert(By.XPATH, updated_date_loc) updated_date = ( self.waiter(3, By.XPATH, updated_date_loc).text.strip()) date_str = updated_date.replace("Updated: ", "") try: datetime.strptime(date_str, "%d %b %Y") except ValueError as e: raise AssertionError( f"Invalid date in 'Updated:' field: {date_str}") from e def swipe_back_through_spotify_chart_cards(self): spotify_top_charts = self.waiter(10, By.XPATH, self.spotify_top_charts_card) for _ in range(3): self.swipe_element_right(spotify_top_charts) sleep(1) def tap_first_spotify_top_charts_card(self): try: self.click_with_retry(10, By.XPATH, self.spotify_top_charts_card) except TimeoutException: self.click_with_retry(5, By.XPATH, self.top_charts_card) def tap_first_apple_top_charts_card(self): try: self.click_with_retry(10, By.XPATH, self.apple_top_charts_card) except TimeoutException: self.click_with_retry(5, By.XPATH, self.top_charts_card) def assert_spotify_charts_page_displayed(self, context, charts_name): self.waiter(10, By.XPATH, self.spotify_charts_title) if context.visible_chart_bullets > 1: self.waiter(10, By.XPATH, content_desc_xpath(charts_name)) def assert_apple_charts_page_displayed(self, context, charts_name): self.waiter(20, By.XPATH, self.apple_charts_title) self.waiter(10, By.XPATH, text_view_xpath(charts_name)) def change_top_charts_category_to(self, context, charts_name): title = f'({text_view_xpath("Spotify Charts")})[1]' self.click_with_retry(10, By.XPATH, resource_id_xpath('chevDownIcon')) self.waiter(10, By.XPATH, title) self.click_with_retry(5, By.XPATH, resource_id_xpath(charts_name)) self.assert_spotify_charts_page_displayed(context, charts_name) def change_top_songs_sorting_to(self, sorting_option): title = text_view_xpath('Sort By') sorting_option_xpath = \ (f'//android.view.ViewGroup[@content-desc="{sorting_option}" ' f'and @resource-id="sortingOption"]') self.click_with_retry(10, By.XPATH, resource_id_xpath('chevDownIcon')) self.waiter(10, By.XPATH, title) self.click_with_retry(5, By.XPATH, sorting_option_xpath) def assert_selected_sorting(self, sorting_option): actual_sorting_loc = \ (f'//android.view.ViewGroup[@content-desc="{sorting_option}" ' f'and @resource-id="sortButton"]') self.waiter(10, By.XPATH, actual_sorting_loc) def assert_top_charts_cards_sorted_by_market_name(self): country_names_loc = \ ("//android.widget.TextView[not(starts-with(@text, '#')) and " "following-sibling::android.widget.TextView[1]" "[starts-with(@text, '#')]]") country_names = ( self.waiter(10, By.XPATH, country_names_loc, multiple=True)) country_names = [i.text for i in country_names] assert country_names[0] == 'Global' assert sorted(country_names[1:]) def assert_top_charts_cards_sorted_by_position(self): positions_loc = '//android.widget.TextView[contains (@text, "#")]' positions = self.waiter(10, By.XPATH, positions_loc, multiple=True) positions = [int(i.text.split('# ')[1]) for i in positions] assert positions == sorted(positions) def change_top_charts_sorting_to(self, option): title_loc = text_view_xpath('Sort By') market_name_option_loc = resource_id_xpath('country_name') position_option_loc = resource_id_xpath('current_position') new_entries_option_loc = resource_id_xpath('sortItem-NEW_ENTRIES') major_moves_option_loc = resource_id_xpath('sortItem-MAJOR_MOVES') options = { 'MARKET NAME': market_name_option_loc, 'NEW ENTRIES': new_entries_option_loc, 'MAJOR MOVES': major_moves_option_loc, 'POSITION': position_option_loc } self.click_with_retry(10, By.XPATH, self.top_markets_sort_btn) self.waiter(10, By.XPATH, title_loc) self.click_with_retry(10, By.XPATH, options[option]) def tap_spotify_charts_i_icon(self): self.click_with_retry(10, By.XPATH, self.spotify_charts_i_icon) def tap_apple_charts_i_icon(self): self.click_with_retry(10, By.XPATH, self.apple_charts_i_icon) def close_i_popup(self): close_btn = resource_id_xpath('closeButton') self.click_with_retry(3, By.XPATH, close_btn) self.wait_for_element_invisibility(By.XPATH, close_btn) def verify_spotify_charts_i_popup_content_for(self, charts_name): title = f'({text_view_xpath(charts_name)})[1]' content = text_view_xpath( f'Spotify {charts_name} streams show the chart qualifying streams ' f'for each market. This data comes directly from Spotify Charts, ' f'so the value may differ slightly from our typical streaming ' f'numbers.') self.waiter(10, By.XPATH, title) if charts_name in ['Daily Top 200', 'Weekly Top 200']: self.waiter(10, By.XPATH, content) elif charts_name == 'Daily Viral 100': self.waiter(10, By.XPATH, resource_id_xpath('content')) else: raise ValueError(f"Unsupported chart name: {charts_name}") self.close_i_popup() def verify_apple_charts_i_popup_content_for(self, charts_name): title = f'({text_view_xpath(charts_name)})[1]' self.waiter(10, By.XPATH, title) if charts_name == 'Daily Top 100': self.waiter(10, By.XPATH, resource_id_xpath('content')) self.close_i_popup()