import os import unittest from sys import stderr import pytest from flask import Response, url_for from google.cloud.bigtable.column_family import MaxVersionsGCRule from moto import mock_secretsmanager from structlog import getLogger from delphi_api.seeds.data_seeder import TEST_DATASET, run_seeder from delphi_api.v3.constants import DSP_APPLE, DSP_SPOTIFY, ENABLED_DSPS from delphi_api.v3.data_models.bigtable.chart_position import ChartPosition from delphi_api.v3.data_models.bigtable.dsp_streams import ( AmazonTrackStream, AppleTrackStream, SpotifyTrackStream, YouTubeTrackStream, ) from delphi_api.v3.data_models.bigtable.playlist_position import PlaylistPosition from delphi_api.v3.enums import ColumnFamily from delphi_api.v3.proto.gen.amazon_v3_pb2 import AmazonCountryStats, AmazonStreamStats from delphi_api.v3.proto.gen.apple_v3_pb2 import AppleCountryStats, AppleStreamStats from delphi_api.v3.proto.gen.spotify_v3_pb2 import SpotifyCountryStats, SpotifyStreamStats from delphi_api.v3.proto.gen.youtube_v3_pb2 import YouTubeTrackStreamStats from delphi_api.v3.view_models.artist import ArtistsViewModel from delphi_api.v3.view_models.stream import StreamsViewModel from tests.mocks import ( patch_decode_token, patch_get_image_data, patch_get_secret, patch_log_request, ) from tests.utils import UnitTestUtils from tests.v3.constants import ( ARTISTS_FOLLOWERS_ROUTES, ARTISTS_ROUTES, CHARTS_ROUTES, DSPS_ROUTES, HEALTH_ROUTES, PLAYLISTS_ROUTES, POPULARITY_ROUTES, PRODUCTS_ROUTES, REGIONS_ROUTES, STATUS_ROUTES, STREAMS_ROUTES, TRACKS_ROUTES, TRACK_POSITIONS_CHARTS_ROUTES, TRACK_POSITIONS_PLAYLISTS_ROUTES, ) LOG = getLogger(__name__) _SERVER_NAME = 'localhost:5000' _stream_stats = 'stream_stats' _countries = 'countries' # Models mapping to create tables and fixture data for streams _MODELS_PROTO_MAP = { # Amazon AmazonTrackStream: { _stream_stats: AmazonStreamStats, _countries: AmazonCountryStats }, # Apple AppleTrackStream: { _stream_stats: AppleStreamStats, _countries: AppleCountryStats }, # Spotify SpotifyTrackStream: { _stream_stats: SpotifyStreamStats, _countries: SpotifyCountryStats, }, # YouTube YouTubeTrackStream: { _stream_stats: YouTubeTrackStreamStats, _countries: YouTubeTrackStreamStats, }, } CHART_POS_MODELS = { DSP_SPOTIFY: ChartPosition, DSP_APPLE: ChartPosition, } PLAYLIST_POS_MODELS = { DSP_SPOTIFY: PlaylistPosition, DSP_APPLE: PlaylistPosition, } _ENABLED_STREAMS_MODELS = StreamsViewModel.ENABLED_DATA_MODELS _ENABLED_PLAYLIST_POS_MODELS = {k: v for k, v in PLAYLIST_POS_MODELS.items() if k in ENABLED_DSPS} _ENABLED_CHART_POS_MODELS = {k: v for k, v in CHART_POS_MODELS.items() if k in ENABLED_DSPS} _ENABLED_CHARTMETRIC_MODELS = ArtistsViewModel.ENABLED_CHARTMETRIC_MODELS # Note only routes that are registered within the app (via swagger) will be called # Any routes included here but not registered will simply be ignored _ROUTES = (HEALTH_ROUTES + ARTISTS_ROUTES + ARTISTS_FOLLOWERS_ROUTES + CHARTS_ROUTES + DSPS_ROUTES + PLAYLISTS_ROUTES + POPULARITY_ROUTES + PRODUCTS_ROUTES + REGIONS_ROUTES + STATUS_ROUTES + STREAMS_ROUTES + TRACKS_ROUTES + TRACK_POSITIONS_PLAYLISTS_ROUTES + TRACK_POSITIONS_CHARTS_ROUTES ) # _ROUTES = STREAMS_ROUTES @pytest.mark.usefixtures('_db') @mock_secretsmanager class TestEndpoints(unittest.TestCase): def setUp(self): run_seeder(TEST_DATASET, truncate=True) self.mock_decode = patch_decode_token() self.mock_decode.start() self.patch_get_secret = patch_get_secret() self.patch_get_secret.start() self.patch_log_request = patch_log_request() self.patch_log_request.start() self.patch_get_image_data = patch_get_image_data() self.patch_get_image_data.start() from delphi_api.core.app import get_app self.app = get_app().app self.app.testing = True self.app.config['SERVER_NAME'] = _SERVER_NAME self.client = self.app.test_client() self.endpoints = UnitTestUtils.get_endpoints(self.app) emulator_host = os.getenv('BIGTABLE_EMULATOR_HOST') if not emulator_host or os.getenv('ENVIRONMENT') != 'test': raise EnvironmentError('BigTable emulator environment variable not set. ' 'Ensure the emulator is running and the env variable is set.') print('BIGTABLE_EMULATOR_HOST = %s' % emulator_host, file=stderr) # Create fixture tables TestEndpoints.create_test_data() @staticmethod def create_test_data(): for model_cls, cls_map in _MODELS_PROTO_MAP.items(): # Create dummy streams (and playlist pos) data for each model table in bigtable emulator if model_cls not in _ENABLED_STREAMS_MODELS.values(): continue model = model_cls() model.create_table({ ColumnFamily.META.value: MaxVersionsGCRule(1), ColumnFamily.METRICS.value: MaxVersionsGCRule(1), ColumnFamily.POS.value: MaxVersionsGCRule(1), }) # populate nested proto models stats_instance = cls_map[_stream_stats]() stream_stats = UnitTestUtils.populate_protobuf_models(stats_instance) countries_model = cls_map[_countries] UnitTestUtils.create_v3_track_streams(model, stream_stats, countries_model) for dsp, model_cls in _ENABLED_CHART_POS_MODELS.items(): # create dummy chart track positions data for each model table in bigtable emulator model = model_cls() model.create_table({ ColumnFamily.META.value: MaxVersionsGCRule(1), ColumnFamily.METRICS.value: MaxVersionsGCRule(1), ColumnFamily.POS.value: MaxVersionsGCRule(1), }) UnitTestUtils.create_v3_chart_track_positions(model, dsp) for dsp, model_cls in _ENABLED_PLAYLIST_POS_MODELS.items(): # create dummy chart track positions data for each model table in bigtable emulator model = model_cls() model.create_table({ ColumnFamily.META.value: MaxVersionsGCRule(1), ColumnFamily.METRICS.value: MaxVersionsGCRule(1), ColumnFamily.POS.value: MaxVersionsGCRule(1), }) UnitTestUtils.create_v3_playlist_track_positions(model, dsp) for dsp, model_cls in _ENABLED_CHARTMETRIC_MODELS.items(): # create dummy chartmetrics data for each model table in bigtable emulator model = model_cls() model.create_table({ ColumnFamily.META.value: MaxVersionsGCRule(1), ColumnFamily.METRICS.value: MaxVersionsGCRule(1), }) UnitTestUtils.create_v3_chartmetrics(model, dsp) def test_endpoints(self): with self.app.app_context(): headers = {'Authorization': 'Bearer 100'} for item in _ROUTES: route = item['route'] params = item['params'] msg = f'{route}, params: {params}' with self.subTest(msg=msg): endpoint = self.endpoints.get(route) if not endpoint: print('Skipping testing for route: %s' % route, file=stderr) continue url = url_for(endpoint, **params) response: Response = self.client.get(url, headers=headers) data = response.json if data.get('items'): self.assertTrue(len(data.get('items', [])), msg=f'No data in items; \n\t URL: {url}') else: self.assertTrue(bool(data), msg=f'Empty data found; \n\t URL: {url}') self.assertEqual(response.status_code, 200, msg=f'Status code != 200; \n' f'DETAILS: ' f'\n\t route={route} ' f'\n\t params={params} ' f'\n\t url={url} ' f'\n\t status_code={response.status_code}' f'\n\t body={data}') def tearDown(self): for model in _MODELS_PROTO_MAP: model().delete_table() self.mock_decode.stop() self.patch_get_secret.stop() self.patch_log_request.stop() self.patch_get_image_data.stop()