"""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 YouTubeStatsSFExecutor def get_args(f, params=False): """Get args dict.""" 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, 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': dict( current_subscribers=0, last_week_subscribers=0, weekly_change_in_subscribers=0, daily_change_in_subscribers=0, current_weekly_change_in_subscribers=0, percentage_change_in_subscribers=0, acceleration_subscribers=0, current_videos=0, current_likes=0, last_week_likes=0, weekly_change_in_likes=0, daily_change_in_likes=0, current_weekly_change_in_likes=0, current_dislikes=0, percentage_change_in_likes=0, acceleration_likes=0, last_week_dislikes=0, weekly_change_in_dislikes=0, current_views=0, last_week_views=0, weekly_change_in_views=0, percentage_change_in_dislikes=0, daily_change_in_views=0, current_weekly_change_in_views=0, percentage_change_in_views=0, acceleration_views=0, current_comments=0, last_week_comments=0, weekly_change_in_comments=0, percentage_change_in_comments=0, likes_engagement=0, comments_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://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 = YouTubeStatsSFExecutor(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.YouTubeStatsSFExecutor' 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://youtube.com/channel/UC5kuP-o0jopVpHCD8KSrPpg']] @pytest.fixture def mock_select_artists_links(): """Yield SELECT artists' links SQL query result.""" select_path = ( 'youtube_with_pyyoutube.YouTubeStatsSFExecutor.select_artists_links') with patch(select_path) as select: select.return_value = get_artists_links yield select @pytest.fixture def mock_create_youtube_statistics(): """Yield CREATE youtube_statistics table SQL query result.""" create_path = ( 'youtube_with_pyyoutube.YouTubeStatsSFExecutor.' 'create_youtube_statistics') with patch(create_path) as create: yield create @pytest.fixture def mock_create_staging_raw_youtube_statistics(): """Yield CREATE staging_raw_youtube_statistics table SQL query result.""" create_path = ( 'youtube_with_pyyoutube.YouTubeStatsSFExecutor.' 'create_staging_raw_youtube_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 = ( 'youtube_with_pyyoutube.YouTubeStatsSFExecutor.' '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 = ( 'youtube_with_pyyoutube.YouTubeStatsSFExecutor.select_weekly_change') with patch(select_path) as select: select.return_value = {'subscribers': 45, 'likes': 0, 'dislikes': 6, 'views': 4, 'comments': 66} yield select @pytest.fixture def mock_compare_current_and_last_week_stats(): """Yield mock_compare_current_and_last_week_stats method.""" calc_path = ( 'youtube_with_pyyoutube.YouTubeTracker.' '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 = ( 'youtube_with_pyyoutube.YouTubeTracker.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 = ( 'youtube_with_pyyoutube.YouTubeTracker.get_and_update_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_profiles(): """Yield get_all_stats_and_update method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_profiles') with patch(get_path) as get: get.return_value = ['cijikin', 'tsigan04ka', 'iguriku', 'otvaga'] yield get @pytest.fixture def mock_get_channel_stats(): """Yield get_channel_stats method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_channel_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_channel_info(): """Yield get_channel_info method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_channel_info') with patch(get_path) as get: yield get @pytest.fixture def mock_get_channel(): """Yield get_channel method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_channel') with patch(get_path) as get: get.return_value = {'subscribers': 400, 'total_views': 76879} yield get @pytest.fixture def mock_get_video_ids(mock_argument=0): """Yield get_video_ids method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_video_ids') with patch(get_path) as get: get.return_value = [] yield get @pytest.fixture def mock_get_likes_dislikes_stats(): """Yield get_likes_dislikes_stats method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_likes_dislikes_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_get_views_comments_stats(): """Yield get_views_comments_stats method from tracker class.""" get_path = ( 'youtube_with_pyyoutube.YouTubeTracker.get_views_comments_stats') with patch(get_path) as get: yield get @pytest.fixture def mock_update_profiles(): """Yield update_profiles method from tracker class.""" update_path = ( 'youtube_with_pyyoutube.YouTubeTracker.update_profiles') with patch(update_path) as update: yield update