"""Functional tests for validating product tracks. Test endpoint: GET /product//tracks/validate """ from collections import ChainMap import json from oto import response from oto import status as response_code from owsrequest import request from owsrequest import test_utils import pytest from backend import features from backend.connectors import mysql from backend.constants import error from backend.constants import feature_flag from backend.logic import track as track_logic from backend.models import ows_account from backend.utils import validation import tests.conftest from tests.testutils import db from tests.testutils import mocks from backend.constants.product import IS_ERR_CORR_MODE from tests.testutils import constants as test_consts from backend.models import track_spatial @db.test_schema_no_seed def test_validate_tracks_for_product( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1) db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_tracks_for_product_without_grass_headers( client, mocker, track_factory ): """Test validate tracks by product id is successful. without grass headers. """ mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_tracks_for_classical_product( mocker, logic_headers, logic_user, track_factory, test_product_id, classical_genre_id, classical_subgenre_id, mock_app ): """Verify classical track with missing composer has validation error.""" tracks = [ track_factory(performer__count=1, composer__count=1, writers__count=1), track_factory(performer__count=1, writers__count=1)] mocks.head_response(mocker) db.merge_model_objects(tracks) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_id=test_product_id, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id) validation_response = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert validation_response.message['total_tracks'] == len(tracks) assert validation_response.message['valid_tracks'] == 1 validation_error = validation_response.message['errors'][0] assert 'composer' in validation_error assert validation_error['tuid'] == 2 @db.test_schema_no_seed def test_validate_tracks_for_product_missing_localization( client, mocker, track_factory ): """Should return error because Chinese traditional isn't localized.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) track = track_factory.create( writers__count=1, performer__count=1, meta_language_code='CHI') db.merge_model_objects([track]) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_digital_get_chinese_product_by_id(mocker) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocks.ows_product_get_localization_languages(mocker) localization_error = { 'message': error.VALIDATION_ERROR_CHINESE_LOCALIZATION_REQUIRED.format( track_language='Chinese Traditional', product_meta_language='Chinese Simplified'), 'validator': 'required', 'validator_value': True, 'error_code': error.MISSING_CHINESE_LOCALIZATION } expected_error = { 'errors': [{'localizations': [localization_error], 'tuid': 1}], 'total_tracks': 1, 'valid_tracks': 0, 'valid': False} res = client.get( 'product/{}/tracks/validate'.format(track.to_dict()['product_id'])) res_body = json.loads(res.data.decode()) assert res_body == expected_error @db.test_schema_no_seed def test_validate_tracks_for_product_missing_complementary_localization( mocker, client, track_factory ): """Track has Chinese Simplified localization but no Chinese Traditional. Should return error because both localizations required. """ mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) track = track_factory.create( performer__count=1, writers__count=1) db.merge_model_objects([track]) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) localizations_data = { 'items': [{'tuid': 1, 'language_id': 6, 'artists': {}}] } mocks.ows_product_get_track_localizations(mocker, localizations_data) mocks.ows_product_digital_get_product_by_id(mocker) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocks.ows_product_get_localization_languages(mocker) localization_error = { 'message': error.VALIDATION_ERROR_BOTH_CHINESE_LANGS_REQUIRED, 'validator': 'required', 'validator_value': True, 'error_code': error.MISSING_BOTH_CHINESE_LANGUAGES } expected_error = { 'errors': [{'localizations': [localization_error], 'tuid': 1}], 'total_tracks': 1, 'valid_tracks': 0, 'valid': False} url = 'product/{}/tracks/validate'.format( track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body == expected_error @db.test_schema_no_seed def test_validate_tracks_for_product_success( mocker, client, track_factory ): """Track has no Chinese localizations, no errors should return.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_track_localizations(mocker, {'items': []}) track = track_factory.create( performer__count=1, writers__count=1) db.merge_model_objects([track]) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_product_by_id(mocker) url = 'product/{}/tracks/validate'.format( track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} @db.test_schema_no_seed def test_validate_tracks_for_product_too_many_localizations( mocker, client, track_factory ): """Should return error because there are too many localizations.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2(mocker) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( performer__count=1, writers__count=1) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) db.merge_model_objects([track]) localizations_data = { 'items': [ {'tuid': 1, 'language_id': 1, 'artists': {}}, {'tuid': 1, 'language_id': 2, 'artists': {}}, {'tuid': 1, 'language_id': 3, 'artists': {}}, {'tuid': 1, 'language_id': 4, 'artists': {}}, {'tuid': 1, 'language_id': 5, 'artists': {}}, {'tuid': 1, 'language_id': 8, 'artists': {}} ] } mocks.ows_product_get_track_localizations(mocker, localizations_data) mocks.ows_product_digital_get_product_by_id(mocker) mocker.patch.object( track_logic.track_validators, 'validate_track_audio_exists', return_value={}) localization_error = { 'message': error.LOCALIZATION_ERROR_TOO_MANY_LOCALIZATIONS, 'validator': 'required', 'validator_value': True, 'error_code': error.TOO_MANY_LOCALIZATIONS } expected_error = { 'errors': [{'localizations': [localization_error], 'tuid': 1}], 'total_tracks': 1, 'valid_tracks': 0, 'valid': False } url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body == expected_error @db.test_schema_no_seed def test_masters_rights_require_country_and_nationality( mocker, client, track_factory ): """Should return error because master rights require several fields. When Master Rights is chosen, Country of Recording and Nationality of First Producer fields are required. """ mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_product_by_id(mocker, 1, 123) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_product_by_id(mocker) track = track_factory.create( ownership_rights='original_owner', writers__count=1, performer__count=1) db.merge_model_objects([track]) recording_country_id_error = { 'recording_country_id': { 'message': 'Field `recording_country_id` is required', 'validator': 'required', 'validator_value': True, 'error_code': 'missing_recording_country_id' }, } original_rights_holder_country_id_error = { 'original_rights_holder_country_id': { 'message': 'Field `original_rights_holder_country_id` is required', 'validator': 'required', 'validator_value': True, 'error_code': 'missing_original_rights_holder_country_id' }, } expected_error = {'tuid': 1} expected_error.update(original_rights_holder_country_id_error) expected_error.update(recording_country_id_error) url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_error] @db.test_schema_no_seed def test_masters_rights_required( mocker, client, track_factory ): """Should return error because master rights are not chosen.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_product_by_id(mocker) track = track_factory.create( writers__count=1, performer__count=1, ownership_rights=None) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) db.merge_model_objects([track]) expected_error = { 'tuid': 1, 'ownership_rights': validation.format_missing_error( field_name='ownership_rights', error_code=error.MISSING_OWNERSHIP_RIGHTS ), } url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_error] @db.test_schema_no_seed def test_sountrack_composer_required( mocker, client, track_factory ): """Should return error because composer is required. When validating track of soundtrack genre, composer should exist. """ mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_product_by_id(mocker, subgenre_id=1040) track = track_factory.create( writers__count=1, performer__count=1, ownership_rights='no_rights') db.merge_model_objects([track]) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) expected_error = { 'tuid': 1, 'composer': validation.format_error( validator=error.REQUIRED_CODE, message=json.dumps( { 'genre': 'soundtrack', 'subgenre': 'Original Score' } ), error_code=error.MISSING_COMPOSER ), } url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_error] @db.test_schema_no_seed def test_preview_start_time_error_case( mocker, client, track_factory ): """Expect error because preview_start_time is larger than track length.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'duration': 60, 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_digital_get_product_by_id(mocker) track = track_factory.create( writers__count=1, performer__count=1, ownership_rights='no_rights', preview_start_time=61) db.merge_model_objects([track]) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) expected_error = { 'tuid': 1, 'preview_start_time': validation.format_error( validator='invalid_value', message=error.INVALID_PREVIEW_START_TIME_MSG, error_code=error.INVALID_PREVIEW_START_TIME ), } url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_error] def _generate_base_expected_track_validation_result( tuids=None, warnings=None ): generated_data = { 'total_tracks': 2, 'valid_tracks': 0, 'errors': [ { 'meta_language_code': { 'validator': 'required', 'validator_value': True, 'message': 'Field `meta_language_code` is required', 'error_code': 'missing_meta_language_code' }, 'isrc': { 'validator': 'isrc_pattern_mismatch', 'validator_value': True, 'message': 'This ISRC is invalid.', 'error_code': 'invalid_isrc' }, 'p_info': { 'validator': 'required', 'validator_value': True, 'message': 'Field `p_info` is required', 'error_code': 'missing_p_info' }, 'ownership_rights': { 'validator': 'required', 'validator_value': True, 'message': 'Field `ownership_rights` is required', 'error_code': 'missing_ownership_rights' }, 'performer': { 'validator': 'required', 'validator_value': True, 'message': 'Field `performer` is required', 'error_code': 'missing_performer' }, 'writers': { 'validator': 'required', 'validator_value': True, 'message': 'Field `writers` is required', 'error_code': 'missing_writers' }, 'audio_file': { 'validator': 'required', 'validator_value': True, 'message': 'Field `audio_file` is required', 'error_code': 'missing_audio_file' }, 'tuid': tuids and tuids[0] or 132321 }, { 'meta_language_code': { 'validator': 'required', 'validator_value': True, 'message': 'Field `meta_language_code` is required', 'error_code': 'missing_meta_language_code' }, 'isrc': { 'validator': 'isrc_pattern_mismatch', 'validator_value': True, 'message': 'This ISRC is invalid.', 'error_code': 'invalid_isrc' }, 'p_info': { 'validator': 'required', 'validator_value': True, 'message': 'Field `p_info` is required', 'error_code': 'missing_p_info' }, 'ownership_rights': { 'validator': 'required', 'validator_value': True, 'message': 'Field `ownership_rights` is required', 'error_code': 'missing_ownership_rights' }, 'performer': { 'validator': 'required', 'validator_value': True, 'message': 'Field `performer` is required', 'error_code': 'missing_performer' }, 'writers': { 'validator': 'required', 'validator_value': True, 'message': 'Field `writers` is required', 'error_code': 'missing_writers' }, 'audio_file': { 'validator': 'required', 'validator_value': True, 'message': 'Field `audio_file` is required', 'error_code': 'missing_audio_file' }, 'tuid': tuids and tuids[1] or 546456 } ], 'valid': False } if warnings: generated_data['warnings'] = warnings return generated_data @pytest.mark.parametrize( ( 'test_description', 'query_params', 'test_product_id', 'audio_match_status', 'audio_match_json', 'sound_recording_status', 'expected_track_validation_result_status', 'expected_track_validation_result_json', 'ows_sound_recordings_track_matches_exist' ), [ ( 'Default (pre_submission) validation', {}, 72, 200, None, 200, 200, _generate_base_expected_track_validation_result(), False ), ( 'pre_submission validation', {'validation_context': 'pre_submission'}, 72, 200, None, 200, 200, _generate_base_expected_track_validation_result(), False ), ( 'post_submission validation with all post submission warnings', {'validation_context': 'post_submission'}, 72, 200, { 'items': [ { 'asset_final_id': 1, 'code': 'WARNING_MATCHES_FOUND', 'matches': [{'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }], 'track_id': 132321 } ] }, 200, 200, _generate_base_expected_track_validation_result(warnings=( [ { 'isrc': { 'validator': 'max_isrc_reuse', 'validator_value': 5, 'message': json.dumps({ 'id': 132321, 'isrc': 'isrc2', 'isrc_usage': { 'digital_audio': { 'products': [ { 'id': 76, 'name': 'Product R', 'tracks': [ { 'id': 312323, 'name': 'Track 3' } ] } ], 'count': 1 }, 'physical': { 'products': [ { 'id': 101, 'name': 'Product V', 'tracks': [ { 'id': 123324, 'name': 'Track 4' } ] }, { 'id': 301, 'name': 'Product AA', 'tracks': [ { 'id': 123325, 'name': 'Track 55' } ] }, { 'id': 302, 'name': 'Product BB', 'tracks': [ { 'id': 123326, 'name': 'Track 66' } ] }, { 'id': 303, 'name': 'Product CC', 'tracks': [ { 'id': 123327, 'name': 'Track 77' } ] }, { 'id': 304, 'name': 'Product DD', 'tracks': [ { 'id': 123328, 'name': 'Track 88' } ] } ], 'count': 5 }, 'count': 6 } }), 'code': 'excessive_isrc_reuse' }, 'audio_file': { 'validator': 'potential_audio_infringement', 'validator_value': True, 'message': '{"asset_final_id": 1, "matches": [{"artists": ["Focus Sounds"], "title": "PN1", "label": "SME"}]}', # noqa: E501 'code': 'potential_audio_infringement' }, 'tuid': 132321 } ] )), False ), ( 'post_submission validation with no post submission warnings', {'validation_context': 'post_submission'}, 76, 200, { 'items': [ { 'asset_final_id': 1, 'code': 'PASS_NO_MATCHES', 'track_id': 132321 } ] }, 200, 200, _generate_base_expected_track_validation_result( tuids=[123322, 312323]), False ), ( 'post_submission validation with ows-assets 404 error', {'validation_context': 'post_submission'}, 76, 404, { 'code': 'not_found_error', 'message': 'final_assets_not_found' }, 200, 200, _generate_base_expected_track_validation_result( tuids=[123322, 312323]), False ), ( 'post_submission validation with orchard-sound-recording warnings', {'validation_context': 'post_submission'}, 76, 200, {'items': []}, 200, 200, _generate_base_expected_track_validation_result( tuids=[123322, 312323], warnings=[ { 'cross_account_osr_conflict': { 'code': 'cross_account_osr_conflict', 'message': ( '{"matches": [{"tuid": 99999, "product_id": 88888, ' '"isrc": "b", "vendor_id": 2, "release_status": "in_content", ' '"company_brand": "test_brand"}]}' ), 'validator': 'cross_account_osr_conflict', 'validator_value': True }, 'cross_isrc_osr_mismatch': { 'code': 'cross_isrc_osr_mismatch', 'message': ( '{"matches": [{"tuid": 11111, "product_id": 12345, ' '"isrc": "isrc1", "vendor_id": 2, "release_status": "in_content"' '}]}' ), 'validator': 'cross_isrc_osr_mismatch', 'validator_value': True }, 'cross_track_isrc_mismatch': { 'code': 'cross_track_isrc_mismatch', 'message': ( '{"matches": [{"tuid": 99999, "product_id": 88888, ' '"isrc": "b", "vendor_id": 2, "release_status": "in_content", ' '"company_brand": "test_brand"}]}' ), 'validator': 'cross_track_isrc_mismatch', 'validator_value': True }, 'tuid': 123322 } ] ), True ), ( 'post_submission validation with ows-sound-recordings error', {'validation_context': 'post_submission'}, 76, 200, {'items': []}, 504, 504, { 'code': 'ows_sound_recordings_error', 'message': ( 'Failed to get fingerprint matches for '\ '{\'product_id\': 76, \'include_transfer_to_content\': True}. '\ 'Error: "some error"' ) }, False ), ] ) @db.test_schema def test_validate_tracks_for_product_pre_and_post_submission( client, mocker, test_description, query_params, test_product_id, audio_match_status, audio_match_json, sound_recording_status, expected_track_validation_result_status, expected_track_validation_result_json, ows_sound_recordings_track_matches_exist ): """Test validate_tracks_for_product pre and post submission.""" with mysql.db_session() as session: session.execute(tests.conftest.CREATE_TABLE_RELEASES) session.execute(tests.conftest.INSERT_TEST_PRODUCTS) session.execute(tests.conftest.CREATE_TABLE_DISTRIBUTION_FORMAT) session.execute(tests.conftest.INSERT_TEST_DISTRIBUTION_FORMATS) session.execute(tests.conftest.INSERT_TEST_TRACKS) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response(message='test_brand') ) if ows_sound_recordings_track_matches_exist: sound_recording_json = [ mocks.ows_sound_recordings.mock_track_match( { 'tuid': 132321, 'product_id': test_product_id, 'isrc': 'a', 'vendor_id': 1, 'release_status': 'in_content' }, [{ 'tuid': 99999, 'product_id': 88888, 'isrc': 'b', 'vendor_id': 2, 'release_status': 'in_content' }, { 'tuid': 123322, 'product_id': test_product_id, 'isrc': 'isrc1', 'vendor_id': 1, 'release_status': 'transfer_to_content' }] ), mocks.ows_sound_recordings.mock_track_match( { 'tuid': 11111, 'product_id': 12345, 'isrc': 'isrc1', 'vendor_id': 2, 'release_status': 'in_content' }, None, '4218637a-30dc-4b34-b39d-a6001c474820' ) ] elif sound_recording_status == 200: sound_recording_json = [] else: sound_recording_json = 'some error' get_specs = [ { 'service': 'ows-product-digital', 'path': '/product/audio/{}'.format(test_product_id), 'status': 200, 'json': { 'genre_id': 123, 'subgenre_id': 456, }, }, { 'service': 'ows-assets', 'path': '/v2/asset/product/{}'.format(test_product_id), 'status': 200, 'json': { 'assets': [] }, }, { 'service': 'ows-product', 'path': '/localization/languages', 'status': 200, 'json': { 'items': [] }, }, { 'service': 'ows-product', 'path': '/product/72', 'status': 200, 'json': { 'vendor_id': 1 } }, { 'service': 'ows-assets', 'path': f'/v2/asset/product/{test_product_id}/match-audio', 'status': audio_match_status, 'json': audio_match_json }, { 'service': 'ows-sound-recordings', 'path': '/sound_recordings/cr', 'status': sound_recording_status, 'json': sound_recording_json }, ] post_specs = [ { 'service': 'ows-product', 'path': '/localization/track', 'status': 200, 'json': { 'items': [] }, } ] mocks.ows_product_get_product_by_id(mocker, 1, 123) mocker.patch.object( request, 'get', test_utils.mock_ows_requests(get_specs)) mocker.patch.object( request, 'post', test_utils.mock_ows_requests(post_specs)) track_validation_result = client.get( 'product/{}/tracks/validate'.format(test_product_id), query_string=query_params, headers={ 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': 123123123, }, ) assert track_validation_result.status_code == expected_track_validation_result_status assert track_validation_result.json == expected_track_validation_result_json @db.test_schema_no_seed def test_validate_tracks_for_product_missing_producer_success( client, mocker, client_headers, track_factory, ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_validate_missing_producer_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1, producer__count=1) db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_tracks_for_product_missing_producer_fail( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is failed.""" mocker.patch.object( features, 'is_validate_missing_producer_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1) db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker) mocks.ows_product_get_product_by_id(mocker, track.to_dict()['product_id'], 123) url = 'product/{}/tracks/validate'.format(track.to_dict()['product_id']) res = client.get(url, headers=client_headers) missing_producer_expected_response = { 'errors': [{'producer': { 'error_code': 'missing_producers', 'message': 'Field `producer` is required', 'validator': 'required', 'validator_value': True}, 'tuid': 1}], 'total_tracks': 1, 'valid': False, 'valid_tracks': 0} res_body = json.loads(res.data.decode()) assert res_body == missing_producer_expected_response assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_single_track_mismatches_with_mismatches( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1, featuring__count=1, version="remix") product_artists = [{'role': 'featuring', 'name': 'test name 2', 'id': 123, 'artist_info_id': None}, {'role': 'primary_artist', 'name': 'test name 3', 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_artists=product_artists) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'track_name': validation.format_error( validator='stNameAndProductNameMismatch', message=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME_MSG, error_code=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME ), 'version': validation.format_error( validator='stVersionAndProductVersionMismatch', message=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION_MSG, error_code=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION ), 'performer': validation.format_error( validator='stPrimaryArtistsAndProductPrimaryArtistsMismatch', message=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS_MSG, error_code=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS ), 'featuring': validation.format_error( validator='stFeaturingArtistsAndProductFeaturingArtistsMismatch', message=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS_MSG, error_code=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_single_track_validations_for_classical_product( client, mocker, client_headers, track_factory ): """Test validate single track by classical product""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocker.patch.object( features, 'is_single_track_composer_mismatch_enabled', return_value=True ) mocker.patch.object( features, 'is_single_track_classical_artists_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1, featuring__count=1, composer__count=2, version="remix") product_artists = [{'role': 'featuring', 'name': 'test name 2', 'id': 123, 'artist_info_id': None}, {'role': 'primary_artist', 'name': 'test name 3', 'id': 124, 'artist_info_id': None}, {'role': 'composer', 'name': 'test name 4', 'id': 125, 'artist_info_id': None}, {'role': 'orchestra', 'name': 'test name 5', 'id': 126, 'artist_info_id': None}, {'role': 'conductor', 'name': 'test name 6', 'id': 127, 'artist_info_id': None}, {'role': 'ensemble', 'name': 'test name 7', 'id': 128, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_id=track.product_id, product_artists=product_artists, genre_id=test_consts.TEST_CLASSICAL_GENRE_ID, subgenre_id=test_consts.TEST_CLASSICAL_SUBGENRE_ID) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'version': validation.format_error( validator='stVersionAndProductVersionMismatch', message=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION_MSG, error_code=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION ), 'performer': validation.format_error( validator='stPrimaryArtistsAndProductPrimaryArtistsMismatch', message=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS_MSG, error_code=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS ), 'featuring': validation.format_error( validator='stFeaturingArtistsAndProductFeaturingArtistsMismatch', message=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS_MSG, error_code=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS ), 'composer': validation.format_error( validator='stComposerMismatch', message=error.SINGLE_TRACK_COMPOSERS_MUST_MATCH_PRODUCT_COMPOSERS_MSG, error_code=error.SINGLE_TRACK_COMPOSERS_MUST_MATCH_PRODUCT_COMPOSERS ), 'orchestra': validation.format_error( validator='stOrchestraMismatch', message=error.SINGLE_TRACK_ORCHESTRAS_MUST_MATCH_PRODUCT_ORCHESTRAS_MSG, error_code=error.SINGLE_TRACK_ORCHESTRAS_MUST_MATCH_PRODUCT_ORCHESTRAS ), 'conductor': validation.format_error( validator='stConductorMismatch', message=error.SINGLE_TRACK_CONDUCTORS_MUST_MATCH_PRODUCT_CONDUCTORS_MSG, error_code=error.SINGLE_TRACK_CONDUCTORS_MUST_MATCH_PRODUCT_CONDUCTORS ), 'ensemble': validation.format_error( validator='stEnsembleMismatch', message=error.SINGLE_TRACK_ENSEMBLES_MUST_MATCH_PRODUCT_ENSEMBLES_MSG, error_code=error.SINGLE_TRACK_ENSEMBLES_MUST_MATCH_PRODUCT_ENSEMBLES ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_single_track_mismatches_with_match( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_artists=product_artists, product_name=track.track_name) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_single_track_primary_artist_mismatch_classical_corrections_valid( client, mocker, client_headers, track_factory, classical_genre_id, classical_subgenre_id ): """Test single track primary artist mismatch validation - classical with corrections.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1, composer__count=1) product_artists = [{'role': 'primary_artist', 'name': 'test name 3', 'id': 123, 'artist_info_id': None}, {'role': 'composer', 'name': 'test name 4', 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [ { 'table_name': 'releases', 'field_name': 'performer', 'key_id': 1, 'key_value': [ { 'role': 'performer', 'artist_name': 'correction test primary artist' } ] }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [ {'name': 'correction test primary artist', 'type': 'performer'}, {'name': 'correction test composer', 'type': 'composer'} ], 'table_name': 'track' }, { 'table_name': 'track', 'field_name': 'performers', 'key_id': track.tuid, 'key_value': [ { 'performer_id': 1, 'type': 'primary', 'performer_role_id': 17, 'birth_name': 'correction test primary artist', 'performer_role': 'Keyboards - Piano' } ] }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id, release_status=IS_ERR_CORR_MODE, product_artists=product_artists ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [] @db.test_schema_no_seed def test_single_track_artists_mismatch_classical_corrections_valid( client, mocker, client_headers, track_factory, classical_genre_id, classical_subgenre_id ): """Test single track artists mismatch validation - classical with corrections.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocker.patch.object( features, 'is_single_track_composer_mismatch_enabled', return_value=True ) mocker.patch.object( features, 'is_single_track_classical_artists_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, performer__count=1, composer__count=1, orchestra__count=1, conductor__count=1) product_artists = [{'role': 'primary_artist', 'name': 'test name 3', 'id': 123, 'artist_info_id': None}, {'role': 'composer', 'name': 'correction test composer', 'id': 124, 'artist_info_id': None}, {'role': 'orchestra', 'name': 'correction test orchestra', 'id': 125, 'artist_info_id': None}, {'role': 'conductor', 'name': 'test name 4', 'id': 126, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [ { 'table_name': 'releases', 'field_name': 'performer', 'key_id': 1, 'key_value': [ { 'role': 'performer', 'artist_name': 'correction test primary artist' } ] }, { "table_name": "releases", "field_name": "conductor", "key_id": 2, "key_value": [] }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [ {'name': 'correction test primary artist', 'type': 'performer'}, {'name': 'correction test composer', 'type': 'composer'}, {'name': 'correction test orchestra', 'type': 'orchestra'} ], 'table_name': 'track' }, { 'table_name': 'track', 'field_name': 'performers', 'key_id': track.tuid, 'key_value': [ { 'performer_id': 1, 'type': 'primary', 'performer_role_id': 17, 'birth_name': 'correction test primary artist', 'performer_role': 'Keyboards - Piano' } ] }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id, release_status=IS_ERR_CORR_MODE, product_artists=product_artists ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [] @db.test_schema_no_seed def test_validate_single_track_mismatches_with_mismatches_from_track_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': 'correction test name', 'table_name': 'track' }, { 'field_name': 'version', 'key_id': track.tuid, 'key_value': 'correction test version', 'table_name': 'track' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': 'correction test primary artist', 'type': 'performer'}], 'table_name': 'track' }, { 'field_name': 'featuring', 'key_id': track.tuid, 'key_value': [{'name': 'correction test featuring artist', 'role': 'featuring'}], 'table_name': 'track' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'track_name': validation.format_error( validator='stNameAndProductNameMismatch', message=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME_MSG, error_code=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME ), 'version': validation.format_error( validator='stVersionAndProductVersionMismatch', message=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION_MSG, error_code=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION ), 'performer': validation.format_error( validator='stPrimaryArtistsAndProductPrimaryArtistsMismatch', message=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS_MSG, error_code=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS ), 'featuring': validation.format_error( validator='stFeaturingArtistsAndProductFeaturingArtistsMismatch', message=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS_MSG, error_code=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_single_track_mismatches_with_matches_from_track_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': track.track_name, 'table_name': 'track' }, { 'field_name': 'version', 'key_id': track.tuid, 'key_value': track.version, 'table_name': 'track' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': track.artists[0].name, 'type': 'performer'}], 'table_name': 'track' }, { 'field_name': 'featuring', 'key_id': track.tuid, 'key_value': [{'name': track.artists[1].name, 'type': 'featuring'}], 'table_name': 'track' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_single_track_mismatches_with_mismatches_from_product_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'release_name', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test name', 'table_name': 'releases' }, { 'field_name': 'delivered_version', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test version', 'table_name': 'releases' }, { 'field_name': 'performer', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': 'correction test primary artist', 'role': 'performer'}], 'table_name': 'releases' }, { 'field_name': 'featuring', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': 'correction test featuring artist', 'role': 'featuring'}], 'table_name': 'releases' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'track_name': validation.format_error( validator='stNameAndProductNameMismatch', message=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME_MSG, error_code=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME ), 'version': validation.format_error( validator='stVersionAndProductVersionMismatch', message=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION_MSG, error_code=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION ), 'performer': validation.format_error( validator='stPrimaryArtistsAndProductPrimaryArtistsMismatch', message=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS_MSG, error_code=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS ), 'featuring': validation.format_error( validator='stFeaturingArtistsAndProductFeaturingArtistsMismatch', message=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS_MSG, error_code=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_single_track_mismatches_with_matches_from_product_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'release_name', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': track.track_name, 'table_name': 'releases' }, { 'field_name': 'delivered_version', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': '', 'table_name': 'releases' }, { 'field_name': 'performer', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': track.artists[0].name, 'role': 'performer'}], 'table_name': 'releases' }, { 'field_name': 'featuring', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': track.artists[1].name, 'role': 'featuring'}], 'table_name': 'releases' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_single_track_mismatches_with_matches_from_track_and_product_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': 'correction test name', 'table_name': 'track' }, { 'field_name': 'version', 'key_id': track.tuid, 'key_value': 'correction test version', 'table_name': 'track' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': track.artists[0].name, 'type': 'performer'}], 'table_name': 'track' }, { 'field_name': 'featuring', 'key_id': track.tuid, 'key_value': [{'name': track.artists[1].name, 'type': 'featuring'}], 'table_name': 'track' }, { 'field_name': 'release_name', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test name', 'table_name': 'releases' }, { 'field_name': 'delivered_version', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test version', 'table_name': 'releases' }, { 'field_name': 'performer', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': track.artists[0].name, 'role': 'performer'}], 'table_name': 'releases' }, { 'field_name': 'featuring', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': track.artists[1].name, 'role': 'featuring'}], 'table_name': 'releases' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_single_track_mismatches_with_matches_from_track_and_product_corrections_with_no_featuring_artist( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': 'correction test name', 'table_name': 'track' }, { 'field_name': 'version', 'key_id': track.tuid, 'key_value': 'correction test version', 'table_name': 'track' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': track.artists[0].name, 'type': 'performer'}], 'table_name': 'track' }, { 'field_name': 'featuring', 'key_id': track.tuid, 'key_value': [], 'table_name': 'track' }, { 'field_name': 'release_name', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test name', 'table_name': 'releases' }, { 'field_name': 'delivered_version', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction test version', 'table_name': 'releases' }, { 'field_name': 'performer', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': track.artists[0].name, 'role': 'performer'}], 'table_name': 'releases' }, { 'field_name': 'featuring', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [], 'table_name': 'releases' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body == {'errors': [], 'total_tracks': 1, 'valid_tracks': 1, 'valid': True} assert res.status_code == response_code.OK @db.test_schema_no_seed def test_validate_single_track_mismatches_with_mismatches_from_track_and_product_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocker.patch.object( features, 'is_single_track_and_product_mismatch_enabled', return_value=True ) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}, {'role': 'featuring', 'name': track.artists[1].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': 'correction test name', 'table_name': 'track' }, { 'field_name': 'version', 'key_id': track.tuid, 'key_value': 'correction test version', 'table_name': 'track' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': track.artists[0].name, 'type': 'performer'}], 'table_name': 'track' }, { 'field_name': 'featuring', 'key_id': track.tuid, 'key_value': [{'name': track.artists[1].name, 'type': 'featuring'}], 'table_name': 'track' }, { 'field_name': 'release_name', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'mismatch product name', 'table_name': 'releases' }, { 'field_name': 'delivered_version', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'mismatch product version', 'table_name': 'releases' }, { 'field_name': 'performer', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': 'mismatch primary artist', 'role': 'performer'}], 'table_name': 'releases' }, { 'field_name': 'featuring', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': [{'name': 'mismatch featuring artist', 'role': 'featuring'}], 'table_name': 'releases' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'track_name': validation.format_error( validator='stNameAndProductNameMismatch', message=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME_MSG, error_code=error.SINGLE_TRACK_NAME_MUST_MATCH_PRODUCT_NAME ), 'version': validation.format_error( validator='stVersionAndProductVersionMismatch', message=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION_MSG, error_code=error.SINGLE_TRACK_VERSION_MUST_MATCH_PRODUCT_VERSION ), 'performer': validation.format_error( validator='stPrimaryArtistsAndProductPrimaryArtistsMismatch', message=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS_MSG, error_code=error.SINGLE_TRACK_PRIMARY_ARTISTS_MUST_MATCH_PRODUCT_PRIMARY_ARTISTS ), 'featuring': validation.format_error( validator='stFeaturingArtistsAndProductFeaturingArtistsMismatch', message=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS_MSG, error_code=error.SINGLE_TRACK_FEATURING_ARTISTS_MUST_MATCH_PRODUCT_FEATURING_ARTISTS ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_tracks_with_missing_primary_artist_from_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': 2, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) track2 = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track, track2]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [], 'table_name': 'track' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'performer': validation.format_error( validator='required', message='Field `performer` is required', error_code=error.MISSING_PERFORMER ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_classical_tracks_with_missing_primary_artist_from_corrections( client, mocker, client_headers, track_factory, classical_genre_id, classical_subgenre_id ): """Test validate tracks by product id is successful.""" mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': 1, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': 2, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1, composer__count=1) track2 = track_factory.create( writers__count=1, featuring__count=1, performer__count=1, composer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track, track2]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': track.artists[0].name, 'type': 'composer'}], 'table_name': 'track' }, ] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name, genre_id=classical_genre_id, subgenre_id=classical_subgenre_id ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_errors = { 'performer': validation.format_error( validator='required', message='Field `performer` is required', error_code=error.MISSING_PERFORMER ), 'tuid': 1 } url = 'product/{}/tracks/validate'.format(track.product_id) res = client.get(url, headers=client_headers) res_body = json.loads(res.data.decode()) assert res_body['errors'] == [expected_errors] @db.test_schema_no_seed def test_validate_tracks_with_PAI_and_self_match_filter( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocks.head_response(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': 2, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_assets_get_match_audio_results(mocker, [ { 'asset_final_id': 1, 'code': 'WARNING_MATCHES_FOUND', 'matches': [ {'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }, {'artists': [track.artists[0].name], 'title': track.track_name, 'label': '' } ], 'track_id': track.tuid } ] ) mock_matches = [ mocks.ows_sound_recordings.mock_track_match( { 'tuid': 132321, 'product_id': 1, 'isrc': 'a', 'vendor_id': 1, 'release_status': 'in_content' }, [{ 'tuid': 123322, 'product_id': track.tuid, 'isrc': 'isrc1', 'vendor_id': 1, 'release_status': 'transfer_to_content' }] ), mocks.ows_sound_recordings.mock_track_match( { 'tuid': 11111, 'product_id': 12345, 'isrc': 'isrc1', 'vendor_id': 2, 'release_status': 'in_content' }, None, '4218637a-30dc-4b34-b39d-a6001c474820' ) ] get_mock = [{ 'service': 'ows-sound-recordings', 'path': '/sound_recordings/cr', 'status':200, 'json': mock_matches }] mocker.patch.object( request, 'get', test_utils.mock_ows_requests(get_mock)) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response(message='test_brand') ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track2 = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track, track2]) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id( mocker, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_warnings = { 'audio_file': { 'code': 'potential_audio_infringement', 'message': '{"asset_final_id": 1, "matches": [{"artists": ["Focus Sounds"], ' '"title": "PN1", "label": "SME"}]}', 'validator': 'potential_audio_infringement', 'validator_value': True, }, 'tuid': 1, } track_validation_result = client.get( 'product/{}/tracks/validate'.format(track.product_id), query_string={'validation_context': 'post_submission'}, headers={ 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': 123123123, }, ) res_body = json.loads(track_validation_result.data.decode()) assert res_body['warnings'] == [expected_warnings] @db.test_schema_no_seed def test_validate_tracks_with_PAI_and_self_match_filter_and_corrections( client, mocker, client_headers, track_factory ): """Test validate tracks by product id is successful.""" mocks.head_response(mocker) track = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) mocks.ows_assets_get_product_assets_v2( mocker, [{'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': 2, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}], ) mocks.ows_assets_get_match_audio_results(mocker, [ { 'asset_final_id': 1, 'code': 'WARNING_MATCHES_FOUND', 'matches': [ {'artists': ['Focus Sounds'], 'title': 'PN1', 'label': 'SME' }, { 'artists': ['correction test artist'], 'title': 'correction test name', 'label': 'correction product imprint' } ], 'track_id': track.tuid } ] ) mock_matches = [ mocks.ows_sound_recordings.mock_track_match( { 'tuid': 132321, 'product_id': 1, 'isrc': 'a', 'vendor_id': 1, 'release_status': 'in_content' }, [{ 'tuid': 123322, 'product_id': track.tuid, 'isrc': 'isrc1', 'vendor_id': 1, 'release_status': 'transfer_to_content' }] ), mocks.ows_sound_recordings.mock_track_match( { 'tuid': 11111, 'product_id': 12345, 'isrc': 'isrc1', 'vendor_id': 2, 'release_status': 'in_content' }, None, '4218637a-30dc-4b34-b39d-a6001c474820' ) ] get_mock = [{ 'service': 'ows-sound-recordings', 'path': '/sound_recordings/cr', 'status':200, 'json': mock_matches }] mocker.patch.object( request, 'get', test_utils.mock_ows_requests(get_mock)) mocker.patch.object( ows_account, 'get_vendor_company_brand', return_value=response.Response(message='test_brand') ) mocks.ows_product_get_track_localizations(mocker, {'items': []}) mocks.ows_product_get_localization_languages(mocker) track2 = track_factory.create( writers__count=1, featuring__count=1, performer__count=1) product_artists = [{'role': 'primary_artist', 'name': track.artists[0].name, 'id': 123, 'artist_info_id': None}] db.merge_model_objects([track, track2]) mocks.ows_product_get_track_localizations(mocker) corrections = { 'items': [{ 'field_name': 'track_name', 'key_id': track.tuid, 'key_value': 'correction test name', 'table_name': 'track' }, { 'field_name': 'label', 'key_id': test_consts.TEST_PRODUCT_ID, 'key_value': 'correction product imprint', 'table_name': 'releases' }, { 'field_name': 'track_artist', 'key_id': track.tuid, 'key_value': [{'name': 'correction test artist', 'type': 'performer'}], 'table_name': 'track' },] } mocks.ows_product_digital_get_product_by_id( mocker, corrections=corrections, release_status=IS_ERR_CORR_MODE, product_artists=product_artists, product_name=track.track_name ) mocks.ows_product_get_product_by_id(mocker, track.product_id, 123) expected_warnings = { 'audio_file': { 'code': 'potential_audio_infringement', 'message': '{"asset_final_id": 1, "matches": [{"artists": ["Focus Sounds"], ' '"title": "PN1", "label": "SME"}]}', 'validator': 'potential_audio_infringement', 'validator_value': True, }, 'tuid': 1, } track_validation_result = client.get( 'product/{}/tracks/validate'.format(track.product_id), query_string={'validation_context': 'post_submission'}, headers={ 'Orchard-Profile-Type': 'ContentProfile', 'Orchard-Profile-Id': 123123123, }, ) res_body = json.loads(track_validation_result.data.decode()) assert res_body['warnings'] == [expected_warnings] @db.test_schema_no_seed def test_validate_atmos_audio_valid( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Atmos asset with encoding_completed and a valid spatial ISRC should pass validation.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = True track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) track_spatial.create_track_spatial({'track_id': track.tuid, 'isrc': 'USRC17607839'}) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'atmos'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert result.message['valid_tracks'] == 1 assert result.message['errors'] == [] @db.test_schema_no_seed def test_validate_atmos_audio_missing_spatial_isrc( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Atmos asset with encoding_completed but no spatial ISRC in DB should fail validation.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = True track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'atmos'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert result.message['valid_tracks'] == 0 validation_error = result.message['errors'][0] assert 'spatial_isrc' in validation_error assert validation_error['spatial_isrc']['error_code'] == error.MISSING_SPATIAL_ISRC @db.test_schema_no_seed def test_validate_atmos_audio_encoding_not_completed( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Atmos asset that has not finished encoding should fail validation.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = True track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': track.tuid, 'status': 'uploading', 'asset_upload_type': 'atmos'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert result.message['valid_tracks'] == 0 validation_error = result.message['errors'][0] assert 'atmos_file' in validation_error assert validation_error['atmos_file']['error_code'] == error.MISSING_ATMOS_AUDIO_FILE assert validation_error['atmos_file']['validator'] == 'encoding_not_completed' @db.test_schema_no_seed def test_validate_atmos_audio_unexpected_spatial_isrc( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Spatial ISRC in DB but no Atmos asset present should fail validation.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = True track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) track_spatial.create_track_spatial({'track_id': track.tuid, 'isrc': 'USRC17607839'}) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert result.message['valid_tracks'] == 0 validation_error = result.message['errors'][0] assert 'spatial_isrc' in validation_error assert validation_error['spatial_isrc']['error_code'] == error.UNEXPECTED_SPATIAL_ISRC assert validation_error['spatial_isrc']['validator'] == 'unexpected_spatial_isrc' @db.test_schema_no_seed def test_validate_atmos_audio_invalid_spatial_isrc( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Atmos asset with encoding_completed but invalid spatial ISRC format should fail.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = True track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) track_spatial.create_track_spatial({'track_id': track.tuid, 'isrc': 'NOT-AN-ISRC'}) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'atmos'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) assert result.message['valid_tracks'] == 0 validation_error = result.message['errors'][0] assert 'spatial_isrc' in validation_error assert validation_error['spatial_isrc']['error_code'] == error.INVALID_ISRC assert validation_error['spatial_isrc']['validator'] == 'isrc_pattern_mismatch' @db.test_schema_no_seed def test_validate_atmos_audio_skipped_when_flag_disabled( mocker, logic_headers, logic_user, track_factory, test_product_id, mock_app, feature_flags, ): """Atmos/spatial ISRC validation should be skipped when the feature flag is off.""" feature_flags[feature_flag.CDAM_SPATIAL_ISRC_VALIDATION] = False track = track_factory(writers__count=1, performer__count=1) db.merge_model_objects([track]) track_spatial.create_track_spatial({'track_id': track.tuid, 'isrc': 'NOT-AN-ISRC'}) mocks.head_response(mocker) mocks.ows_assets_get_product_assets_v2( mocker, [ {'track_unique_id': track.tuid, 'status': 'encoding_completed', 'asset_upload_type': 'stereo'}, {'track_unique_id': track.tuid, 'status': 'uploading', 'asset_upload_type': 'atmos'}, ], ) mocks.ows_product_get_product_by_id(mocker, test_product_id, 123) mocks.ows_product_get_localization_languages(mocker) mocks.ows_product_get_track_localizations(mocker) mocks.ows_product_digital_get_product_by_id(mocker, product_id=test_product_id) mock_spatial_lookup = mocker.patch.object( track_logic.track_spatial_model, 'get_spatial_isrc_map_by_track_ids') result = track_logic.validate_tracks_for_product( product_id=test_product_id, **ChainMap(logic_headers, logic_user)) mock_spatial_lookup.assert_not_called() assert result.message['valid_tracks'] == 1 assert result.message['errors'] == []