from google.protobuf.descriptor import FieldDescriptor from google.protobuf.pyext._message import Message from delphi_api.bigtable import BigTableModel from delphi_api.v3.constants import ENABLED_DSPS from delphi_api.v3.data_models.bigtable.row_key_builder import RowKeyBuilder, RowKeyGroupType from delphi_api.v3.enums import RowKeyAgg from delphi_api.v3.view_models.params import Params from tests.conftest import ( MOCK_DATE_RANGE, MOCK_DSP_SLUGS, MOCK_ISRCS, MOCK_PLAYLIST_IDS, TEST_ARTIST_IDS, TEST_CHART_IDS, TEST_ISRCS, TEST_PLAYLIST_IDS, TEST_PRODUCT_IDS, TEST_TRACK_IDS, ) from tests.v2.constants import INT_VAL, STR_VAL class UnitTestUtils: """Static class for unit test helpers (reduce code duplication) """ @staticmethod def get_endpoints(app) -> dict: """Returns a dictionary of the application's registered endpoints. Keys are the url path, values are the rule's (flask) endpoint. """ endpoints = {} with app.app_context(): for rule in app.url_map.iter_rules(): # Filter to only add GET methods if 'GET' in rule.methods: endpoints.update({rule.rule: rule.endpoint}) return endpoints @staticmethod def populate_protobuf_models(message: Message): """This function will populate protobuf model classes with some basic test data, including for use with nested protobuf models (recursive). """ field: FieldDescriptor for field in message.DESCRIPTOR.fields: int_types = ( field.TYPE_FIXED32, field.TYPE_FIXED64, field.TYPE_INT32, field.TYPE_INT64, field.TYPE_UINT32, field.TYPE_UINT64, ) # Set (the same) integer value for tests (nearly all fields are int64) if field.type in int_types: setattr(message, field.name, INT_VAL) # Set (the same) integer value for tests (nearly all fields are int64) elif field.type == field.TYPE_STRING: setattr(message, field.name, STR_VAL) elif field.type == field.TYPE_MESSAGE: child = UnitTestUtils.populate_protobuf_models(getattr(message, field.name)) parent = getattr(message, field.name) # akin to setattr(message, field.name, child) which is not allowed by protobuf parent.CopyFrom(child) return message @staticmethod def create_track_streams(model: BigTableModel, stream_stats): """Creates permutation of rows with mock data in database (emulator)""" for dsp in MOCK_DSP_SLUGS: for isrc in MOCK_ISRCS: for date_str in MOCK_DATE_RANGE: model.isrc = isrc model.date = date_str model.dsp = dsp model.us = stream_stats model.worldwide = stream_stats model.de = bytes() # empty data in protobuf column model.save() @staticmethod def create_playlists(model: BigTableModel, stream_stats): """Creates permutation of rows with mock data in database (emulator)""" for dsp in MOCK_DSP_SLUGS: for playlist_id in MOCK_PLAYLIST_IDS: for date_str in MOCK_DATE_RANGE: model.playlist_id = playlist_id model.date = date_str model.dsp = dsp model.us = stream_stats model.worldwide = stream_stats model.de = bytes() # empty data in protobuf column model.save() @staticmethod def create_playlist_track_streams(model: BigTableModel, stream_stats): """Creates permutation of rows with mock data in database (emulator)""" for dsp in MOCK_DSP_SLUGS: for playlist_id in MOCK_PLAYLIST_IDS: for isrc in MOCK_ISRCS: for date_str in MOCK_DATE_RANGE: model.isrc = isrc model.playlist_id = playlist_id model.date = date_str model.dsp = dsp model.us = stream_stats model.worldwide = stream_stats model.de = bytes() # empty data in protobuf column model.save() @staticmethod def create_v3_track_streams(model: BigTableModel, stream_stats, countries_model): """Creates permutation of rows with mock data in database (emulator)""" current_position = 1 artist_id = TEST_ARTIST_IDS[0] countries = { 'de': stream_stats, 'us': stream_stats, 'worldwide': stream_stats } for dsp in ENABLED_DSPS: for i in range(len(TEST_TRACK_IDS)): isrc = TEST_ISRCS[i] track_id = TEST_TRACK_IDS[i] product_id = TEST_PRODUCT_IDS[i] playlist_id = TEST_PLAYLIST_IDS[i] params_by_isrc = Params(**{ 'artist_id': artist_id, 'dsp': [dsp], 'isrc': [isrc], }) params_by_playlist = Params(**{ # 'artist_id': artist_id, 'dsp': [dsp], 'isrc': [isrc], # 'track_id': [track_id], 'playlist_id': [playlist_id], }) params_variations = [params_by_isrc, params_by_playlist] for date_str in MOCK_DATE_RANGE: current_position += 1 model.artist_id = artist_id model.date = date_str model.dsp = dsp model.isrc = isrc model.playlist_id = playlist_id model.product_id = product_id model.track_id = track_id model.streams = countries_model(countries=countries) model.demographics = countries_model(countries=countries) for params in params_variations: # build multiple row key variations for the data row_keys = RowKeyBuilder.get_row_keys_single( params, RowKeyGroupType.STREAMS, date=date_str) for row_key in row_keys: model.save(row_key) @staticmethod def create_v3_chart_track_positions(model: BigTableModel, dsp: str): current_position = 1 for i in range(len(TEST_ISRCS)): isrc = TEST_ISRCS[i] for chart_id in TEST_CHART_IDS: params_by_chart = Params(**{ 'dsp': [dsp], 'isrc': [isrc], 'chart_id': chart_id, }) params_by_isrc = Params(**{ 'dsp': [dsp], 'isrc': [isrc], }) params_by_dsp = Params(**{ 'dsp': [dsp], }) params_variations = [params_by_chart, params_by_isrc, params_by_dsp] for date_str in MOCK_DATE_RANGE: current_position += 1 model.date = date_str model.dsp = dsp model.isrc = isrc model.chart_id = chart_id model.current = current_position for params in params_variations: # build multiple row key variations for the data row_keys = RowKeyBuilder.get_row_keys_single( params, RowKeyGroupType.CHART_POSITIONS, date=date_str) for row_key in row_keys: model.save(row_key) @staticmethod def create_v3_playlist_track_positions(model: BigTableModel, dsp: str): current_position = 1 for i in range(len(TEST_ISRCS)): isrc = TEST_ISRCS[i] for playlist_id in TEST_CHART_IDS: params_variations = [ Params(**{ 'dsp': [dsp], 'isrc': [isrc], 'playlist_id': playlist_id, }) ] for date_str in MOCK_DATE_RANGE: current_position += 1 model.date = date_str model.dsp = dsp model.isrc = isrc model.playlist_id = playlist_id model.current = current_position for params in params_variations: # build multiple row key variations for the data row_keys = RowKeyBuilder.get_row_keys_single( params, RowKeyGroupType.PLAYLIST_POSITIONS, date=date_str) for row_key in row_keys: model.save(row_key) @staticmethod def create_v3_chartmetrics(model: BigTableModel, dsp: str): base_count = 1 for artist_id in TEST_ARTIST_IDS: params_by_artist = Params(**{ 'artist_id': artist_id, }) params_variations = [params_by_artist] for date_str in MOCK_DATE_RANGE: model.date = date_str model.dsp = dsp model.artist_id = artist_id model.fans = base_count + 1 model.followers = base_count + 10 model.likes = base_count + 20 model.listeners = base_count + 30 model.popularity = base_count + 40 model.talks = base_count + 50 model.subscribers = base_count + 60 model.views = base_count + 70 for params in params_variations: # build multiple row key variations for the data row_keys = RowKeyBuilder.get_row_keys_single( params, RowKeyGroupType.CHARTMETRIC, date=date_str, agg=RowKeyAgg.DAILY) for row_key in row_keys: model.save(row_key)