"""Conftest file being observed by tests.""" import datetime from unittest.mock import MagicMock, patch from freezegun import freeze_time import pytest from snowflake import connector from snowflake_executor import SpotifyStatsSFExecutor def get_args(f, params=False): """Get args dict. Args: f (func): a called function. params (bool): to get params or not. """ if params: return f.call_args[1]['params'] return f.call_args[0] def check_query(query: str, contains: list, check=True): """Check if query contains. Args: query (str): a SQL query to check. contains (list): list of substrings to check. check (bool): check if contained or not. """ for substring in contains: check = check and (substring in query) return check @freeze_time('1946-09-27') def artist_statistics_from_table(mock_argument) -> list: """Return processed dataset example.""" stats = [['cijikin', None, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, datetime.datetime.today(), datetime.datetime.today(), datetime.datetime.today()], ] return stats @freeze_time('1946-09-27') def artist_last_stats(last_week_processing_date=datetime.date.today( ) - datetime.timedelta(days=4)): """Return artist's last run statistics.""" return {'cijikin': { 'id': None, 'current_monthly_listeners': 0, 'last_week_monthly_listeners': 0, 'weekly_change_in_monthly_listeners': 0, 'percentage_change_in_monthly_listeners': 0, 'current_followers': 0, 'last_week_followers': 0, 'weekly_change_in_followers': 0, 'percentage_change_in_followers': 0, 'current_popularity': 0, 'last_week_popularity': 0, 'editorial_playlists_adds': 0, 'editorial_playlists_adds_count': 0, 'last_week_editorial_playlists_adds': 0, 'last_week_editorial_playlists_adds_count': 0, 'notable_playlists_adds': 0, 'notable_playlists_adds_count': 0, 'last_week_notable_playlists_adds': 0, 'last_week_notable_playlists_adds_count': 0, 'the_latest_release': datetime.date.today( ) - datetime.timedelta(days=457), 'last_processing_date': datetime.date.today( ) - datetime.timedelta(days=1), 'last_week_processing_date': last_week_processing_date}} @pytest.fixture def processed_links() -> dict: """Return processed dataset example.""" links = {'artist': 'cijikin', 'instagram': 'https://www.instagram.com/cijikin/', 'spotify': 'https://open.spotify.com/artist/' '060CEbkq3Bubu0AvJu4NKf', 'youtube': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg', 'soundcloud': 'https://soundcloud.com/cijikin', 'tiktok': 'To check corresponding links', 'shazam': None, 'id': None, 'date': None} return links @pytest.fixture def sf_config_mock(): """Fixture returning the dict with Snowflake connection params.""" return { 'account': 'test_acc', 'role': 'test_role', 'host': 'test_host', 'warehouse': 'test_wh', 'port': 10, 'user': 'test_user', 'password': 'test_pass', 'db': 'test_db', 'schema': 'test_schema' } @pytest.fixture def mock_executor(sf_config_mock, monkeypatch): """Yield executor.""" connect_mock = MagicMock() monkeypatch.setattr(connector, 'connect', connect_mock) executor = SpotifyStatsSFExecutor(sf_config_mock) with patch.object(executor, 'execute', wraps=executor.execute) as \ executor.ex_mock: executor.fetchall = MagicMock() yield executor @pytest.fixture def mock_executor_context(): """Yield executor context.""" sf_executor_class_path = 'snowflake_executor.SpotifyStatsSFExecutor' with patch(sf_executor_class_path) as sf_executor: mock_executor_context = sf_executor.return_value.__enter__.return_value yield mock_executor_context @pytest.fixture def get_artists_links(): """Fixture returning the list of artists' links.""" return [['cijikin', 'https://open.spotify.com/artist/060CEbkq3Bubu0AvJu4NKf']] @pytest.fixture def mock_select_artists_links(): """Yield SELECT artists' links SQL query result.""" select_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.select_artists_links') with patch(select_path) as select: select.return_value = get_artists_links yield select @pytest.fixture def mock_create_artist_link_table(): """Yield CREATE artists' links table SQL query result.""" create_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.create_artist_link_table') with patch(create_path) as create: yield create @pytest.fixture def mock_create_spotify_statistics(): """Yield CREATE spotify_statistics table SQL query result.""" create_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.' 'create_spotify_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_create_staging_raw_spotify_statistics(): """Yield CREATE staging_raw_spotify_statistics table SQL query result.""" create_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.' 'create_staging_raw_spotify_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_delete_from_staging_raw(): """Yield DELETE from staging_raw table SQL query result.""" delete_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.delete_from_staging_raw') with patch(delete_path) as delete: yield delete @pytest.fixture def mock_select_weekly_change(): """Yield DELETE from select_weekly_change table SQL query result.""" select_path = ( 'spotify_with_spotipy.SpotifyStatsSFExecutor.select_weekly_change') with patch(select_path) as select: yield select @pytest.fixture def mock_compare_current_and_last_week_stats(): """Yield mock_compare_current_and_last_week_stats method.""" calc_path = ( 'spotify_with_spotipy.SpotifyTracker.' 'compare_current_and_last_week_stats') with patch(calc_path) as calc: yield calc @pytest.fixture def mock_create_tables(): """Yield create_tables method from tracker class.""" create_path = ( 'spotify_with_spotipy.SpotifyTracker.create_tables') with patch(create_path) as create: yield create @pytest.fixture def mock_get_all_stats_and_update(): """Yield get_all_stats_and_update method from tracker class.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.get_all_stats_and_update') with patch(get_path) as get: yield get class MockDom: """DOM Mock class.""" def __init__(self, text): """DOM mock attributes.""" self.__text__ = text pass def xpath(self, *args, **kwargs): """Mock xpath method.""" return [ MockDom('220,083 monthly listeners'), ] @property def attrib(self): """Get attributes.""" # spaces are need to check how get_artist_info # processes them and numbers itself return {'title': ' 43 6 likes', 'datetime': '2022-03-22T'} @property def text(self): """Get text.""" return self.__text__ @pytest.fixture def mock_get_dom_of_artist_page(): """Yield get_dom_of_artist_page method.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.get_dom_of_artist_page') with patch(get_path) as get: get.return_value = MockDom yield get @pytest.fixture def mock_spotify(): """Yield get_dom_of_artist_page method.""" get_path = ( 'spotify_with_spotipy.SpotifyClientCredentials') with patch(get_path) as get: yield get @pytest.fixture def mock_get_monthly_listeners(): """Yield get_monthly_listeners method from tracker class.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.get_monthly_listeners') with patch(get_path) as get: yield get @pytest.fixture def mock_get_common_stats(): """Yield get_common_stats method from tracker class.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.get_common_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_playlists_stats(): """Yield get_playlists_stats method from tracker class.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.get_playlists_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_update_new_artists_stats(): """Yield update_all_stats method from tracker class.""" get_path = ( 'spotify_with_spotipy.SpotifyTracker.update_new_artists_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_set_webdriver(): """Yield get_posts_stats method from tracker class.""" set_path = ( 'spotify_with_spotipy.set_webdriver') with patch(set_path) as _set: yield _set @pytest.fixture def mock_abbreviated_value_to_int(): """Yield abbreviated_value_to_int method.""" path = ( 'spotify_with_spotipy.abbreviated_value_to_int') with patch(path) as f: yield f @pytest.fixture def mock_update_staging_raw_table(): """Yield update_staging_raw_table method.""" path = ( 'spotify_with_spotipy.SpotifyTracker.update_staging_raw_table') with patch(path) as f: yield f @pytest.fixture def mock_load_profile(): """Yield load_profile method.""" path = ( 'spotify_with_spotipy.SpotifyTracker.load_profile') with patch(path) as f: f.return_value = [{ 'followers': { 'total': 400 }, 'popularity': 40, }, ] yield f