"""conftest. This file gets picked up when running py.test tests: http://pytest.org/latest/writing_plugins.html#conftest """ from collections import namedtuple from copy import deepcopy import json import random import string from unittest.mock import MagicMock, patch # Patch before importing application to prevent a live Auth0 JWKS network call # during unit tests. flask_request.setup() unconditionally fetches JWKs at # import time regardless of the verify_access=False setting. _jwks_patcher = patch('owsrequest.auth.get_auth0_jwks', return_value={}) _jwks_patcher.start() import application from oto import response from neo4j.graph import Graph from neo4j.graph import Node from owslogger import flask_logger from owsresponse.response import Response from backend.features import pythonfeatures from backend import features import pytest from backend.constants import api as api_consts from backend.constants import error from backend.constants import field as field_consts from backend.constants import header as header_consts from backend.constants import track_field from backend.constants import us_publishing_obligation as pub_obl from backend.constants import validation as validation_const from backend.models.track_query import TrackQuery from tests.testutils import constants as test_consts from tests.testutils import factories from tests.testutils import functions as test_funcs from tests.testutils.fixtures import localization_fixture from tests.testutils.fixtures import product_all_tracks_response from tests.testutils.fixtures import product_all_valid_tracks_response from tests.testutils.fixtures import product_apply_to_all_fixture from tests.testutils.fixtures import track_bulk_create_fixture from tests.testutils.fixtures import track_create_fixture from tests.testutils.fixtures import track_delete_fixture from tests.testutils.fixtures import track_reorder_fixture from tests.testutils.fixtures import track_sample_fixture from tests.testutils.fixtures import track_spatial_fixture from tests.testutils.fixtures import track_update_fixture from tests.testutils.fixtures import track_validate_response from tests.testutils.functions import wrap_api_results from tests.testutils.mocks.log_adapter import MockLogAdaptor from tests.testutils.seed import track_seed from tests.testutils.seed.languages_seed import languages_seed_data from tests.testutils.seed.track_master_rights_seed import \ tuid_to_owner_type_mapping from tests.testutils.seed.track_producer_natl_seed import \ track_to_producer_natl_mapping # Set feature flag default values across all unit tests FEATURE_FLAG_SETTINGS = {} TestDataSet = namedtuple( 'TestDataSet', [ 'product_id', 'logic_headers', 'request_data', 'ows_product_track_localizations', 'persister_response', 'logic_response']) @pytest.fixture def fixture_app(): """Fixture app context.""" with application.app.app_context() as context: context.g.ows = MagicMock() context.g.request_context = MagicMock() yield context @pytest.fixture(scope='session', autouse=True) def set_global_feature_flags(request, session_mocker): """Global fixture for patching feature flags client. Use the "feature_flags" fixture to set specific values. """ def get_single_feature(flag_name, *args, **kwargs): return Response( status=200, message=("enabled" if FEATURE_FLAG_SETTINGS.get(flag_name) else "control") ) session_mocker.patch.object(pythonfeatures, "get_single_feature", side_effect=get_single_feature) session_mocker.patch.object(features, "is_validate_missing_producer_enabled", return_value=False) @pytest.fixture def feature_flags(): """Fixture to set up specific feature flags. Usage: def test_my_feature(feature_flags): feature_flags["my_flag"] = True # to make it "enabled" feature_flags["my_another_flag] = False # to make it "control" """ global FEATURE_FLAG_SETTINGS yield FEATURE_FLAG_SETTINGS FEATURE_FLAG_SETTINGS = {} @pytest.fixture def client(): """Return flask test client. Returns: flask client: flask test client """ return application.app.test_client() @pytest.fixture def valid_headers(): """Generally-required headers for calls to this service. Returns: dict: header values. """ return { header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL } @pytest.fixture def client_headers(): """Return test headers when using test client for functional tests. Returns: dict: header values. """ return { header_consts.GRASS_ACCOUNT_TYPE: header_consts.GRASS_ACCOUNT_TYPE_VENDOR, header_consts.GRASS_ACCOUNT_ID: test_consts.TEST_VENDOR_ID, header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL, header_consts.ORCHARD_USER_ID: 'alw:123' } @pytest.fixture def profile_headers(): """Return test profile headers when using test client for functional tests. Returns: dict: header values. """ return { header_consts.ORCHARD_PROFILE_TYPE: 'ContentProfile', header_consts.ORCHARD_PROFILE_ID: '123', header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL, } @pytest.fixture def oa_client_headers(): """Return test headers when using test client for functional tests. Returns: dict: header values. """ return { header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL, header_consts.ORCHARD_USER_ID: 'oa:123' } @pytest.fixture def content_profile_client_headers(): """Return test headers when using test client for functional tests. Returns: dict: header values. """ return { header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: 'application/json', header_consts.ORCHARD_PROFILE_TYPE: 'ContentProfile', header_consts.ORCHARD_PROFILE_ID: '789', } @pytest.fixture def microservice_headers(): """Return test headers when using test client for functional tests. This is for microservice to microservice calls and do not include Grass Headers Returns: dict: header values. """ return { header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL } @pytest.fixture def account_query_params(): """Account data query params. This is for microservice to microservice calls and do not include Grass Headers Returns: dict: header values. """ return { field_consts.ACCOUNT_TYPE: header_consts.GRASS_ACCOUNT_TYPE_VENDOR, field_consts.ACCOUNT_ID: test_consts.TEST_VENDOR_ID } missing_grass_account_type_header = { header_consts.GRASS_ACCOUNT_ID: test_consts.TEST_VENDOR_ID, header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL } missing_grass_account_id_header = { header_consts.GRASS_ACCOUNT_TYPE: header_consts.GRASS_ACCOUNT_TYPE_VENDOR, header_consts.CORRELATION_ID: 'abc-123', header_consts.CONTENT_TYPE: header_consts.CONTENT_TYPE_JSON_VAL } @pytest.fixture(params=[ missing_grass_account_type_header, missing_grass_account_id_header]) def invalid_grass_headers(request): """Return invalid grass headers. Args: request (_pytest.python.SubRequest): a sub request for handling getting a fixture from a test function/fixture. Returns: dict: header values. """ return request.param @pytest.fixture def logic_headers(): """Return headers required to call logic layer functions. Returns: dict: Headers that need to be passed """ return { 'account_type': 'vendor', 'account_id': test_consts.TEST_VENDOR_ID } @pytest.fixture(params=[ {'user_id': None, 'user_type': None}, {'user_id': 'alw', 'user_type': '475'} ]) def logic_user(request): """Return user id and type for a user. Returns: dict: Properties that identify a user. """ return request.param @pytest.fixture def track_factory(): """Return a track factory.""" factories.TrackFactory.reset_sequence() factories.TrackArtistFactory.reset_sequence() return factories.TrackFactory @pytest.fixture def track_localization_factory(): """Return a track factory.""" return factories.TrackLocalizationFactory @pytest.fixture def performer_factory(): """Return a performer factory.""" factories.PerformerFactory.reset_sequence() return factories.PerformerFactory @pytest.fixture def test_vendor_id(): """Return test vendor_id.""" return test_consts.TEST_VENDOR_ID @pytest.fixture def test_upc_code(): """Return test UPC.""" return test_consts.TEST_NEW_UPC @pytest.fixture def classical_genre_id(): """Return test UPC.""" return test_consts.TEST_CLASSICAL_GENRE_ID @pytest.fixture def classical_subgenre_id(): """Return test UPC.""" return test_consts.TEST_CLASSICAL_SUBGENRE_ID @pytest.fixture def test_track(): """Return a track dict for testing.""" return track_seed.track_seed_data[1] @pytest.fixture def test_delete_product_id(): """Return a product id for testing of reorder functionality. Used in test_track_persister.py """ return test_consts.TEST_DELETE_PRODUCT_ID @pytest.fixture def test_single_track_product_id(): """Return id of product which has only one track.""" return test_consts.TEST_SINGLE_TRACK_PRODUCT_ID @pytest.fixture def test_track_response( test_track_model_response, track_localizations_with_artist_types): """Return a track response.""" track = test_track_model_response.copy() track[track_field.LOCALIZATIONS] = track_localizations_with_artist_types return track @pytest.fixture def test_track_valid_response( test_track_model_valid_response, track_localizations_with_artist_types): """Return a track response.""" track = test_track_model_valid_response.copy() track[track_field.LOCALIZATIONS] = track_localizations_with_artist_types return track @pytest.fixture def test_track_model_response(): """Return a track response.""" tuid = 2 owner_type = tuid_to_owner_type_mapping(tuid) track = { track_field.TUID: tuid, track_field.PRODUCT_ID: test_consts.TEST_PRODUCT_ID, track_field.TRACK_NAME: 'Nine Lives vol 1 track 2', track_field.VOLUME_NUMBER: 1, track_field.TRACK_NUMBER: 2, track_field.VERSION: None, track_field.META_LANGUAGE_CODE: None, track_field.ISRC: None, track_field.P_INFO: None, track_field.EXPLICIT: 'N', track_field.OWNERSHIP_RIGHTS: owner_type, track_field.RECORDING_COUNTRY_ID: 1, track_field.THIRD_PARTY_PUBLISHER: 'N', track_field.US_PUBLISHING_OBLIGATION: None, track_field.UPC: 123456789012, track_field.ARTISTS: [ {'track_artist_id': 4, 'type': 'performer', 'name': 'Snowball'}], track_field.PUBLISHERS: [ {'track_publisher_id': 3, 'type': 'publisher', 'name': 'Kitty Cat Productions'}, {'track_publisher_id': 4, 'type': 'publisher', 'name': 'Meow Meow Meow'}], track_field.WRITERS: [ {'track_writer_id': 3, 'type': 'writer', 'name': 'Furball'}] } track[track_field.ORIGINAL_RIGHTS_HOLDER_COUNTRY_ID] = \ track_to_producer_natl_mapping(track) return track @pytest.fixture def test_track_model_valid_response(): """Return a track response.""" tuid = 2 owner_type = tuid_to_owner_type_mapping(tuid) track = { track_field.TUID: tuid, track_field.PRODUCT_ID: test_consts.TEST_PRODUCT_ID, track_field.TRACK_NAME: 'Nine Lives vol 1 track 2', track_field.VOLUME_NUMBER: 1, track_field.TRACK_NUMBER: 2, track_field.VERSION: None, track_field.META_LANGUAGE_CODE: None, track_field.ISRC: 'US1234567890', track_field.P_INFO: '2015 Meow Records', track_field.EXPLICIT: 'Y', track_field.OWNERSHIP_RIGHTS: owner_type, track_field.RECORDING_COUNTRY_ID: 1, track_field.THIRD_PARTY_PUBLISHER: 'N', track_field.US_PUBLISHING_OBLIGATION: None, track_field.UPC: 123456789012, track_field.PREVIEW_START_TIME: None, track_field.ARTISTS: [ {'track_artist_id': 4, 'type': 'performer', 'name': 'Snowball'}], track_field.PUBLISHERS: [ {'track_publisher_id': 3, 'type': 'publisher', 'name': 'Kitty Cat Productions'}, {'track_publisher_id': 4, 'type': 'publisher', 'name': 'Meow Meow Meow'}], track_field.WRITERS: [ {'track_writer_id': 3, 'type': 'writer', 'name': 'Furball'}] } track[track_field.ORIGINAL_RIGHTS_HOLDER_COUNTRY_ID] = \ track_to_producer_natl_mapping(track) return track @pytest.fixture def test_product_tracks_response(): """Return a list of tracks for testing.""" return product_all_tracks_response.test_all_tracks_on_product_id.copy() @pytest.fixture def test_product_valid_tracks_response(): """Return a list of tracks for testing.""" return product_all_valid_tracks_response\ .test_all_valid_tracks_on_product_id.copy() @pytest.fixture def test_product_valid_tracks_logic_response(): """Return a list of tracks for testing.""" return test_funcs.make_track_logic_response( product_all_valid_tracks_response .test_all_valid_tracks_on_product_id.copy()) @pytest.fixture def test_product_valid_tracks_light_logic_response(): """Return a list of tracks for testing.""" return test_funcs.make_track_logic_response( product_all_valid_tracks_response .test_all_valid_tracks_on_product_id_light.copy()) @pytest.fixture def track_max_tuid(): """Return max tuid in test database.""" return track_seed.get_max_tuid() @pytest.fixture def test_track_correction_data(): """Return track correction post data.""" return [ {'key_value': [{'type': 'performer', 'name': 'test'}], 'field_name': 'track_artist', 'key_id': 12345, 'table_name': 'track'}, {'key_value': [{'type': 'performer', 'name': 'test'}], 'field_name': 'track_artist', 'key_id': 12345, 'table_name': 'track'}, {'key_value': [{'type': 'performer', 'name': 'test'}], 'field_name': 'track_artist', 'key_id': 12345, 'table_name': 'track'}, {'key_value': [{'type': 'performer', 'name': 'test'}], 'field_name': 'track_artist', 'key_id': 12345, 'table_name': 'track'} ] @pytest.fixture def test_multi_tracks_product(): """Return product 2 info for testing.""" return { 'product_id': test_consts.TEST_MULTI_TRACK_PRODUCT_ID, 'volumes': (5, 4, 3,), 'upc': test_consts.TEST_NEW_UPC } @pytest.fixture def track_max_product_id(): """Return max product_id in test database.""" return track_seed.get_max_product_id() @pytest.fixture def track_create_data(): """Return test create data.""" return track_create_fixture.track_create_data.copy() @pytest.fixture def track_create_expected_response(): """Return expected result from track creation.""" return deepcopy(track_create_fixture.track_create_response) @pytest.fixture def track_create_expected_logic_response(): """Return expected result from track creation.""" return test_funcs.make_track_logic_response( track_create_fixture.track_create_response) @pytest.fixture def track_bulk_create_data(): """Return test track bulk create data.""" return track_bulk_create_fixture.request_data.copy() @pytest.fixture def track_bulk_create_new_prod_response(): """Return expected result from new product track bulk creation.""" return track_bulk_create_fixture.new_prod_response @pytest.fixture def track_bulk_create_new_prod_logic_response(): """Return expected result from new product track bulk creation. This adds the localizations after the logic layer processes the response. """ return test_funcs.make_track_logic_response( track_bulk_create_fixture.new_prod_response) @pytest.fixture def track_bulk_create_existing_prod_response(): """Return expected result from existing product track bulk creation.""" return deepcopy(track_bulk_create_fixture.existing_prod_response) @pytest.fixture def track_bulk_create_existing_prod_logic_response(): """Return expected result from existing product track bulk creation. This adds the localizations after the logic layer processes the response. """ return test_funcs.make_track_logic_response( track_bulk_create_fixture.existing_prod_response) @pytest.fixture def product_get_response(): """Return expected result from track creation.""" return { 'product_id': test_consts.TEST_PRODUCT_ID, 'upc': 123456789012 } @pytest.fixture def track_update_tuid(): """Return track tuid to update.""" return track_update_fixture.TUID @pytest.fixture def track_update_data(): """Return test track update data.""" return track_update_fixture.track_update_data.copy() @pytest.fixture def tds_get_pub_obl(track_factory): """Test data for getting publishing obligation data for product tracks.""" seed_data = [] seed_data.append(track_factory()) seed_data.append(track_factory( us_publishing_obligation=pub_obl.PUBLIC_DOMAIN, third_party_publisher='')) seed_data.append(track_factory( us_publishing_obligation=pub_obl.CONTROLLED_BY_YOUR_LABEL)) seed_data.append(track_factory( us_publishing_obligation=pub_obl.COMPOSITION, third_party_publisher='N')) seed_data.append(track_factory( us_publishing_obligation=pub_obl.COMPOSITION, third_party_publisher='Y', publishers=['pub 1', 'pub 2'])) num_seed_items = len(seed_data) def validate_response_message(response_message): assert len(response_message['items']) == num_seed_items # Set of valid fields valid_fields = set(track_field.PUBLISHING_OBLIGATION_FIELDS) valid_fields |= {track_field.PUBLISHER_NAMES} item_no_pub_obl = response_message['items'][0] assert item_no_pub_obl[track_field.US_PUBLISHING_OBLIGATION] is None assert item_no_pub_obl[track_field.THIRD_PARTY_PUBLISHER] is None for item in response_message['items']: for field in item: assert field in valid_fields assert len(item) == len(valid_fields) assert item[track_field.THIRD_PARTY_PUBLISHER] \ in ('Y', 'N', None) return { 'product_id': 1, 'seed_data': seed_data, 'validate_response_message': validate_response_message } @pytest.fixture def tds_update_pub_obl_for_mech_admin(track_factory): """Test data for updating publishing obligations.""" request_data = [ { track_field.TUID: 1, track_field.US_PUBLISHING_OBLIGATION: pub_obl.CONTROLLED_BY_YOUR_LABEL }, { track_field.TUID: 2, track_field.THIRD_PARTY_PUBLISHER: 'N', track_field.US_PUBLISHING_OBLIGATION: pub_obl.COMPOSITION }, { track_field.TUID: 3, track_field.THIRD_PARTY_PUBLISHER: 'Y', track_field.PUBLISHER_NAMES: ['pub1'], track_field.US_PUBLISHING_OBLIGATION: pub_obl.COMPOSITION } ] seed_data = [track_factory() for delta in request_data] def validate_response_message(response_message): i = 0 required_fields = { *track_field.PUBLISHING_OBLIGATION_FIELDS, track_field.PUBLISHER_NAMES, } for track in response_message['items']: assert len(track) == len(required_fields) delta = request_data[i] for key, value in delta.items(): assert key in track assert track[key] == value i += 1 return { 'product_id': 1, 'seed_data': seed_data, 'request_data': request_data, 'validate_response_message': validate_response_message } @pytest.fixture def tds_validate_pub_obl_for_mech_admin(track_factory): """Test data for publishing obligations.""" seed_data = [] seed_data.append(track_factory()) seed_data.append(track_factory( us_publishing_obligation=pub_obl.PUBLIC_DOMAIN)) seed_data.append(track_factory( us_publishing_obligation=pub_obl.CONTROLLED_BY_YOUR_LABEL)) seed_data.append(track_factory( us_publishing_obligation=pub_obl.COMPOSITION, third_party_publisher='Y', publishers=['pub 1', 'pub 2'])) seed_data.append(track_factory( us_publishing_obligation=pub_obl.COMPOSITION, third_party_publisher='Y')) def validate_response_message(response_message): assert response_message['total_tracks'] == 5 assert response_message['valid_tracks'] == 3 # Make sure errors are properly filled in errors = response_message['errors'] assert len(errors) == 2 for index, field in [ (0, track_field.US_PUBLISHING_OBLIGATION,), (1, track_field.PUBLISHER_NAMES,)]: assert field in errors[index] assert errors[index][field]['validator'] == 'required' assert errors[index][field]['validator_value'] return { 'product_id': 1, 'seed_data': seed_data, 'validate_response_message': validate_response_message } @pytest.fixture def track_reorder_data(): """Return test track reorder data.""" return track_reorder_fixture.track_reorder_data.copy() @pytest.fixture def track_reorder_expected_response(): """Return expected result from track reordering.""" return track_reorder_fixture.track_reorder_expected_response.copy() @pytest.fixture def product_apply_to_all_data(): """Return test apply to all data.""" return product_apply_to_all_fixture.request.copy() @pytest.fixture def product_apply_to_all_generate_isrc_response(): """Return expected result for generating isrc.""" return product_apply_to_all_fixture.generate_isrc_response.copy() @pytest.fixture def product_apply_to_all_response(): """Return expected result from track reordering.""" return product_apply_to_all_fixture.response.copy() @pytest.fixture def tds_product_apply_to_all_version( logic_headers, test_product_id, test_product_tracks_response): """Return test apply to all data.""" fixture = product_apply_to_all_fixture track_persister_response = [] version = fixture.request_apply_version_data[track_field.VERSION] for track in test_product_tracks_response[api_consts.ITEMS]: track_delta = { track_field.TUID: track[track_field.TUID], track_field.META_LANGUAGE_CODE: 'ENG', track_field.VERSION: version } track_persister_response.append(track_delta) # Make logic response data logic_response_data = [] localizations = fixture.request_apply_version_data[ track_field.LOCALIZATIONS] for track in track_persister_response: track_delta = track.copy() del track_delta[track_field.META_LANGUAGE_CODE] # Copy localizations track_delta[track_field.LOCALIZATIONS] = [] for item in localizations: track_delta[track_field.LOCALIZATIONS].append( test_funcs.make_localization_dict( tuid=track_delta[track_field.TUID], **item)) logic_response_data.append(track_delta) test_ows_product_loc = test_funcs.make_localization_dict( tuid=1, language_id=1, track_name='', version='slow') test_ows_product_loc[track_field.ARTISTS] = {} return TestDataSet( product_id=test_product_id, logic_headers=logic_headers, request_data=fixture.request_apply_version_data.copy(), ows_product_track_localizations=test_funcs.wrap_api_results( [test_ows_product_loc]), persister_response=response.Response( status=200, message=test_funcs.wrap_api_results( track_persister_response)), logic_response=response.Response( status=200, message=test_funcs.wrap_api_results( logic_response_data)) ) @pytest.fixture def product_role_apply_to_all_data(): """Return test apply role to request data.""" return { 'names': ['MC Neroh', 'Jailim'] } @pytest.fixture(params=validation_const.VARIOUS_ARTISTS) def invalid_artist_name(request): """Return invalid artist name.""" return request.param @pytest.fixture def product_role_apply_to_all_invalid_name(invalid_artist_name): """Return test apply role to request data with invalid name.""" return { 'names': [invalid_artist_name] } @pytest.fixture def product_role_apply_to_all_role_type(): """Return test apply role to request role_type.""" return 'remixer' @pytest.fixture def track_update_expected_response(): """Return expected result from track update.""" return track_update_fixture.track_update_response.copy() @pytest.fixture def track_update_expected_response_without_localizations(): """Return expected result from track update without localizations.""" data = track_update_fixture.track_update_response.copy() data.pop(track_field.LOCALIZATIONS) return data @pytest.fixture def track_update_expected_response_with_isrc(): """Return expected result from track update.""" update_isrc = track_update_fixture.track_update_response.copy() update_isrc['isrc'] = 'ABC12345678' return update_isrc @pytest.fixture def test_tuids(): """Return non-contiguous list of tuids belonging to the same product.""" return [2, 3, 5, 8, 9] @pytest.fixture def test_tuids_multiple_products(): """Return non-contiguous list of tuids belonging to the same product.""" return [2, 3, 5, 8, 9, 12, 16, 20] @pytest.fixture def test_product_id(): """Return a product_id for testing.""" return test_consts.TEST_PRODUCT_ID @pytest.fixture def test_valid_product_id(): """Return a product_id for testing.""" return test_consts.TEST_VALID_PRODUCT_ID @pytest.fixture def test_product_ids(): """Return product_ids for testing.""" return test_consts.TEST_PRODUCT_IDS @pytest.fixture def test_invalid_product_ids(): """Return product_ids for testing with one invalid value.""" return test_consts.TEST_INVALID_PRODUCT_IDS @pytest.fixture def test_upc(): """Return a upc for testing.""" return 19876543312 @pytest.fixture def test_correction_id(): """Return correction id for testing.""" return 123 @pytest.fixture def test_correlation_id(): """Return correlation id for testing.""" return '1234' @pytest.fixture def test_orchard_user_id(): """Return orchard user id for testing.""" return 'alw:123' @pytest.fixture def track_localization_data(): """Fixture for track localization data json.""" return { 'track_name': 'Gangsta\'s Paradise', 'version': 'Explicit', 'artists': [ { 'track_artist_id': 1, 'name': 'Coolio', 'type': 'performer' } ] } @pytest.fixture def track_role_localization_data(): """Fixture for track localization data json.""" return { 'names': ['Hoofy', 'Woofy'], track_field.LOCALIZATIONS: [ { track_field.LANGUAGE_ID: 2, 'names': ['Loopie', 'Whompy'] } ] } @pytest.fixture def update_track_localization_response(track_localization_data): """Fixture for track localization response.""" data = track_localization_data.copy() data.update({ 'tuid': 123, 'language_id': 1 }) return response.Response(data) @pytest.fixture def track_localization_failed_response(): """Fixture for track localization failed response.""" return response.create_error_response( error.OWS_PRODUCT_ERROR_CODE, 'Server error', 500) @pytest.fixture def patch_ows_product_get_product(mocker, test_upc, test_product_id): """Path response from ows_product for get product by id.""" get_response = response.Response( status=200, message={ 'upc': test_upc, 'vendor_id': 100, 'product_id': test_product_id, 'status': 'in_content' }) mocker.patch( 'backend.models.ows_product.get_product_by_product_id', return_value=get_response) @pytest.fixture def patch_ows_product_get_product_failed(mocker, test_upc, test_product_id): """Path failed response from ows_product for get product by id.""" get_response = response.create_error_response( status=500, message='Server error', code=error.OWS_PRODUCT_ERROR_CODE) mocker.patch( 'backend.models.ows_product.get_product_by_product_id', return_value=get_response) @pytest.fixture def track_localization_get_data(): """Fixture for ows-product response for get localizations.""" return localization_fixture.track_localization_ows_product_response() @pytest.fixture def track_localizations_with_artist_types(): """Fixture for list of track localizations for full track JSON.""" return localization_fixture.track_localizations_with_artist_types() @pytest.fixture def track_localization_put_data(): """Fixture for ows-product response for get localizations.""" return ( localization_fixture. track_localization_ows_product_response_put_track()) @pytest.fixture def track_localizations_with_artist_types_put_track(): """Fixture for list of track localizations for full track JSON.""" return ( localization_fixture.track_localizations_with_artist_types_put_track()) @pytest.fixture def get_validate_tracks_response(): """Fixture for get track validate response.""" return response.Response(track_validate_response.response) @pytest.fixture def mock_claim_new_isrcs(mocker): """Mock isrc generation.""" mocker.patch.object( TrackQuery, 'claim_new_isrcs', side_effect=lambda session, number_of_isrcs: [ ''.join([random.choice(string.ascii_letters) for _ in range(16)]) for i in range(number_of_isrcs) ]) @pytest.fixture def tds_copy_localization_single_track_no_artist( track_factory, track_localization_factory): """Test DataSet for copying single track with single localization.""" source_track = track_factory( version='radio mix', writers__count=2, publishers__count=1) localizations = track_localization_factory( track=source_track, language_ids=[1, 2]) dest_track_dict = source_track.to_dict() dest_track_dict[track_field.TUID] += 1 dest_track_dict[track_field.PRODUCT_ID] += 1 dest_track_dict[track_field.UPC] = test_consts.TEST_NEW_UPC def validate_copied_track(track_dict, validate_logical=False): exclude_fields = [] if validate_logical: exclude_fields = [ track_field.PUBLISHERS, track_field.US_PUBLISHING_OBLIGATION, track_field.THIRD_PARTY_PUBLISHER] is_valid, error_msg = test_funcs.validate_track_is_copy( source_track.to_dict(), track_dict, exclude_fields=exclude_fields) assert is_valid, error_msg test_funcs.validate_localizations_copy_result( track_dict, localizations) if validate_logical: error_msg = test_funcs.validate_track_response(track_dict) assert not error_msg source_dest_track_list = [{ 'source': source_track.to_dict(), 'destination': deepcopy(dest_track_dict)}] return { 'seed_data': [source_track], 'source_track': source_track.to_dict(), 'destination_track': deepcopy(dest_track_dict), 'source_localizations': deepcopy(localizations), 'source_dest_track_list': source_dest_track_list, 'destination_product_id': source_track.product_id + 1, 'destination_product_upc': test_consts.TEST_NEW_UPC, 'tuids': [source_track.tuid], 'validate_copied_track': validate_copied_track } @pytest.fixture def tds_copy_localization_single_track_with_artists( track_factory, track_localization_factory): """Test DataSet for copying single track with single localization.""" source_track = track_factory( version='radio mix', performer__count=3, remixer__count=2, featuring__count=2) localizations = track_localization_factory( track=source_track, language_ids=[1, 2]) # Remove remixer localizations for loc in localizations: loc[track_field.ARTISTS] = [ artist for artist in loc[track_field.ARTISTS] if artist[track_field.TYPE] != 'remixer'] dest_track_dict = source_track.to_dict() dest_track_dict[track_field.TUID] += 1 dest_track_dict[track_field.PRODUCT_ID] += 1 dest_track_dict[track_field.UPC] = test_consts.TEST_NEW_UPC for artist in dest_track_dict[track_field.ARTISTS]: artist[track_field.TRACK_ARTIST_ID] += len(source_track.artists) def validate_copied_track(track_dict, validate_logical=False): exclude_fields = [] if validate_logical: exclude_fields = [ track_field.PUBLISHERS, track_field.US_PUBLISHING_OBLIGATION, track_field.THIRD_PARTY_PUBLISHER] is_valid, error_msg = test_funcs.validate_track_is_copy( source_track.to_dict(), track_dict, exclude_fields=exclude_fields) assert is_valid, error_msg test_funcs.validate_localizations_copy_result( track_dict, localizations) if validate_logical: error_msg = test_funcs.validate_track_response(track_dict) assert not error_msg source_dest_track_list = [{ 'source': source_track.to_dict(), 'destination': deepcopy(dest_track_dict)}] return { 'seed_data': [source_track], 'source_track': source_track.to_dict(), 'destination_track': deepcopy(dest_track_dict), 'source_localizations': deepcopy(localizations), 'source_dest_track_list': source_dest_track_list, 'destination_product_id': source_track.product_id + 1, 'destination_product_upc': test_consts.TEST_NEW_UPC, 'tuids': [source_track.tuid], 'validate_copied_track': validate_copied_track } @pytest.fixture def copy_assets_response(): """Mock success response for ows_assets.copy_assets.""" return response.Response({'status': 'ok'}) @pytest.fixture def copy_assets_failed_response(): """Mock failure response for ows_assets.copy_assets.""" message = { 'message': 'Error', 'code': 'error_code' } return response.create_error_response( code=error.OWS_ASSETS_ERROR_CODE, message=message) @pytest.fixture def tracks_delete_response(): """Mock successfull response for deleting multiple tracks.""" return response.Response( track_delete_fixture.track_delete_expected_response) @pytest.fixture def a_very_long_track_name(): """Track name that is more than 200 characters long.""" return localization_fixture.LONG_LOCALIZED_TRACK_NAME @pytest.fixture def localization_languages(): """Fixture for localization languages.""" return languages_seed_data @pytest.fixture def localization_languages_data(): """Fixture for localization languages response from ows-product.""" return wrap_api_results(languages_seed_data) @pytest.fixture def instant_grat_factory(): """Return an Instant Grat factory.""" factories.InstantGratFactory.reset_sequence() return factories.InstantGratFactory @pytest.fixture def context(): """Return flask app context, including a mock logger. Returns: AppContext: flask app context, including a mock logger. """ context = application.app.app_context() context.g.ows = flask_logger.Ows() context.g.ows.log = MockLogAdaptor() context.g.request_context = MagicMock() return context @pytest.fixture def mock_app(): """Mock app.""" with application.app.app_context() as context: context.g.ows = flask_logger.Ows() context.g.ows.correlation_id = 'correlation_id' context.g.ows.log = MockLogAdaptor() context.g.request_context = MagicMock() yield context @pytest.fixture def track_sample_tuid(): """Fixture for track_sample tuid.""" return track_sample_fixture.TUID @pytest.fixture def track_sample_id(): """Fixture for track_sample id.""" return track_sample_fixture.SAMPLE_ID @pytest.fixture def valid_track_sample(): """Fixture for track_sample data.""" return track_sample_fixture.valid_track_sample.copy() @pytest.fixture def valid_track_sample_artist(): """Fixture for track_sample artist data.""" return track_sample_fixture.valid_track_sample_artist.copy() @pytest.fixture def valid_track_sample_request(valid_track_sample, valid_track_sample_artist): """Fixture for valid track_sample request.""" request_data = valid_track_sample.copy() request_data['artists'] = valid_track_sample_artist return request_data @pytest.fixture def valid_track_sample_response(valid_track_sample_request): """Fixture for valid track_sample response.""" response_data = valid_track_sample_request.copy() response_data['sample_id'] = 1 response_data['unique_track_id'] = track_sample_fixture.TUID response_data['artists'][0]['artist_id'] = 1 response_data['artists'][1]['artist_id'] = 2 response_data['artists'][2]['artist_id'] = 3 return response_data @pytest.fixture def track_bulk_create_track_with_metadata_data(track_update_data): """Return test track bulk create with metadata data.""" track_1 = track_update_data.copy() input_track_1 = track_1.copy() del input_track_1['tuid'] track_2 = track_update_data.copy() input_track_2 = track_2.copy() del input_track_2['tuid'] return {'tracks': [input_track_1, input_track_2]} @pytest.fixture def track_bulk_create_track_with_metadata_response(track_update_data): """Return test track bulk create with metadata data.""" track_1 = track_update_data.copy() track_2 = track_update_data.copy() return {'items': [track_1, track_2]} @pytest.fixture def track_bulk_create_data_for_explicit(): """Return test track bulk create data.""" return track_bulk_create_fixture.request_data_for_explicit.copy() @pytest.fixture def track_bulk_create_new_prod_response_for_explicit(): """Return expected result from new product track bulk creation.""" return track_bulk_create_fixture.new_prod_response_for_explicit @pytest.fixture def track_bulk_create_new_prod_logic_response_for_explicit(): """Return expected result from new product track bulk creation. This adds the localizations after the logic layer processes the response. """ return test_funcs.make_track_logic_response( track_bulk_create_fixture.new_prod_response_for_explicit) @pytest.fixture def track_bulk_create_existing_prod_response_for_explicit(): """Return expected result from existing product track bulk creation.""" return deepcopy( track_bulk_create_fixture.existing_prod_response_for_explicit) @pytest.fixture def track_bulk_create_existing_prod_logic_response_for_explicit(): """Return expected result from existing product track bulk creation. This adds the localizations after the logic layer processes the response. """ return test_funcs.make_track_logic_response( track_bulk_create_fixture.existing_prod_response_for_explicit) @pytest.fixture def track_artist_role_update_request_classical(): """Return test classical genre and subgenre request data.""" return { 'genre_id': 12, 'subgenre_id': 123 } @pytest.fixture def track_artist_role_update_request_nonclassical(): """Return test non-classical genre and subgenre request data.""" return { 'genre_id': 13, 'subgenre_id': 124 } @pytest.fixture def track_release_correction_details(): """Return test track correction detail data.""" return {'release_correction_id': 246069, 'release_id': 2664741, 'status': 'active', 'items': [{'release_correction_detail_id': 3811735, 'table_name': 'releases', 'field_name': 'genre_id', 'key_id': 2664741, 'key_value': 12}, {'release_correction_detail_id': 3811736, 'table_name': 'releases', 'field_name': 'release_subgenre', 'key_id': 2664741, 'key_value': [269]}, {'release_correction_detail_id': 3811737, 'table_name': 'releases', 'field_name': 'featuring', 'key_id': 2664741, 'key_value': [{'artist_name': 'test feat', 'role': 'featuring'}]}, {'release_correction_detail_id': 3811737, 'table_name': 'releases', 'field_name': 'composer', 'key_id': 2664741, 'key_value': [{'artist_name': 'test composer', 'role': 'composer'}]}, {'release_correction_detail_id': 3812040, 'table_name': 'track', 'field_name': 'track_artist', 'key_id': 30452965, 'key_value': [{'name': 'test performer', 'type': 'performer'}]}, {'release_correction_detail_id': 3821785, 'table_name': 'track', 'field_name': 'featuring', 'key_id': 30452965, 'key_value': [{'name': 'test', 'type': 'performer'}]}, {'release_correction_detail_id': 3821785, 'table_name': 'track', 'field_name': 'remixer', 'key_id': 30452965, 'key_value': [{'name': 'test', 'type': 'remixer'}]}, {'release_correction_detail_id': 3821785, 'table_name': 'track', 'field_name': 'producer', 'key_id': 30452965, 'key_value': [{'name': 'test', 'type': 'producer'}]}]} @pytest.fixture def get_tracks_details(): """Returns tracks for the product.""" return {'items': [ {'tuid': 30452965, 'product_id': 2664741, 'upc': 194491104580}, {'tuid': 30452966, 'product_id': 2664741, 'upc': 194491104580}]} @pytest.fixture def track_correction_details_for_all_tracks(): """Returns correction detail post data for tracks.""" return [{'key_value': [{'name': 'test performer', 'type': 'performer'}], 'field_name': 'track_artist', 'key_id': 30452965, 'table_name': 'track'}, {'key_value': [], 'field_name': 'featuring', 'key_id': 30452965, 'table_name': 'track'}, {'key_value': [], 'field_name': 'remixer', 'key_id': 30452965, 'table_name': 'track'}, {'key_value': [], 'field_name': 'producer', 'key_id': 30452965, 'table_name': 'track'}, {'key_value': [], 'field_name': 'featuring', 'key_id': 30452966, 'table_name': 'track'}, {'key_value': [], 'field_name': 'producer', 'key_id': 30452966, 'table_name': 'track'}, {'key_value': [{'type': 'composer', 'name': 'BTS'}], 'field_name': 'track_artist', 'key_id': 30452966, 'table_name': 'track'}] @pytest.fixture def track_spatial_data(): """Fixture for track_spatial data.""" return track_spatial_fixture.track_spatial_data.copy() @pytest.fixture def track_spatial_data_invalid(): """Fixture for track_spatial data.""" return track_spatial_fixture.track_spatial_data_invalid.copy() CREATE_TABLE_RELEASES = """ create table if not exists releases ( release_id int not null, release_name varchar(255) not null, release_status varchar(255) not null, distribution_format_id tinyint not null ); """ INSERT_TEST_PRODUCTS = """ replace into releases (release_id, release_name, release_status, distribution_format_id) values (1, 'Product A', 'label_processing', 1), (2, 'Product B', 'in_content', 1), (3, 'Product C', 'transfer_to_content', 1), (4, 'Product D', 'in_content', 57), (5, 'Product E', 'in_content', 80), (72, 'Product Q', 'transfer_to_content', 57), (76, 'Product R', 'in_content', 1), (79, 'Product S', 'in_content', 1), (89, 'Product T', 'in_content', 57), (90, 'Product U', 'label_processing', 82), (101, 'Product V', 'in_content', 80), (205, 'Product W', 'in_content', 81), (301, 'Product AA', 'in_content', 80), (302, 'Product BB', 'in_content', 80), (303, 'Product CC', 'in_content', 80), (304, 'Product DD', 'in_content', 80); """ CREATE_TABLE_DISTRIBUTION_FORMAT = """ create table if not exists distribution_format ( distribution_format_id int not null, context_type varchar(255) not null ); """ INSERT_TEST_DISTRIBUTION_FORMATS = """ replace into distribution_format (distribution_format_id, context_type) values (1, 'digital'), (57, 'digital'), (80, 'physical'), (81, 'physical'), (82, 'physical'); """ INSERT_TEST_TRACKS = """ replace into track (id, release_id, track_name, isrc, explicit_lyrics, upc, cd, track_id) values (132321, 72, 'Track 1', 'isrc2', 'N', 1232234684624, 1, 1), (123322, 76, 'Track 2', 'isrc1', 'N', 12322344684624, 1, 1), (312323, 76, 'Track 3', 'isrc2', 'N', 123223484624, 1, 2), (123324, 101, 'Track 4', 'isrc2', 'N', 123223484624, 1, 3), (123325, 301, 'Track 55', 'isrc2', 'N', 123223484624, 1, 3), (123326, 302, 'Track 66', 'isrc2', 'N', 123223484624, 1, 3), (123327, 303, 'Track 77', 'isrc2', 'N', 123223484624, 1, 3), (123328, 304, 'Track 88', 'isrc2', 'N', 123223484624, 1, 3), (546456, 72, 'Track 5', 'isrc1', 'N', 12325234624, 1, 1), (666661, 551, 'Track 551', 'isrc3', 'N', 111111111111, 1, 1), (666662, 552, 'Track 552', 'isrc3', 'N', 111111111112, 1, 1), (666663, 553, 'Track 553', 'isrc4', 'N', 111111111113, 1, 1), (666664, 554, 'Track 554', 'isrc4', 'N', 111111111114, 1, 1), (666665, 555, 'Track 555', 'isrc4', 'N', 111111111115, 1, 1), (666666, 556, 'Track 556', 'isrc4', 'N', 111111111116, 1, 1), (666667, 557, 'Track 557', 'isrc4', 'N', 111111111117, 1, 1), (666668, 558, 'Track 558', 'isrc4', 'N', 111111111118, 1, 1), (666669, 559, 'Track 559', 'isrc4', 'N', 111111111119, 1, 1), (666670, 560, 'Track 560', 'isrc4', 'N', 111111111120, 1, 1), (666671, 561, 'Track 561', 'isrc4', 'N', 111111111121, 1, 1), (666672, 562, 'Track 562', 'isrc4', 'N', 111111111122, 1, 1); """ @pytest.fixture def test_get_cross_track_isrc_mismatch_response(): """Return cross track ISRC mismatch response for testing.""" return [ { 'isrc': 'US1234567890', 'tuid': 333, 'cross_track_isrc_mismatch': { 'message': json.dumps({ 'matches': [ { 'tuid': 444, 'isrc': 'US1234567891', 'product_id': 555, 'upc': '199999999999', 'vendor_id': 25824 } ] }) } } ] @pytest.fixture def test_get_cross_isrc_osr_mismatch_response(): """Return cross ISRC OSR mismatch response for testing.""" return [ { 'isrc': 'US1234567890', 'tuid': 333, 'cross_isrc_osr_mismatch': { 'message': json.dumps({ 'matches': [ { 'tuid': 555, 'isrc': 'US1234567892', 'product_id': 666, 'upc': '288888888888', 'vendor_id': 12345 } ] }) } } ] @pytest.fixture def test_sound_recording_matches(): """Return sound recording matches for testing.""" return { 1: { "tuid": 1, "isrc": "DUMMYISRC001", "matched_tracks": [ { "tuid": 3333331, "isrc": "MATCHISRC001", "product_id": 4444001, "upc": "UPC000001", "release_status": "in_content", "subaccount_id": 0, "vendor_id": 55555 } ] } } @pytest.fixture def test_validate_cross_track_isrc_mismatch_reporting_response(): """Return cross track ISRC mismatch response for testing.""" return [ { "product_id": 1, "track_tuid": 1, "track_isrc": "DUMMYISRC001", "warning_type": "Cross Track ISRC Mismatch", "match_tuid": 3333331, "match_isrc": "MATCHISRC001", "match_product_id": 4444001, "match_upc": "UPC000001", "match_vendor_id": 55555 } ] @pytest.fixture def test_validate_cross_isrc_osr_mismatch_reporting_response(): """Return cross track ISRC mismatch response for testing.""" return [ { "product_id": 1111111, "track_tuid": 2222222, "track_isrc": "DUMMYISRC001", "warning_type": "Potential ISRC Misuse", "match_tuid": 3333332, "match_isrc": "MATCHISRC002", "match_product_id": 4444002, "match_upc": "UPC000002", "match_vendor_id": 55555 } ] @pytest.fixture def test_get_same_isrc_different_osr_response(): """Return same ISRC different OSR response for testing.""" return [ { 'tuid': 3333331, 'isrc': 'MATCHISRC001', 'product_id': 4444001, 'upc': 'UPC000001', 'release_status': 'in_content', 'subaccount_id': 0, 'vendor_id': 55555 } ] @pytest.fixture def test_isrc_response_validation_response(): """Return validation response for testing.""" return { 'items': [ { "product_id": 1111111, "distribution_format_id": 1, "project_id": 2222222, "product_type_id": 1, "upc": 123434567890, "display_upc": "123434567890", "context_type": "digital", "vendor_id": 12345, "subaccount_id": 12345, "release_date": "2050-01-01", "status": "label_processing", "product_name": "Test Product", "deletions": "N", "not_for_distribution": "N" }, ] } @pytest.fixture def make_graph_node(): """Node generating factory as fixture.""" graph = Graph() def _make_node(node_id, labels=(), data={}): data['id'] = node_id return Node(graph, node_id, node_id, n_labels=labels, properties=data) return _make_node @pytest.fixture def validate_artists_request_payload(): """Sample track_artists payload for ows-blacklist-manager /validate-artists.""" return { 'track_artists': [ {'tuid': 1, 'artists': [{'name': 'Some Artist', 'role': 'performer'}]}, {'tuid': 2, 'artists': [{'name': 'Quadron', 'role': 'performer'}]}, ] } @pytest.fixture def spotify_watchlist_validation_errors_response(): """Response body from ows-blacklist-manager when watchlist matches are found.""" return { 'validation_errors': { 'track_artists': [ { 'matches': [ { 'name': 'Quadron', 'role': 'performer', 'reason': 'Spotify Watchlist Artist' } ], 'tuid': 2 } ] } } @pytest.fixture def all_artist_types_track(): """A track dict containing one artist of every type in tf.ARTIST_TYPES. Each artist is uniquely named so tests can assert on exact membership. The artists list uses a distinct 'original_' prefix so corrections can be clearly distinguished from originals in assertions. """ return { 'tuid': 10, 'artists': [ {'type': 'featuring', 'name': 'original_featuring'}, {'type': 'performer', 'name': 'original_performer'}, {'type': 'producer', 'name': 'original_producer'}, {'type': 'remixer', 'name': 'original_remixer'}, {'type': 'composer', 'name': 'original_composer'}, {'type': 'orchestra', 'name': 'original_orchestra'}, {'type': 'conductor', 'name': 'original_conductor'}, {'type': 'ensemble', 'name': 'original_ensemble'}, ], }