"""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 SoundCloudStatsSFExecutor def get_args(f, params=False): """Return passed arguments.""" 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 substrings.""" 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, 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_followers_count': 503, 'last_week_followers_count': 455, 'weekly_change_in_followers_count': 156, 'percentage_change_in_followers_count': 0.34, 'current_followings_count': 79, 'last_week_followings_count': 78, 'weekly_change_in_followings_count': 1, 'percentage_change_in_followings_count': 0, 'current_tracks_count': 8, 'last_week_tracks_count': 8, 'weekly_change_in_tracks_count': 0, 'percentage_change_in_tracks_count': 0, 'current_plays_count': 23988, 'last_week_plays_count': 8967, 'weekly_change_in_plays_count': 7665, 'percentage_change_in_plays_count': 0.57, 'current_likes_count': 1298, 'last_week_likes_count': 585, 'weekly_change_in_likes_count': 649, 'percentage_change_in_likes_count': 0.67, 'current_reposts_count': 50, 'last_week_reposts_count': 45, 'weekly_change_in_reposts_count': 2, 'percentage_change_in_reposts_count': 0, 'plays_engagement': 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', 'shazam': None, 'tiktok': 'To check corresponding links', '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 = SoundCloudStatsSFExecutor(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.SoundCloudStatsSFExecutor' 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://soundcloud.com/cijikin']] @pytest.fixture def mock_select_artists_links(): """Yield SELECT artists' links SQL query result.""" select_path = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' '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 = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' 'create_artist_link_table') with patch(create_path) as create: yield create @pytest.fixture def mock_create_soundcloud_statistics(): """Yield CREATE soundcloud_statistics table SQL query result.""" create_path = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' 'create_soundcloud_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_create_staging_raw_soundcloud_statistics(): """Yield CREATE staging_raw table SQL query result.""" create_path = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' 'create_staging_raw_soundcloud_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 = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' '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 = ( 'soundcloud_with_selenium.SoundCloudStatsSFExecutor.' 'select_weekly_change') with patch(select_path) as select: select.return_value = { 'followers_count': 45, 'followings_count': 0, 'tracks_count': 6, 'plays_count': 4, 'likes_count': 66, 'reposts_count': 4} yield select @pytest.fixture def mock_compare_current_and_last_week_stats(): """Yield mock_compare_current_and_last_week_stats method.""" calc_path = ( 'soundcloud_with_selenium.SoundCloudTracker.' '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 = ( 'soundcloud_with_selenium.SoundCloudTracker.create_tables') with patch(create_path) as create: yield create @pytest.fixture def mock_get_and_update_stats(): """Yield get_all_stats_and_update method from tracker class.""" get_path = ( 'soundcloud_with_selenium.SoundCloudTracker.get_and_update_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_stat_from_(): """Yield get_stat_from_ method from helpers.""" get_path = ( 'soundcloud_with_selenium.SoundCloudTracker.get_stat_from_') with patch(get_path) as get: yield get class MockXPath: """Mocked XPath class.""" def __iter__(self): """Return __iter__ method for XPath.""" class Iterator: def __init__(self): self.index = 0 def __next__(self): if self.index < 2: self.index += 1 return '897' raise StopIteration return Iterator() @property def attrib(self): """Return attribute to parse.""" # 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): """Return text to parse.""" return 'text986756' def get_attribute(self, attribute): """Get attribute by its name.""" return self.attrib[attribute] @pytest.fixture def mock_get_artist_info(): """Yield get_artist_info method from tracker class.""" get_path = ( 'soundcloud_with_selenium.SoundCloudTracker.get_artist_info') with patch(get_path) as get: yield get @pytest.fixture def mock_get_tracks_info(): """Yield get_tracks_info method from tracker class.""" get_path = ( 'soundcloud_with_selenium.SoundCloudTracker.get_tracks_info') with patch(get_path) as get: yield get @pytest.fixture def mock_set_webdriver(): """Yield get_posts_stats method from tracker class.""" set_path = ( 'soundcloud_with_selenium.set_webdriver') with patch(set_path) as _set: yield _set @pytest.fixture def mock_get_driver_of_artist_page(): """Yield get_driver_of_artist_page method from tracker class.""" set_path = ( 'soundcloud_with_selenium.SoundCloudTracker.get_driver_of_artist_page') with patch(set_path) as _set: yield _set @pytest.fixture def mock_abbreviated_value_to_int(): """Yield abbreviated_value_to_int method from helpers.""" path = ( 'helpers.abbreviated_value_to_int') with patch(path) as f: yield f