"""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 InstagramStatsSFExecutor def get_args(f, params=False): """Get function 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.""" 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', 'https://www.instagram.com/cijikin/', '', 37, 0, 37, 0, 3, 0, 3, 220, 220, 0, 0, 14, 0, 0, 0, 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': { 'link': 'https://www.instagram.com/cijikin/', 'verified': None, 'current_followers': 54, 'weekly_change_in_followers': 0, 'last_week_followers': 37, 'percentage_change_in_followers': 0.46, 'current_posts': 4, 'weekly_change_in_posts': 0, 'last_week_posts': 3, 'current_likes': 275, 'weekly_change_in_likes': 30, 'percentage_change_in_likes': 0, 'current_comments': 16, 'weekly_change_in_comments': 2, 'current_engagement': 0, 'last_week_engagement': 0, '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 = InstagramStatsSFExecutor(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.InstagramStatsSFExecutor' 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://www.instagram.com/cijikin/', 'test_id', 'https://open.spotify.com/artist/060CEbkq3Bubu0AvJu4NKf', 'https://www.youtube.com/channel/UC5kuP-o0jopVpHCD8KSrPpg', 'https://soundcloud.com/cijikin', None, None]] @pytest.fixture def mock_select_artists_links(): """Yield SELECT artists' links SQL query result.""" select_path = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.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 = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.create_artist_link_table') with patch(create_path) as create: yield create @pytest.fixture def mock_update_artist_link_table(): """Yield UPDATE artists' links table SQL query result.""" update_path = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.update_artist_link_table') with patch(update_path) as update: yield update @pytest.fixture def mock_insert_artist_link_table(): """Yield INSERT artists' links table SQL query result.""" insert_path = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.insert_artist_link_table') with patch(insert_path) as insert: yield insert @pytest.fixture def mock_create_instagram_statistics(): """Yield CREATE instagram_statistics table SQL query result.""" create_path = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.create_instagram_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_create_staging_raw_instagram_statistics(): """Yield CREATE staging_raw_instagram_statistics table SQL query result.""" create_path = ( 'instagram_with_instaloader.InstagramStatsSFExecutor' '.create_staging_raw_instagram_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 = ( 'instagram_with_instaloader.' 'InstagramStatsSFExecutor.delete_from_staging_raw') with patch(delete_path) as delete: yield delete @pytest.fixture def mock_compare_current_and_last_week_stats(): """Yield mock_compare_current_and_last_week_stats method.""" calc_path = ( 'instagram_with_instaloader.InstagramTracker.' 'compare_current_and_last_week_stats') with patch(calc_path) as calc: yield calc @pytest.fixture def mock_load_profile(): """Yield Instagram artist's profile.""" class MockProfile(): def __init__(self): pass @property def followers(self): return 5603 @property def followees(self): return 57 @property def is_verified(self): return True profile_path = ( 'instagram_with_instaloader.InstagramTracker.load_profile') with patch(profile_path) as profile: profile.return_value = MockProfile() yield profile @pytest.fixture def mock_create_tables(): """Yield create_tables method from tracker class.""" create_path = ( 'instagram_with_instaloader.InstagramTracker.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 = ( 'instagram_with_instaloader.InstagramTracker.get_all_stats_and_update') 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 = ( 'instagram_with_instaloader.InstagramTracker.get_common_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_posts_stats(): """Yield get_posts_stats method from tracker class.""" get_path = ( 'instagram_with_instaloader.InstagramTracker.get_posts_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_unreachable_profiles(): """Yield get_unreachable_profiles method from tracker class.""" get_path = ( 'instagram_with_instaloader.InstagramTracker.get_unreachable_profiles') with patch(get_path) as get: get.return_value = [{'artist': 'cijikin', 'instagram': 'cijikin', 'reason': 'test'}] yield get @pytest.fixture def mock_sent_a_message_to_slack(): """Yield sent_a_message_to_slack method from fÄngare module.""" get_path = ( 'fangare.catchers.sent_a_message_to_slack') with patch(get_path) as get: get.return_value = {'status': 'sent'} yield get @pytest.fixture def test_sheet(): """Fixture returning the GS data.""" return { 'properties': { 'title': 'Test Sheet', 'locale': 'en_US', 'autoRecalc': 'ON_CHANGE', 'timeZone': 'Europe/Bucharest', 'defaultFormat': { 'backgroundColor': {'red': 1, 'green': 1, 'blue': 1}, 'padding': {'top': 2, 'right': 3, 'bottom': 2, 'left': 3}, 'verticalAlignment': 'BOTTOM', 'wrapStrategy': 'OVERFLOW_CELL', 'textFormat': { 'foregroundColor': {}, 'fontFamily': 'arial,sans,sans-serif', 'fontSize': 10, 'bold': False, 'italic': False, 'strikethrough': False, 'underline': False, 'foregroundColorStyle': {'rgbColor': {}}}, 'backgroundColorStyle': { 'rgbColor': {'red': 1, 'green': 1, 'blue': 1}}}, 'spreadsheetTheme': { 'primaryFontFamily': 'Arial', 'themeColors': [{ 'colorType': 'TEXT', 'color': {'rgbColor': {}}}, {'colorType': 'BACKGROUND', 'color': { 'rgbColor': { 'red': 1, 'green': 1, 'blue': 1}}}, {'colorType': 'ACCENT1', 'color': { 'rgbColor': { 'red': 0.25882354, 'green': 0.52156866, 'blue': 0.95686275}}}, {'colorType': 'ACCENT2', 'color': {'rgbColor': { 'red': 0.91764706, 'green': 0.2627451, 'blue': 0.20784314}}}, {'colorType': 'ACCENT3', 'color': { 'rgbColor': { 'red': 0.9843137, 'green': 0.7372549, 'blue': 0.015686275}}}, {'colorType': 'ACCENT4', 'color': {'rgbColor': { 'red': 0.20392157, 'green': 0.65882355, 'blue': 0.3254902}}}, {'colorType': 'ACCENT5', 'color': { 'rgbColor': {'red': 1, 'green': 0.42745098, 'blue': 0.003921569}}}, {'colorType': 'ACCENT6', 'color': {'rgbColor': { 'red': 0.27450982, 'green': 0.7411765, 'blue': 0.7764706}}}, {'colorType': 'LINK', 'color': { 'rgbColor': { 'red': 0.06666667, 'green': 0.33333334, 'blue': 0.8}}}]}}, 'sheets': [ {'data': [{'rowData': [ {'values': [ {}, {'userEnteredValue': {'stringValue': 'Instagram'}, 'effectiveValue': {'stringValue': 'Instagram'}, 'formattedValue': 'Instagram'}, {'userEnteredValue': {'stringValue': 'Spotify'}, 'effectiveValue': {'stringValue': 'Spotify'}, 'formattedValue': 'Spotify'}, {'userEnteredValue': {'stringValue': 'YouTube'}, 'effectiveValue': {'stringValue': 'YouTube'}, 'formattedValue': 'YouTube'}, {'userEnteredValue': {'stringValue': 'SoundCloud'}, 'effectiveValue': {'stringValue': 'SoundCloud'}, 'formattedValue': 'SoundCloud'}, {'userEnteredValue': {'stringValue': 'Shazam'}, 'effectiveValue': {'stringValue': 'Shazam'}, 'formattedValue': 'Shazam'}, {'userEnteredValue': {'stringValue': 'TikTok'}, 'effectiveValue': {'stringValue': 'TikTok'}, 'formattedValue': 'TikTok'}]}, {'values': [ {'userEnteredValue': {'stringValue': 'cijikin'}, 'effectiveValue': {'stringValue': 'cijikin'}, 'formattedValue': 'cijikin'}, {'userEnteredValue': { 'stringValue': 'https://www.instagram.com/cijikin/'}, 'effectiveValue': { 'stringValue': 'https://www.instagram.com/cijikin/' }, 'formattedValue': 'https://www.instagram.com/cijikin/', 'userEnteredFormat': { 'textFormat': { 'link': {'uri': 'https://www.instagram.' 'com/cijikin/'}}}, 'hyperlink': 'https://www.instagram.com/cijikin/'}, {'userEnteredValue': { 'stringValue': 'https://open.spotify.com/' 'artist/060CEbkq3Bubu0AvJu4NKf'}, 'effectiveValue': { 'stringValue': 'https://open.spotify.com/' 'artist/060CEbkq3Bubu0AvJu4NKf'}, 'formattedValue': 'https://open.spotify.com/' 'artist/060CEbkq3Bubu0AvJu4NKf', 'userEnteredFormat': {'textFormat': { 'link': {'uri': 'https://open.spotify.com/' 'artist/060CEbkq3Bubu0AvJu4NKf'}}}, 'hyperlink': 'https://open.spotify.com/artist/' '060CEbkq3Bubu0AvJu4NKf'}, {'userEnteredValue': { 'stringValue': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg'}, 'effectiveValue': { 'stringValue': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg'}, 'formattedValue': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg', 'userEnteredFormat': {'textFormat': { 'link': { 'uri': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg'}}}, 'hyperlink': 'https://www.youtube.com/channel/' 'UC5kuP-o0jopVpHCD8KSrPpg'}, {'userEnteredValue': { 'stringValue': 'https://soundcloud.com/cijikin'}, 'effectiveValue': { 'stringValue': 'https://soundcloud.com/cijikin'}, 'formattedValue': 'https://soundcloud.com/cijikin', 'userEnteredFormat': { 'textFormat': {'link': { 'uri': 'https://soundcloud.com/cijikin'}}}, 'hyperlink': 'https://soundcloud.com/cijikin'}, {}, {'userEnteredValue': { 'stringValue': 'To check corresponding links'}, 'effectiveValue': { 'stringValue': 'To check corresponding links'}, 'formattedValue': 'To check corresponding links'} ]}]}]}]}