"""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 TikTokStatsSFExecutor def get_args(f, params=False): if params: return f.call_args[1]['params'] return f.call_args[0] def check_query(query: str, contains: list, check=True): 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.date.today(), datetime.date.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': {'tiktok_link': None, 'current_followers': 2345, 'weekly_change_in_followers': 32, 'daily_change_in_followers': 4, 'current_weekly_change_in_followers': 56, 'last_week_followers': 2296, 'percentage_change_in_followers': 0, 'current_followings': 0, 'weekly_change_in_followings': 0, 'daily_change_in_followings': 0, 'current_weekly_change_in_followings': 0, 'last_week_followings': 0, 'percentage_change_in_followings': 0, 'current_likes': 43642, 'weekly_change_in_likes': 244, 'daily_change_in_likes': 105, 'current_weekly_change_in_likes': 554, 'last_week_likes': 43002, 'percentage_change_in_likes': 1.2, 'last_processing_date': datetime.date.today() - datetime.timedelta(days=1), 'last_week_processing_date': last_week_processing_date}} def get_tiktok_stats(mock_argument): """Fixture returning the list of artists' links.""" return { 'followers': 43567, 'followings': 0, 'likes': 345678, } @pytest.fixture def processed_links() -> dict: """Return processed dataset example.""" links = {'artist': 'cijikin', 'instagram': 'To check the correctness of the output.', 'spotify': 'https://open.spotify.com/artist/060CEbkq3Bubu0AvJu4NKf', 'youtube': 'https://www.youtube.com/channel/UC5kuP-o0jopVpHCD8KSrPpg', 'soundcloud': 'https://soundcloud.com/cijikin', 'shazam': None, 'tiktok': 'https://tiktok.com/@cijikin', '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 = TikTokStatsSFExecutor(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.TikTokStatsSFExecutor' 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://tiktok.com/@cijikin']] @pytest.fixture def mock_select_artists_links(): """Yield SELECT artists' links SQL query result.""" select_path = ( 'tiktok_with_requests.TikTokStatsSFExecutor.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 = ( 'tiktok_with_requests.TikTokStatsSFExecutor.create_artist_link_table') with patch(create_path) as create: yield create @pytest.fixture def mock_create_tiktok_statistics(): """Yield CREATE tiktok_statistics table SQL query result.""" create_path = ( 'tiktok_with_requests.TikTokStatsSFExecutor.create_tiktok_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_create_staging_raw_tiktok_statistics(): """Yield CREATE staging_raw_tiktok_statistics table SQL query result.""" create_path = ( 'tiktok_with_requests.TikTokStatsSFExecutor.create_staging_raw_tiktok_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_delete_from_staging_raw(): """Yield DELETE from staging_raw_tiktok_statistics table SQL query result.""" delete_path = ( 'tiktok_with_requests.TikTokStatsSFExecutor.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 = ( 'tiktok_with_requests.TikTokStatsSFExecutor.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 = ( 'tiktok_with_requests.TikTokTracker.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 = ( 'tiktok_with_requests.TikTokTracker.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 = ( 'tiktok_with_requests.TikTokTracker.get_and_update_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_process_tiktok_stats(): """Yield process_tiktok_stats method from tracker class.""" get_path = ( 'tiktok_with_requests.TikTokTracker.process_tiktok_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_abbreviated_value_to_int(): """Yield abbreviated_value_to_int method from helpers.""" path = ( 'tiktok_with_requests._abbreviated_value_to_int') with patch(path) as f: yield f