"""Unit tests for TikTok Statistics Workflow.""" import datetime from unittest import mock from freezegun import freeze_time import tiktok_with_requests import tiktok_with_requests as tiktok from conftest import artist_last_stats, artist_statistics_from_table, get_tiktok_stats class TestTikTokTracker: """Test TikTok Tracker related methods.""" def test_create_tables_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_tiktok_statistics, mock_create_tiktok_statistics, mock_select_artists_links, mock_executor, processed_links, get_artists_links): """Test create_tables method working correctly during the very first run.""" mock_select_artists_links.return_value = get_artists_links tracker = tiktok.TikTokTracker() tracker.create_tables() called_methods = \ [mock_select_artists_links.called, mock_create_tiktok_statistics.called, mock_delete_from_staging_raw.called, mock_create_staging_raw_tiktok_statistics.called] for call in called_methods: assert call assert tracker.__first_time__ assert 'cijikin' in tracker.get_profiles @mock.patch('tiktok_with_requests.TikTokStatsSFExecutor.select_artists_stats', artist_statistics_from_table) def test_create_tables_not_for_the_first_time( self, mock_delete_from_staging_raw, mock_create_staging_raw_tiktok_statistics, mock_create_tiktok_statistics, mock_select_artists_links, mock_executor, processed_links, get_artists_links): """Test create_tables method working correctly if the run is not first.""" mock_select_artists_links.return_value = get_artists_links mock_create_staging_raw_tiktok_statistics.return_value, \ mock_create_tiktok_statistics.return_value = [['already exists']] * 2 tracker = tiktok.TikTokTracker() tracker.create_tables() called_methods = \ [mock_select_artists_links.called, mock_create_tiktok_statistics.called, mock_delete_from_staging_raw.called, mock_create_staging_raw_tiktok_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('tiktok_with_requests.TikTokStatsSFExecutor.select_weekly_change') def test_compare_current_and_last_week_stats_for_weekly_statistics( self, mock_select_weekly_change, mock_executor, processed_links): """Test correctness of compare_current_and_last_week_stats method calculations.""" def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'followers': 9, 'followings': 0, 'likes': 0} else: return {'followers': 6, 'followings': 0, 'likes': 0} mock_select_weekly_change.side_effect = weekly_side_effect current_followers = 578 tiktok_with_requests.today = datetime.date.today() tracker = tiktok.TikTokTracker() calculations = tracker.compare_current_and_last_week_stats( current_followers, 'followers', artist_last_stats(), 'cijikin') artist_lasts = artist_last_stats()['cijikin'] assert calculations == { 'current_followers': current_followers, 'last_week_followers': artist_lasts['last_week_followers'], 'daily_change_in_followers': current_followers - artist_lasts[ 'current_followers'], 'current_weekly_change_in_followers': calculations['daily_change_in_followers'] + weekly_side_effect('cijikin')['followers'], 'weekly_change_in_followers': weekly_side_effect(first_day_of_period=-6)['followers'], 'percentage_change_in_followers': artist_lasts[ 'percentage_change_in_followers'], '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'], 'last_processing_date': datetime.date(1946, 10, 1), 'last_week_processing_date': artist_lasts['last_week_processing_date']} @freeze_time('1946-10-01') @mock.patch('tiktok_with_requests.TikTokStatsSFExecutor.select_weekly_change') def test_compare_current_and_last_week_stats_for_week_period_statistics( self, mock_select_weekly_change, mock_executor, processed_links): """Test correctness of compare_current_and_last_week_stats method calculations for posts.""" def weekly_side_effect(*args, **kwargs): if -6 in list(kwargs.values()): return {'followers': 9, 'followings': 0, 'likes': 0} else: return {'followers': 6, 'followings': 0, 'likes': 0} tiktok_with_requests.today = datetime.date.today() tracker = tiktok.TikTokTracker() current_followers = 25409 lasts = artist_last_stats(last_week_processing_date= datetime.date.today() - datetime.timedelta(days=7)) mock_select_weekly_change.side_effect = weekly_side_effect calculations = tracker.compare_current_and_last_week_stats( current_followers, 'followers', lasts, 'cijikin') artist_lasts = lasts['cijikin'] # checking the calculation when week period ends assert mock_select_weekly_change.called assert calculations == { 'current_followers': current_followers, 'last_week_followers': artist_lasts['current_followers'], 'daily_change_in_followers': current_followers - artist_lasts['current_followers'], 'current_weekly_change_in_followers': calculations['daily_change_in_followers'] + weekly_side_effect('cijikin')['followers'], 'weekly_change_in_followers': weekly_side_effect(first_day_of_period=-6)['followers'], 'percentage_change_in_followers': (current_followers - artist_lasts['last_week_followers']) / artist_lasts['last_week_followers'], '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'], 'last_processing_date': datetime.date.today(), 'last_week_processing_date': datetime.date.today() } @mock.patch('tiktok_with_requests.TikTokTracker.get_profiles', artist_last_stats()) @mock.patch('tiktok_with_requests.get_tiktok_stats', get_tiktok_stats) def test_process_tiktok_stats(self, mock_abbreviated_value_to_int, mock_executor_context, mock_compare_current_and_last_week_stats, mock_executor): """Test get_artist_info method.""" tracker = tiktok.TikTokTracker() for artist in artist_last_stats(): tracker.process_tiktok_stats(artist) stats, index = get_tiktok_stats(mock_argument=0), 0 stats.update(followings=None) called_methods = \ [mock_compare_current_and_last_week_stats.called, mock_compare_current_and_last_week_stats.call_count == 3, mock_compare_current_and_last_week_stats.call_args[0][0]['followers'] == 43567] for metric in ['followers', 'followings', 'likes']: called_methods.append( (stats, metric, {}, artist) in mock_compare_current_and_last_week_stats. call_args_list[index]) index += 1 for method in called_methods: assert method @mock.patch('tiktok_with_requests.TikTokTracker.update_stats') @mock.patch('tiktok_with_requests.TikTokTracker.get_profiles', artist_last_stats()) def test_get_and_update_stats(self, mock_update_stats, mock_process_tiktok_stats, mock_executor, mock_executor_context): """Test get_and_update_stats method.""" tracker = tiktok.TikTokTracker() tracker.get_and_update_stats() assert mock_process_tiktok_stats.called assert mock_update_stats.called @mock.patch('tiktok_with_requests.TikTokTracker.update_staging_raw_table_with_stats') def test_perform_tracking(self, mock_update_staging_raw_table_with_stats, mock_create_tables, mock_get_and_update_stats, mock_executor, mock_executor_context): """Test perform_tracking method.""" tracker = tiktok.TikTokTracker() tracker.perform_tracking() assert mock_create_tables.called assert mock_get_and_update_stats.called assert mock_update_staging_raw_table_with_stats.called