"""Create persistant changes for the test environment. The goal of those changes is to be able to test mecchanisms without touching specific parts of our architecture. For instance: tests should be running against a database – but this database does not have to be our live database, it should be a 'in memory' db. """ from concurrent import futures import datetime import random from unittest import mock from unittest.mock import MagicMock import fakeredis from oto import response as oto_response from owsfeatures import features from owsrequest import constants as context import pytest from raven.contrib.flask import Client from requests import exceptions as requests_exceptions from analytics import config from analytics.api import app # noqa - required for owsrequest to work in test from analytics.connectors import redis from analytics.connectors import statsd from analytics.models import streams as streams_model config.environment = config.TEST_ENVIRONMENT statsd.send_timer = MagicMock() statsd.send_metric = MagicMock() # Headers with missing account id. HEADERS_ACCOUNT_ID_MISSING = {context.GRASS_ACCOUNT_TYPE: 'vendor'} # Headers with missing account type. HEADERS_ACCOUNT_TYPE_MISSING = {context.GRASS_ACCOUNT_ID: '232323'} @pytest.fixture(params=['vendor', 'subaccount']) def headers(request): """Test fixture for account types.""" def random_id(): return int(1000 + 4000 * random.random()) return { context.GRASS_ACCOUNT_ID: str(random_id()), context.GRASS_ACCOUNT_TYPE: request.param } @pytest.fixture(params=[ HEADERS_ACCOUNT_ID_MISSING, HEADERS_ACCOUNT_TYPE_MISSING ]) def incorrect_headers(request): """Test fixture for incorrect headers.""" return request.param @pytest.fixture(params=[ None, 'USI4R1053228' ]) def isrc(request): """Test fixture for impermanent isrc.""" return request.param @pytest.fixture def playlist_placements_url(): """Test fixture for playlist placements url.""" return '/playlist-placements/' @pytest.fixture def mock_sentry(monkeypatch): """Mock for sentry client.""" sentry_mock = MagicMock(spec=Client) monkeypatch.setattr( 'analytics.connectors.sentry.sentry_client', sentry_mock) return sentry_mock @pytest.fixture def mock_track_custom_parameters(monkeypatch): """Mock for track_custom_parameters.""" track_mock = MagicMock() monkeypatch.setattr( 'analytics.handlers.track_custom_parameters', track_mock) return track_mock @pytest.fixture(autouse=True) def mock_requests(monkeypatch): """Automatically mock requests.sessions.Session. We do not want tests to perform any real HTTP requests """ session_mock = MagicMock() monkeypatch.setattr('requests.sessions.Session', session_mock) monkeypatch.setattr('requests.Session', session_mock) return session_mock @pytest.fixture() def mock_config(monkeypatch): """Automatically mock config.""" class ConfigMock: def __init__(self): self['environment'] = config.TEST_ENVIRONMENT for key in dir(config): if key.startswith('_') or key == 'environment': continue self[key] = MagicMock() def __setitem__(self, key, value): monkeypatch.setattr(config, key, value) return ConfigMock() @pytest.fixture def demographics_db_result(): """Test fixture for demographics db query result.""" return [ ['other', 'unknown', 5672], ['<18', 'women', 89027], ['<18', 'men', 54712], ] @pytest.fixture def streams_params(): """Return dictionary of parameters to be used for get_streams testing.""" return { 'start_date': '2017-08-01', 'end_date': '2017-08-03', 'labelid': 22622, 'subaccountid': 123, 'storeids': [ streams_model.Store.SPOTIFY, streams_model.Store.APPLE_MUSIC], 'isrcs': ['ISRC123', 'ISRC234', 'ISRC345'], 'artistids': [11, 22]} @pytest.fixture def streams_db_result(): """Test fixture for streams db query result.""" return [ (datetime.date(2017, 8, 1), 134, 90, 206, 430), (datetime.date(2017, 8, 3), 121, 82, 229, 432)] @pytest.fixture def streams_db_formatted_result(): """Test formatted fixture for streams db query result.""" return [ { 'date': '2017-08-01', 'streams_from_passive_discovery': 134, 'streams_from_active_discovery': 90, 'streams_from_collection': 206, 'overall_number_of_streams': 430}, { 'date': '2017-08-03', 'streams_from_passive_discovery': 121, 'streams_from_active_discovery': 82, 'streams_from_collection': 229, 'overall_number_of_streams': 432}] @pytest.fixture def streams_formatted_result_with_full_dates(): """Test formatted fixture for streams query result with full dates.""" return [ { 'date': '2017-08-01', 'streams_from_passive_discovery': 134, 'streams_from_active_discovery': 90, 'streams_from_collection': 206, 'overall_number_of_streams': 430}, { 'date': '2017-08-02', 'streams_from_passive_discovery': 0, 'streams_from_active_discovery': 0, 'streams_from_collection': 0, 'overall_number_of_streams': 0}, { 'date': '2017-08-03', 'streams_from_passive_discovery': 121, 'streams_from_active_discovery': 82, 'streams_from_collection': 229, 'overall_number_of_streams': 432}] @pytest.fixture def streams_logic_result(streams_db_formatted_result): """Test fixture for expected get_streams logic.""" return { 'streams_from_passive_discovery_pct': 30, 'streams_from_active_discovery': 172, 'streams_from_passive_discovery': 255, 'overall_number_of_streams': 862, 'streams_from_active_discovery_pct': 20, 'streams_from_collection': 435, 'streams_from_collection_pct': 50, 'items': streams_db_formatted_result} @pytest.fixture def streams_placements_db_result(): """Test fixture for placements db query result.""" return [ (21786, 286, 1613578, 'http://hey', 'Hey', 7891430166878, 'BRWRT1700048', 'Safado', 12345, datetime.date(2017, 6, 2)), (21786, 286, 1341866, 'http://hi', 'Acredita', 7891430169275, 'BXMGD1700109', 'Neto', 12346, datetime.date(2017, 6, 16))] @pytest.fixture def streams_placements_db_formatted_result(): """Test formatted fixture for placements db query result.""" return [ { 'order_number': '1', 'labelid': 21786, 'storeid': 286, 'streams': 1613578, 'playlist_link': 'http://hey', 'track_title': 'Hey', 'upc': 7891430166878, 'isrc': 'BRWRT1700048', 'artist': 'Safado', 'artist_id': 12345, 'date_added': '2017-06-02'}, { 'order_number': '2', 'labelid': 21786, 'storeid': 286, 'streams': 1341866, 'playlist_link': 'http://hi', 'track_title': 'Acredita', 'upc': 7891430169275, 'isrc': 'BXMGD1700109', 'artist': 'Neto', 'artist_id': 12346, 'date_added': '2017-06-16'}] @pytest.fixture def geographics_params(): """Return dictionary to be used for get_geographic_insights testing.""" return { 'start_date': '2017-01-01', 'end_date': '2017-12-01', 'label_id': 123, 'subaccount_id': 456} @pytest.fixture def default_geographics_vendor_params(): """Return default dictionary for get_geographic_insights testing.""" return { 'start_date': '2017-01-01', 'end_date': '2017-12-01', 'isrcs': None, 'artist_ids': None, 'store_ids': None, 'account_type': 'vendor', 'account_id': 1} @pytest.fixture def default_geographics_subaccount_params(): """Return default dictionary for get_geographic_insights testing.""" return { 'start_date': '2017-01-01', 'end_date': '2017-12-01', 'isrcs': None, 'artist_ids': None, 'store_ids': None, 'account_type': 'subaccount', 'account_id': 1} @pytest.fixture def default_geographics_subaccount_one_month_params(): """Return default dictionary for get_geographic_insights with one month.""" return { 'start_date': '2017-01-01', 'end_date': '2017-01-31', 'isrcs': None, 'artist_ids': None, 'store_ids': None, 'account_type': 'subaccount', 'account_id': 1} @pytest.fixture def default_geographics_params(): """Return default dictionary for get_geographic_insights testing.""" return { 'start_date': '2017-01-01', 'end_date': '2017-12-01', 'isrcs': None, 'artist_ids': None, 'store_ids': [1, 286], 'labelid': 1, 'subaccountid': 2} @pytest.fixture def geographics_db_result(): """Test fixture for geographics db query result.""" return [ ('GB', 321), ('FR', 654) ] @pytest.fixture def geographics_db_result_orchard_regions(): """Test fixture for geographics orchard regions db query result.""" return [ ('FR', None, 'Provence-Alpes'), ('GB', None, 'Glasgow'), ('US', None, 'New-York') ] @pytest.fixture def geographics_db_result_orchard_max_date_by_isrc(): """Test fixture for geographics orchard max date by isrc result.""" return (datetime.date(2018, 4, 30),) @pytest.fixture def geographics_db_result_orchard_max_date_by_label(): """Test fixture for geographics orchard max date by label result.""" return (datetime.date(2018, 4, 30),) @pytest.fixture def geographics_db_result_orchard_max_date_by_artist(): """Test fixture for geographics orchard max date by artist result.""" return (datetime.date(2018, 4, 29),) @pytest.fixture def geographics_db_result_by_region(): """Test fixture for geographics by region db query result.""" return [ ('Provence-Alpes', 'FR-U', 'FR', 654), ('Glasgow', 'GB-GLG', 'GB', 321) ] @pytest.fixture def geographics_logic_expected_result(): """Test formatted fixture for geographic insights db query result.""" return { 'total_listeners': 0, 'total_streams': 9, 'items': [ { 'territory_code': 'FR', 'number_of_streams': 2, 'unique_listeners': 0 }, { 'territory_code': 'GB', 'number_of_streams': 3, 'unique_listeners': 0 }, { 'territory_code': 'US', 'number_of_streams': 4, 'unique_listeners': 0 } ] } @pytest.fixture def geographics_logic_expected_result_with_growth(): """Test formatted fixture for geographic insights db query result.""" return { 'total_streams_growth': 50, 'total_streams': 9, 'total_listeners_growth': 0, 'total_listeners': 0, 'items': [ { 'number_of_streams': 2, 'unique_listeners': 0, 'listeners_growth': 0.0, 'streams_growth': 100.0, 'territory_code': 'FR' }, { 'number_of_streams': 3, 'unique_listeners': 0, 'listeners_growth': 0, 'streams_growth': 50.0, 'territory_code': 'GB' }, { 'number_of_streams': 4, 'unique_listeners': 0, 'listeners_growth': 0, 'streams_growth': 33.33, 'territory_code': 'US' } ] } @pytest.fixture def geographics_db_result_formatted_result(): """Test formatted fixture for geographic insights db query result.""" return [ { 'territory_code': 'FR', 'unique_listeners': 0, 'number_of_streams': 2 }, { 'territory_code': 'GB', 'unique_listeners': 0, 'number_of_streams': 3 }, { 'territory_code': 'US', 'unique_listeners': 0, 'number_of_streams': 4 } ] @pytest.fixture def previous_geographics_db_result_formatted_result(): """Test formatted fixture for previous geographic insights db result.""" return [ { 'territory_code': 'FR', 'unique_listeners': 0, 'number_of_streams': 1 }, { 'territory_code': 'GB', 'unique_listeners': 0, 'number_of_streams': 2 }, { 'territory_code': 'US', 'unique_listeners': 0, 'number_of_streams': 3 } ] @pytest.fixture def geographics_by_region_logic_expected_result(): """Test formatted fixture for geographic insights db query result.""" return { 'total_streams': 6, 'total_listeners': 0, 'items': [ { 'number_of_streams': 1, 'region_code': 'FR-U', 'territory_code': 'FR', 'unique_listeners': 0, 'region_name': 'Provence-Alpes' }, { 'number_of_streams': 3, 'region_code': '501', 'territory_code': 'US', 'unique_listeners': 0, 'region_name': 'New-York' }, { 'number_of_streams': 2, 'region_code': 'GB-GLG', 'territory_code': 'GB', 'unique_listeners': 0, 'region_name': 'Glasgow' } ] } @pytest.fixture def geographics_by_region_db_result_formatted_result(): """Test formatted fixture for geographic insights db query result.""" return [ { 'region_name': 'Provence-Alpes', 'region_code': 'FR-U', 'territory_code': 'FR', 'unique_listeners': 0, 'number_of_streams': 1 }, { 'region_name': 'New-York', 'region_code': '501', 'territory_code': 'US', 'unique_listeners': 0, 'number_of_streams': 3 }, { 'region_name': 'Glasgow', 'region_code': 'GB-GLG', 'territory_code': 'GB', 'unique_listeners': 0, 'number_of_streams': 2 } ] @pytest.fixture def previous_geographics_by_region_db_result_formatted_result(): """Test previous geographic insights by region result.""" return [ { 'region_name': 'Provence-Alpes', 'region_code': 'FR-U', 'territory_code': 'FR', 'unique_listeners': 0, 'number_of_streams': 1 }, { 'region_name': 'New-York', 'region_code': '501', 'territory_code': 'US', 'unique_listeners': 0, 'number_of_streams': 1 }, { 'region_name': 'Glasgow', 'region_code': 'GB-GLG', 'territory_code': 'GB', 'unique_listeners': 0, 'number_of_streams': 2 } ] @pytest.fixture def geographics_orchard_regions_db_formatted_result(): """Test formatted fixture for geographic orchard regions.""" return [ { 'region_name': 'Provence-Alpes', 'region_code': None, 'territory_code': 'FR' }, { 'region_name': 'Glasgow', 'region_code': None, 'territory_code': 'GB' }, { 'region_name': 'New-York', 'region_code': None, 'territory_code': 'US' } ] @pytest.fixture def geographics_orchard_dates_db_result_formatted( geographics_db_result_orchard_max_date_by_label): """Test formatted fixture for geographic orchard dates.""" return (str(geographics_db_result_orchard_max_date_by_label[0])) @pytest.fixture def get_orchard_dates_result(): """Test formatted fixture for geographic orchard date.""" return {'last_activity_date': '2018-04-01'} @pytest.fixture def get_orchard_dates_response(geographics_orchard_dates_db_result_formatted): """Return a response with result of get_geographic_orchard_date.""" return oto_response.Response( message=geographics_orchard_dates_db_result_formatted) @pytest.fixture def geographics_by_region_expected_logic_result_with_growth(): """Test geographic insights by region result with growth.""" return { 'total_streams_growth': 50, 'total_listeners_growth': 0, 'total_listeners': 0, 'total_streams': 6, 'items': [ { 'region_code': 'FR-U', 'unique_listeners': 0, 'territory_code': 'FR', 'listeners_growth': 0.0, 'streams_growth': 0.0, 'region_name': 'Provence-Alpes', 'number_of_streams': 1 }, { 'region_code': '501', 'unique_listeners': 0, 'territory_code': 'US', 'listeners_growth': 0.0, 'streams_growth': 200.0, 'region_name': 'New-York', 'number_of_streams': 3 }, { 'region_code': 'GB-GLG', 'unique_listeners': 0, 'territory_code': 'GB', 'listeners_growth': 0.0, 'streams_growth': 0.0, 'region_name': 'Glasgow', 'number_of_streams': 2 } ] } @pytest.fixture def mock_success_get_geographic_insights( monkeypatch, geographics_db_result_formatted_result, previous_geographics_db_result_formatted_result): """Mock success for get_geographic_insights.""" mock_get_geographic_insights = mock.Mock() mock_get_geographic_insights.side_effect = [ oto_response.Response(message=geographics_db_result_formatted_result), oto_response.Response( message=previous_geographics_db_result_formatted_result)] monkeypatch.setattr( 'analytics.models.geographics.get_geographic_insights', mock_get_geographic_insights) return mock_get_geographic_insights @pytest.fixture def mock_no_content_get_geographic_insights(monkeypatch): """Mock no content for get_geographic_insights.""" mock_get_geographic_insights = mock.Mock() mock_get_geographic_insights.return_value = oto_response.Response( status=204) monkeypatch.setattr( 'analytics.models.geographics.get_geographic_insights', mock_get_geographic_insights) return mock_get_geographic_insights @pytest.fixture def mock_success_get_geographic_insights_by_region( monkeypatch, geographics_by_region_db_result_formatted_result, previous_geographics_by_region_db_result_formatted_result): """Mock success for get_geographic_insights_by_region.""" mock_get_geographic_insights_by_region = mock.Mock() mock_get_geographic_insights_by_region.side_effect = [ oto_response.Response( message=geographics_by_region_db_result_formatted_result), oto_response.Response( message=previous_geographics_by_region_db_result_formatted_result)] monkeypatch.setattr( 'analytics.models.geographics.get_geographic_insights_by_region', mock_get_geographic_insights_by_region) return mock_get_geographic_insights_by_region @pytest.fixture def mock_no_content_get_geographic_insights_by_region(monkeypatch): """Mock no content for get_geographic_insights_by_region.""" mock_get_geographic_insights = mock.Mock() mock_get_geographic_insights.return_value = oto_response.Response( status=204) monkeypatch.setattr( 'analytics.models.geographics.get_geographic_insights_by_region', mock_get_geographic_insights) return mock_get_geographic_insights @pytest.fixture def mock_success_get_vendor_id_by_subaccount_id( monkeypatch, geographics_params): """Mock success for get_vendor_id_by_subaccount_id.""" mock_ows_account = mock.Mock() mock_ows_account.return_value = mock.Mock( message=geographics_params['label_id']) monkeypatch.setattr( 'analytics.models.ows_account.get_vendor_id_by_subaccount_id', mock_ows_account) return mock_ows_account @pytest.fixture def mock_error_get_vendor_id_by_subaccount_id( monkeypatch, geographics_params): """Mock error for get_vendor_id_by_subaccount_id.""" mock_ows_account = MagicMock( return_value=oto_response.create_error_response( status=500, code='error code', message='error message')) monkeypatch.setattr( 'analytics.models.ows_account.get_vendor_id_by_subaccount_id', mock_ows_account) return mock_ows_account @pytest.fixture def mock_success_get_geographic_orchard_regions( monkeypatch, geographics_orchard_regions_db_formatted_result): """Mock success for get_geographic_insights_by_region.""" mock_get_geographic_orchard_regions = mock.Mock() mock_get_geographic_orchard_regions.return_value = oto_response.Response( message=geographics_orchard_regions_db_formatted_result) monkeypatch.setattr( 'analytics.models.geographics.get_geographic_orchard_regions', mock_get_geographic_orchard_regions) return mock_get_geographic_orchard_regions @pytest.fixture def mock_no_content_get_geographic_orchard_regions( monkeypatch, geographics_orchard_regions_db_formatted_result): """Mock success for get_geographic_insights_by_region.""" mock_get_geographic_orchard_regions = mock.Mock() mock_get_geographic_orchard_regions.return_value = oto_response.Response( status=204) monkeypatch.setattr( 'analytics.models.geographics.get_geographic_orchard_regions', mock_get_geographic_orchard_regions) return mock_get_geographic_orchard_regions @pytest.fixture def mock_success_get_geographics_latest_date( monkeypatch, geographics_orchard_dates_db_result_formatted): """Mock success for get_geographics_latest_date.""" mock_get_geographics_latest_date = mock.Mock() mock_get_geographics_latest_date.return_value = mock.Mock( message=geographics_orchard_dates_db_result_formatted) monkeypatch.setattr( 'analytics.models.geographics.get_geographics_latest_date', mock_get_geographics_latest_date) return mock_get_geographics_latest_date @pytest.fixture def mock_error_get_geographics_latest_date(monkeypatch): """Mock error for get_geographics_latest_date.""" mock_get_geographics_latest_date = MagicMock( return_value=oto_response.create_error_response( status=500, code='error code', message='error message')) monkeypatch.setattr( 'analytics.models.geographics.get_geographics_latest_date', mock_get_geographics_latest_date) return mock_get_geographics_latest_date @pytest.fixture def placements_logic_result(streams_placements_db_formatted_result): """Test fixture for expected get_placements logic.""" return { 'total': '2', 'items': streams_placements_db_formatted_result} @pytest.fixture def get_streams_model_response(streams_db_formatted_result): """Return a response with result of get_streams to data store.""" return oto_response.Response(message=streams_db_formatted_result) @pytest.fixture def get_streams_placements_response(streams_placements_db_formatted_result): """Return a response with result of get_placements to data store.""" return oto_response.Response( message=streams_placements_db_formatted_result) @pytest.fixture def get_streams_logic_result_response(streams_logic_result): """Return a response with result of get_streams to data store.""" return oto_response.Response(message=streams_logic_result) @pytest.fixture def get_playlist_placements_logic_result_response(placements_logic_result): """Return a response with result of get_placements to data store.""" return oto_response.Response(message=placements_logic_result) @pytest.fixture(autouse=True) def mock_feature_engine(monkeypatch): """Mock feature_engine of owsfeatures.""" feature_engine_mock = MagicMock(name='feature_engine') feature_engine_mock.enabled_features = [] monkeypatch.setattr( features.feature_engine, 'get', feature_engine_mock.get) monkeypatch.setattr( features.feature_engine, 'load', feature_engine_mock.load) def get_features(): return { feature_name: 'enabled' for feature_name in feature_engine_mock.enabled_features } feature_engine_mock.get.side_effect = get_features return feature_engine_mock @pytest.fixture(autouse=True) def mock_redis(monkeypatch): """ Mock redis client. Creates a new fake redis for each test. """ fake_client = fakeredis.FakeStrictRedis() monkeypatch.setattr(redis, 'client', fake_client) return fake_client @pytest.fixture def placement_total_params(): """Mock parameters for get_placements_totals testing.""" return { 'start_date': '2017-08-01', 'end_date': '2017-08-03', 'labelid': 22622, 'subaccountid': 123, 'storeids': [ streams_model.Store.SPOTIFY, streams_model.Store.APPLE_MUSIC], 'isrcs': ['ISRC123', 'ISRC234', 'ISRC345'], 'artistids': [11, 22]} @pytest.fixture def placement_params(placement_total_params): """Mock parameters for get_placements testing.""" placement_params_with_limit = dict(placement_total_params) placement_params_with_limit['limit'] = 40 return placement_params_with_limit @pytest.fixture def placement_total_params_vendor(placement_total_params): """Mock parameters on vendor-level for get_placements_totals testing.""" placement_total_params['subaccountid'] = None return placement_total_params @pytest.fixture def placement_params_vendor(placement_params): """Mock parameters on vendor-level for get_placements testing.""" placement_params['subaccountid'] = None return placement_params @pytest.fixture def thread_pool_executor_mock(monkeypatch): """Mock futures.ThreadPoolExecutor.""" thread_pool_executor = MagicMock() thread_pool_executor_object = thread_pool_executor.return_value executor = mock.Mock() sumbit_result = executor.submit.return_value thread_pool_executor_object.__enter__.return_value = executor monkeypatch.setattr( futures, 'ThreadPoolExecutor', thread_pool_executor) executor.sumbit_result = sumbit_result return executor @pytest.fixture( params=[ requests_exceptions.Timeout, requests_exceptions.ConnectTimeout, requests_exceptions.ReadTimeout] ) def timeout_exception_class(request): """Timeout exceptions for requests package.""" return request.param