"""Unit tests for Instagram Statistics Workflow.""" import datetime import os from argparse import Namespace from unittest import mock from unittest.mock import MagicMock import config from conftest import artist_last_stats, artist_statistics_from_table from freezegun import freeze_time import instagram_with_instaloader as instagram import pytest as pytest class TestGSConnector: """Test Google Sheets Connector.""" sheet_id = '1oYuiqi3ou7CQyz2zemPiS73-bq2NIOTR27SpRUFrGTw' # test sheet @pytest.fixture def instagram_link(self): """Return unprocessed dataset example.""" unprocessed_link = { '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/'} return unprocessed_link @mock.patch('instagram_with_instaloader.get_google_sheet') def test_get_google_sheet(self, mock_get_google_sheet, instagram_link, test_sheet): """Test get_google_sheet method checking connection to GS.""" mock_get_google_sheet.return_value = test_sheet test_sheet = instagram.get_google_sheet(self.sheet_id) instagram_link = test_sheet['sheets'][0]['data'][0][ 'rowData'][1]['values'][1] assert test_sheet['properties']['title'] == 'Test Sheet' assert instagram_link == instagram_link def test_download_artists_links_from_google_sheets(self, processed_links): """Test download_artists_links_from_google_sheets.""" links = instagram.download_artists_links_from_google_sheets( self.sheet_id) assert links[0] == processed_links def set_cookies_and_import_session(): """Read cookies and import session.""" cookies = instagram.get_cookies( os.path.normpath(os.getcwd() + os.sep + os.pardir) + '/Cookies_') config.instagram['username'] = 'veeetements' config.instagram['password'] = 'instaacc' config.SPREADSHEET_ID = TestGSConnector.sheet_id session = instagram.import_session(cookies, session=None) return cookies, session class TestInstagramTracker: """Test Instagram Tracker related methods.""" def test_get_username_from_link(self, processed_links): """Test get_username_from_link method with reverse parameter also.""" username = instagram.get_username_from_link( processed_links['instagram']) link = instagram.get_username_from_link(username, reverse=True) assert username == 'cijikin' assert link == processed_links['instagram'] @mock.patch('instagram_with_instaloader.import_session', MagicMock()) def test_create_tables_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_instagram_statistics, mock_create_instagram_statistics, mock_insert_artist_link_table, mock_update_artist_link_table, mock_create_artist_link_table, mock_select_artists_links, mock_executor, processed_links, get_artists_links): """Test create_tables working correctly during the very first run.""" mock_select_artists_links.return_value = get_artists_links cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) tracker.create_tables() called_methods = \ [mock_create_artist_link_table.called, not mock_update_artist_link_table.called, mock_insert_artist_link_table.called, mock_select_artists_links.called, mock_create_instagram_statistics.called, mock_delete_from_staging_raw.called, mock_create_staging_raw_instagram_statistics.called] for call in called_methods: assert call assert tracker.__first_time__ assert 'cijikin' in tracker.get_profiles().keys() @mock.patch('instagram_with_instaloader.' 'InstagramTracker.update_artists_table', MagicMock()) @mock.patch('instagram_with_instaloader.' 'InstagramStatsSFExecutor.select_artists_stats', artist_statistics_from_table) @mock.patch('instagram_with_instaloader.import_session', MagicMock()) def test_create_tables_not_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_instagram_statistics, mock_create_instagram_statistics, mock_insert_artist_link_table, mock_update_artist_link_table, mock_create_artist_link_table, mock_select_artists_links, mock_executor, processed_links, get_artists_links): """Test create_tables working correctly if the run's not first.""" mock_select_artists_links.return_value = get_artists_links mock_create_artist_link_table. \ return_value, mock_create_staging_raw_instagram_statistics. \ return_value, mock_create_instagram_statistics.return_value = \ [['already exists']] * 3 cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) tracker.create_tables() called_methods = \ [mock_create_artist_link_table.called, not mock_update_artist_link_table.called, not mock_insert_artist_link_table.called, mock_select_artists_links.called, mock_create_instagram_statistics.called, mock_delete_from_staging_raw.called, mock_create_staging_raw_instagram_statistics.called] for call in called_methods: assert call assert not tracker.__first_time__ assert 'cijikin' in tracker.get_profiles() @freeze_time('1946-10-01') @mock.patch('instagram_with_instaloader.' 'InstagramStatsSFExecutor.select_weekly_change') @mock.patch('instagram_with_instaloader.import_session', MagicMock()) def test_compare_current_and_last_week_stats_for_weekly_statistics( self, mock_select_weekly_change, mock_executor, processed_links): """Test correctness of comparing method calculations.""" def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'followers': 7, 'likes': 17, 'comments': 1} else: return {'followers': 19, 'likes': 30, 'comments': 2} mock_select_weekly_change.side_effect = weekly_side_effect current_followers = 245 cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) calculations = tracker.compare_current_and_last_week_stats( current_followers, 'followers', artist_last_stats(), 'cijikin') artist_lasts = artist_last_stats()['cijikin'] current_weekly_change = \ current_followers - artist_lasts['current_followers'] + \ weekly_side_effect('cijikin')['followers'] calculations_to_assert = [ calculations['last_week_followers'] == artist_lasts['last_week_followers'], calculations['daily_change_in_followers'] == current_followers - artist_lasts['current_followers'], calculations['current_weekly_change_in_followers'] == calculations['daily_change_in_followers'] + weekly_side_effect('cijikin')['followers'], calculations['weekly_change_in_followers'] == weekly_side_effect(first_day_of_period=-6)['followers'], calculations['percentage_change_in_followers'] == current_weekly_change / ( current_followers - current_weekly_change), calculations['acceleration_followers'] == ( calculations['daily_change_in_followers'] + weekly_side_effect('cijikin')['followers'] - weekly_side_effect(first_day_of_period=-6)['followers']) / weekly_side_effect(first_day_of_period=-6)[ 'followers'], calculations['last_week_processing_date'] == artist_lasts['last_week_processing_date']] for calculation in calculations_to_assert: assert calculation @freeze_time('1946-10-01') @mock.patch('instagram_with_instaloader.' 'InstagramStatsSFExecutor.select_weekly_change') @mock.patch('instagram_with_instaloader.import_session', MagicMock()) def test_compare_current_and_last_week_stats_for_week_period_statistics( self, mock_select_weekly_change, mock_executor, processed_links): """Test correctness of comparing method calculations for posts.""" current_posts = 8 cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'posts': 7, 'engagement': 0} else: return {'posts': 19, 'engagement': 0} mock_select_weekly_change.side_effect = weekly_side_effect calculations = tracker.compare_current_and_last_week_stats( current_posts, 'posts', artist_last_stats( last_week_processing_date=datetime.date.today( ) - datetime.timedelta(days=7)), 'cijikin') artist_lasts = artist_last_stats( last_week_processing_date=datetime.date.today( ) - datetime.timedelta(days=7))['cijikin'] # checking the calculation when week period ends current_weekly_change = \ current_posts - artist_lasts['current_posts'] + weekly_side_effect( 'cijikin')['posts'] assert mock_select_weekly_change.called calculations_to_assert = [ calculations['last_week_posts'] == artist_lasts['current_posts'], calculations['daily_change_in_posts'] == current_posts - artist_lasts['current_posts'], calculations['current_weekly_change_in_posts'] == calculations['daily_change_in_posts'] + weekly_side_effect('cijikin')['posts'], calculations['weekly_change_in_posts'] == weekly_side_effect(first_day_of_period=-6)['posts'], calculations['percentage_change_in_posts'] == current_weekly_change / (current_posts - current_weekly_change), calculations['acceleration_posts'] == 0, calculations['last_week_processing_date'] != artist_lasts['last_week_processing_date'] ] for calculation in calculations_to_assert: assert calculation @mock.patch('instagram_with_instaloader.import_session', MagicMock()) @mock.patch('instagram_with_instaloader.' 'InstagramTracker.get_profiles', artist_last_stats) def test_get_common_stats( self, mock_compare_current_and_last_week_stats, mock_load_profile, mock_executor, mock_executor_context): """Test get_common_stats method.""" cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) mock_compare_current_and_last_week_stats.return_value = MagicMock() for artist in artist_last_stats(): tracker.get_common_stats(artist) assert mock_load_profile.called assert mock_compare_current_and_last_week_stats.called @mock.patch( 'instagram_with_instaloader.import_session', MagicMock()) @mock.patch('instagram_with_instaloader.' 'InstagramTracker.update_new_artists_stats') @mock.patch('instagram_with_instaloader.' 'InstagramTracker.update_missing_days_stats') @mock.patch('instagram_with_instaloader.' 'InstagramTracker.get_profiles', artist_last_stats) def test_get_all_stats_and_update( self, mock_update_new_artists_stats, mock_update_missing_days_stats, mock_get_posts_stats, mock_get_common_stats, mock_executor, mock_executor_context): """Test get_all_stats_and_update method.""" cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) tracker.__last_week_stats__.update(dict(cijikin={})) tracker.get_all_stats_and_update() assert mock_get_common_stats.called assert mock_get_posts_stats.called assert mock_update_missing_days_stats.called assert mock_update_new_artists_stats.called @mock.patch('instagram_with_instaloader.import_session', MagicMock()) def test_perform_tracking( self, mock_get_unreachable_profiles, mock_create_tables, mock_sent_a_message_to_slack, mock_get_all_stats_and_update, mock_executor, mock_executor_context): """Test perform_tracking method.""" cookies, session = set_cookies_and_import_session() tracker = instagram.InstagramTracker( Namespace(cookiefile=cookies, sessionfile=None)) tracker.perform_tracking() called_methods = [mock_create_tables.called, mock_get_all_stats_and_update.called, mock_sent_a_message_to_slack.called] for method in called_methods: assert method