"""Tests for Handlers.""" import datetime import json from unittest.mock import MagicMock from unittest.mock import PropertyMock from unittest.mock import call from unittest.mock import patch from uuid import UUID from connector_neo4j import exceptions as neo4j_connector_exceptions import neo4j from owsresponse import response import pytest from sound_recordings import handlers from sound_recordings.logic import acrids from sound_recordings.logic import orchard_sound_recordings from sound_recordings.models.ows_track import BadGateway as OwsTrackBadGateway from sound_recordings.validation import constants as validation_constants from sound_recordings.validation.schemas import FingerprintRuleSchema BODY_500 = { 'code': 'internal_error', 'message': 'The server encountered an internal error and was unable to complete your request.' # noqa:E501 } @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.acrids.upsert') def test_neo4j_database_error_exception(mock_logic, mock_g, app_client): """Test session expired neo4j error handling.""" exc = neo4j.exceptions.DatabaseError() mock_logic.side_effect = exc url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == 502 mock_g.log.exception.assert_called_with(exc) assert response.json == { 'code': 'database_error', 'message': 'neo4j database error' } @pytest.mark.parametrize( 'code, expected_http_code', [ ('Neo.TransientError.Transaction.Terminated', 502), ('Neo.TransientError.Transaction.LockClientStopped', 502), ('Neo.TransientError.Transaction.InstanceStateChanged', 502) ] ) @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.acrids.upsert') def test_neo4j_transient_exception( mock_logic, mock_g, app_client, code, expected_http_code): """Test transient neo4j error handling.""" exc = neo4j.exceptions.TransientError() exc.code = code exc.message = 'error-msg' mock_logic.side_effect = exc url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == expected_http_code if response.status_code != 500: mock_g.log.exception.assert_called_with(exc) assert response.json == { 'code': exc.code, 'message': exc.message } else: assert response.json == BODY_500 @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.acrids.upsert') def test_neo4j_session_expired_exception(mock_logic, mock_g, app_client): """Test session expired neo4j error handling.""" exc = neo4j.exceptions.SessionExpired() mock_logic.side_effect = exc url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == 502 mock_g.log.exception.assert_called_with(exc) assert response.json == { 'code': 'session_expired', 'message': 'neo4j session error' } @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.acrids.upsert') def test_neo4j_service_unavailable_exception(mock_logic, mock_g, app_client): """Test session expired neo4j error handling.""" exc = neo4j.exceptions.ServiceUnavailable() mock_logic.side_effect = exc url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == 502 mock_g.log.exception.assert_called_with(exc) assert response.json == { 'code': 'service_unavailable', 'message': 'neo4j service error' } @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.acrids.upsert') def test_neo4j_connector_session_not_created_exception(mock_logic, mock_g, app_client): """Test session not created neo4j error handling.""" exc = neo4j_connector_exceptions.SessionNotCreated() mock_logic.side_effect = exc url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == 502 mock_g.log.exception.assert_called_with(exc) assert response.json == { 'code': 'session_not_created', 'message': 'neo4j session error' } @patch('sound_recordings.handlers.g', spec=['log']) @patch('sound_recordings.logic.orchard_sound_recordings.upsert') def test_ows_track_exception(mock_logic, mock_g, app_client): """Test bad gateway ows-track handling.""" url = '/sound_recordings' input_data = { 'track_id': 123 } exc = OwsTrackBadGateway() mock_logic.side_effect = exc response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == 502 mock_g.log.exception.assert_called_with(exc) @pytest.mark.parametrize( 'input_data, output_message', [ # empty input ( {}, { 'acr_id': ['Missing data for required field.'], 'asset_id': ['Missing data for required field.'] } ), # bad asset id UUID ( { 'acr_id': 'acr123', 'asset_id': 'invalid-uuid' }, { 'asset_id': ['Not a valid UUID.'] } ) ] ) @patch('sound_recordings.logic.acrids.upsert') def test_post_acrids_validation( mock_logic, app_client, input_data, output_message): """Test validation exceptions for acrid upsert.""" url = '/acrids' response = app_client.post(url, json=input_data) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called @pytest.mark.parametrize( 'exception, output_code, output_body', [ (acrids.AlreadyFingerprinted, 500, BODY_500), (acrids.AssetNotFound, 500, BODY_500) ] ) @patch('sound_recordings.logic.acrids.upsert') def test_post_acrids_exception( mock_logic, app_client, exception, output_code, output_body): """Test acrid exception cases.""" mock_logic.side_effect = exception url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == output_code assert response.json == output_body @pytest.mark.parametrize( 'updated, output_code', [ (False, 200), (True, 201) ] ) @patch('sound_recordings.logic.acrids.upsert') def test_post_acrids_success( mock_logic, app_client, updated, output_code): """Test acrid upsert sucess cases.""" mock_logic.return_value = ([123], updated) url = '/acrids' input_data = { 'acr_id': 'acr123', 'asset_id': '0b43df7e-8ba2-4a59-8871-7762b2d33e1f' } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == output_code assert response.json == {'track_ids': [123]} @pytest.mark.parametrize( 'input_data, output_message', [ # empty input ( {}, { 'track_id': ['Missing data for required field.'] } ), # bad asset id UUID ( { 'track_id': 'invalid', }, { 'track_id': ['Not a valid integer.'] } ) ] ) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('sound_recordings.logic.orchard_sound_recordings.upsert') def test_post_sound_recordings_validation( mock_logic, mock_neo4j_session_enter, app_client, input_data, output_message): """Test validation exceptions for sound recording upsert.""" url = '/sound_recordings' response = app_client.post(url, json=input_data) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_neo4j_session_enter.called assert not mock_logic.called @pytest.mark.parametrize( 'exception, output_code, output_body', [ (orchard_sound_recordings.AssetNotFound, 409, {'error': 'AssetNotFound'}), # noqa:E501 (orchard_sound_recordings.TrackISRCNotSet, 409, {'error': 'TrackISRCNotSet'}), # noqa:E501 (orchard_sound_recordings.MultipleAcrOsrRelationships, 409, {'error': 'MultipleAcrOsrRelationships'}) # noqa:E501 ] ) @patch('sound_recordings.logic.orchard_sound_recordings.upsert') def test_post_osr_exception( mock_logic, app_client, exception, output_code, output_body): """Test acrid exception cases.""" mock_logic.side_effect = exception url = '/sound_recordings' input_data = { 'track_id': 123 } response = app_client.post(url, json=input_data) assert mock_logic.called assert response.status_code == output_code assert response.json == output_body @pytest.mark.parametrize( 'updates, status_code', [ (True, 201), (False, 200) ] ) @patch('sound_recordings.logic.orchard_sound_recordings.upsert') def test_post_sound_recordings_success( mock_logic, app_client, updates, status_code): """Test sound recording upsert success cases.""" url = '/sound_recordings' input_data = { 'track_id': 123 } mock_logic.return_value = ( '1cf25c14-f3c5-4bf5-b8df-9096bbe0fc74', updates ) response = app_client.post(url, json=input_data) assert response.status_code == status_code assert mock_logic.called assert mock_logic.call_args_list == [ call( 123 ) ] assert response.json == { 'orchard_sound_recording_id': '1cf25c14-f3c5-4bf5-b8df-9096bbe0fc74' } @pytest.mark.parametrize( 'input_data, output_message', [ # empty input ( {}, [ 'At least one field required from ids,' 'tuids,upcs,product_ids,project_ids,asset_ids,track_isrcs,term' ] ), # bad tuid ( { 'ids': '1cf25c14-f3c5-4bf5-b8df-9096bbe0fc74,invalid-uuid' }, { 'ids': { '1': ['Not a valid UUID.'] } } ), # bad id ( { 'tuids': '1234,not-and-int' }, { 'tuids': { '1': ['Not a valid integer.'] } } ) ] ) @patch('connector_neo4j.Neo4jSession.__enter__') @patch('sound_recordings.logic.orchard_sound_recordings.fetch') def test_get_sound_recordings_validation( mock_logic, mock_neo4j_session_enter, app_client, input_data, output_message): """Test validation exceptions for sound recording fetch.""" url = '/sound_recordings' response = app_client.get(url, query_string=input_data) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['query'] == output_message assert not mock_neo4j_session_enter.called assert not mock_logic.called @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.orchard_sound_recordings.fetch') def test_get_sound_recordings_success( mock_logic, mock_neo4j_driver, app_client): """Test sound recording fetch success.""" url = '/sound_recordings' input_data = { 'ids': '4984e2a7-ffbc-4a44-8a14-6f5151ba409c,1a345a57-e51d-4d54-8073-4fe7c2e66249', # noqa:E501 'tuids': '123,456' } mock_logic.return_value = [{'test': '1'}] response = app_client.get(url, query_string=input_data) assert response.status_code == 200 assert mock_neo4j_driver.session.called assert mock_logic.called assert mock_logic.call_args_list == [ call( orchard_sound_recording_ids=[ UUID('4984e2a7-ffbc-4a44-8a14-6f5151ba409c'), UUID('1a345a57-e51d-4d54-8073-4fe7c2e66249') ], track_ids=[123, 456], upcs=[], product_ids=[], project_ids=[], asset_ids=[], track_isrcs=[], include_deleted=False, include_transfer_to_content=False, term='', include_inactive=False ) ] assert response.json == [{'test': '1'}] @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.orchard_sound_recordings.fetch') def test_get_sound_recordings_by_project_ids(mock_logic, mock_neo4j_driver, app_client): """Test sound recording fetch by project_ids.""" url = '/sound_recordings' input_data = {'project_ids': '10,20'} mock_logic.return_value = [{'test': '1'}] response = app_client.get(url, query_string=input_data) assert response.status_code == 200 assert mock_logic.call_args_list == [ call( orchard_sound_recording_ids=[], track_ids=[], upcs=[], product_ids=[], project_ids=[10, 20], asset_ids=[], track_isrcs=[], include_deleted=False, include_transfer_to_content=False, term='', include_inactive=False ) ] assert response.json == [{'test': '1'}] @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.orchard_sound_recordings.fetch_product') def test_get_sound_recordings_product(mock_logic, mock_neo4j_driver, app_client): # noqa:E501 """Test sound recording fetch product success.""" url = '/sound_recordings/cr' input_data = { 'product_id': 444, 'include_transfer_to_content': True } mock_logic.return_value = [{'test': '1'}] response = app_client.get(url, query_string=input_data) assert response.status_code == 200 assert mock_neo4j_driver.session.called assert mock_logic.called assert mock_logic.call_args_list == [call(444, [], True)] @patch('sound_recordings.logic.orchard_sound_recordings.fetch_version') def test_get_sound_recording_version(mock_logic, app_client): """Test sound recording fetch version success.""" url = '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c/versions/VERSION_ID' mock_logic.return_value = {'test': '1'} response = app_client.get(url) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call('4984e2a7-ffbc-4a44-8a14-6f5151ba409c', 'VERSION_ID') ] assert response.json == {'test': '1'} @patch('sound_recordings.logic.orchard_sound_recordings.fetch_version') def test_get_sound_recording_version_not_found(mock_logic, app_client): """Test sound recording fetch version not found.""" url = '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c/versions/VERSION_ID' mock_logic.return_value = None response = app_client.get(url) assert response.status_code == 404 assert mock_logic.called assert mock_logic.call_args_list == [ call('4984e2a7-ffbc-4a44-8a14-6f5151ba409c', 'VERSION_ID') ] assert response.json == { 'code': 'not_found_error', 'message': 'Orchard Sound Recording not found' } @patch('sound_recordings.logic.orchard_sound_recordings.fetch_version') def test_get_sound_recording_version_not_uuid(mock_logic, app_client): """Test sound recording fetch version with non uuid param.""" url = '/sound_recordings/499/versions/VERSION_ID' mock_logic.return_value = None response = app_client.get(url) assert response.status_code == 422 assert not mock_logic.called assert response.json == { 'code': 'bad_request', 'message': { 'view_args': { 'osr_id': [ 'Not a valid UUID.' ] } } } @patch('sound_recordings.logic.orchard_sound_recordings.fetch_full_delivery_history') def test_get_sound_recordings_full_delivery_history(mock_logic, app_client): """Test sound recording fetch full history success, including ack fields.""" url = '/sound_recording/4984e2a7-ffbc-4a44-8a14-6f5151ba409c/full_delivery_history' mock_logic.return_value = [ { 'sound_recording_id': '4984e2a7-ffbc-4a44-8a14-6f5151ba409c', 'service': 'TikTok (Audio Fingerprinting)', 'event_type': 'success', 'ack': 'success', 'ack_message': '{"ack": "success"}' }, { 'sound_recording_id': '4984e2a7-ffbc-4a44-8a14-6f5151ba409c', 'service': 'Meta (Audio Fingerprinting)', 'event_type': 'success', 'ack': 'missing', 'ack_message': None } ] response = app_client.get(url) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call(['4984e2a7-ffbc-4a44-8a14-6f5151ba409c'], [], [], [], []) ] assert response.json == mock_logic.return_value assert response.json[0]['ack'] == 'success' assert response.json[0]['ack_message'] == '{"ack": "success"}' assert response.json[1]['ack'] == 'missing' assert response.json[1]['ack_message'] is None @patch('sound_recordings.logic.orchard_sound_recordings.fetch_full_delivery_history') def test_get_sound_recordings_full_delivery_history_not_found(mock_logic, app_client): """Test sound recording fetch history not found.""" url = '/sound_recording/4984e2a7-ffbc-4a44-8a14-6f5151ba409c/full_delivery_history' mock_logic.return_value = [] response = app_client.get(url) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call(['4984e2a7-ffbc-4a44-8a14-6f5151ba409c'], [], [], [], []) ] assert response.json == [] @patch('sound_recordings.logic.orchard_sound_recordings.fetch_full_delivery_history') def test_get_sound_recordings_full_delivery_history_ack_status_filter( mock_logic, app_client): """Test ack_status query param is forwarded to the logic layer.""" url = ( '/sound_recording/4984e2a7-ffbc-4a44-8a14-6f5151ba409c/full_delivery_history' '?ack_status=success,awaited' ) mock_logic.return_value = [] response = app_client.get(url) assert response.status_code == 200 assert mock_logic.call_args_list == [ call(['4984e2a7-ffbc-4a44-8a14-6f5151ba409c'], [], [], [], ['success', 'awaited']) ] @patch('sound_recordings.logic.orchard_sound_recordings.fetch_full_delivery_history') def test_post_sound_recordings_full_delivery_history(mock_logic, app_client): """Test post sound recording fetch full history success, including ack fields.""" url = '/sound_recordings/full_delivery_history' mock_logic.return_value = [ { 'sound_recording_id': '4d827696-b591-4b22-b142-2258157c0be9', 'service': 'TikTok (Audio Fingerprinting)', 'event_type': 'success', 'ack': 'success', 'ack_message': '{"ack": "success"}' } ] mock_body = { 'osr_ids': ['4d827696-b591-4b22-b142-2258157c0be9'], 'service': 'TikTok (Audio Fingerprinting)', 'execution_type': 'METADATA_UPDATE', 'event_type': 'success' } response = app_client.post(url, json=mock_body) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call(['4d827696-b591-4b22-b142-2258157c0be9'], ['METADATA_UPDATE'], ['TikTok (Audio Fingerprinting)'], ['success'], [], None, None, None) ] assert response.json == mock_logic.return_value assert response.json[0]['ack'] == 'success' assert response.json[0]['ack_message'] == '{"ack": "success"}' @patch('sound_recordings.logic.orchard_sound_recordings.fetch_full_delivery_history') def test_post_sound_recordings_full_delivery_history_ack_status_filter( mock_logic, app_client): """Test ack_status body param is forwarded to the logic layer.""" url = '/sound_recordings/full_delivery_history' mock_logic.return_value = [] mock_body = { 'osr_ids': ['4d827696-b591-4b22-b142-2258157c0be9'], 'ack_status': 'error,missing' } response = app_client.post(url, json=mock_body) assert response.status_code == 200, response.json assert mock_logic.call_args_list == [ call(['4d827696-b591-4b22-b142-2258157c0be9'], [], [], [], ['error', 'missing'], None, None, None) ] @patch('sound_recordings.logic.orchard_sound_recordings.update') def test_patch_osr(mock_logic, app_client): """Test sound recording patch success.""" url = '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c' input_data = { 'primary_track_id': 123 } mock_logic.return_value = {'test': '1'} response = app_client.patch(url, json=input_data) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call('4984e2a7-ffbc-4a44-8a14-6f5151ba409c', 123) ] assert response.json == {'test': '1'} @pytest.mark.parametrize( 'exception, output_code, output_body', [ (orchard_sound_recordings.OrchardSoundRecordingNotFound, 404, {'error': 'OrchardSoundRecordingNotFound'}), # noqa:E501 (orchard_sound_recordings.OrchardSoundRecordingWithoutAssets, 409, {'error': 'OrchardSoundRecordingWithoutAssets'}), # noqa:E501 (orchard_sound_recordings.PrimaryTrackNotRelated, 409, {'error': 'PrimaryTrackNotRelated'}) # noqa:E501 ] ) @patch('sound_recordings.logic.orchard_sound_recordings.update') def test_patch_osr_exception( mock_logic, app_client, exception, output_code, output_body): """Test osr update exception cases.""" mock_logic.side_effect = exception url = '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c' input_data = { 'primary_track_id': 123 } response = app_client.patch(url, json=input_data) assert mock_logic.called assert response.status_code == output_code assert response.json == output_body @pytest.mark.parametrize( 'url, input_data, output_message', [ # empty input ( '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c', {}, { 'primary_track_id': ['Missing data for required field.'] # noqa:E501 } ), # unknown field ( '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c', { 'something': 123, }, { 'primary_track_id': ['Missing data for required field.'], 'something': ['Unknown field.'] } ), # not valid type param ( '/sound_recordings/4984e2a7-ffbc-4a44-8a14-6f5151ba409c', { 'primary_track_id': 'something' }, { 'primary_track_id': ['Not a valid integer.'] } ) ] ) @patch('sound_recordings.logic.orchard_sound_recordings.update') def test_patch_sound_recordings_validation(mock_logic, app_client, url, input_data, output_message): # noqa:E501 """Test validation exceptions for sound recording patch.""" response = app_client.patch(url, json=input_data) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called @pytest.mark.parametrize( 'resource_type', [ ('vendor'), ('subaccount') ] ) @patch('sound_recordings.logic.orchard_sound_recordings.touch') def test_patch_touch(mock_logic, app_client, resource_type): """Test touch sound recording patch success.""" url = '/sound_recordings/touch' input_data = { 'resource_id': 123, 'resource_type': resource_type, 'modified_before': '2021-10-14T15:22:35.390000000Z', 'limit': 15 } mock_logic.return_value = {'nodes_updated': 1} response = app_client.patch(url, json=input_data) assert response.status_code == 200 assert mock_logic.called assert mock_logic.call_args_list == [ call(123, resource_type, datetime.datetime(2021, 10, 14, 15, 22, 35, 390000, tzinfo=datetime.timezone.utc), 15) # noqa:E501 ] assert response.json == {'nodes_updated': 1} @patch('connector_neo4j.Neo4jSession.__enter__') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_body_required( mock_logic, mock_neo4j_session_enter, app_client): """Test required fields for a rule are checked.""" url = '/tracks/12345/rules' output_message = { '0': { x: ['Missing data for required field.'] for x in ['end', 'start', 'policy', 'service', 'territory'] } } response = app_client.post(url, json=[{}]) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called assert not mock_neo4j_session_enter.called @pytest.mark.parametrize( 'input_data, output_message', [ # basic valid case ( [('US', 'monetize', 'meta', '2019-06-15T00:00:00+00:00', None)], None ), # non-allowed start ( [('US', 'carveout', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'start': ['cannot be set when policy == carveout']}} ), # not-allowed end ( [('US', 'carveout', 'meta', None, '2019-05-15T00:00:00+00:00')], {'0': {'end': ['cannot be set when policy == carveout']}} ), # start after end ( [('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2019-05-15T00:00:00+00:00')], # noqa:E501 {'0': {'end': ['cannot be before start'], 'start': ['cannot be after end']}} ), # start and end same ( [('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2019-05-15T00:00:00+00:00')], # noqa:E501 {'0': {'end': ['cannot equal start'], 'start': ['cannot equal end']}} ), # carveout policy mixed ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2020-05-15T00:00:00+00:00'), # noqa:E501 ('US', 'carveout', 'meta', None, None) ], {'1': {'policy': ['cannot mix carveout rules with other rules in meta:US']}} ), # more than one carveout ( [ ('US', 'carveout', 'meta', None, None), ('US', 'carveout', 'meta', None, None) ], { '0': {'policy': ['cannot set multiple carveout rules in meta:US']}, '1': {'policy': ['cannot set multiple carveout rules in meta:US']} } ), # bad timezone ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+05:00', '2020-05-15T00:00:00+05:00'), # noqa:E501 ], {'0': {'start': ['must be in UTC timezone'], 'end': ['must be in UTC timezone']}} # noqa:E501 ), # date overlap ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2022-06-15T00:00:00+00:00'), # noqa:E501 ('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2021-06-15T00:00:00+00:00') # noqa:E501 ], { '0': {'end': ['after rule 1 start in meta:US']}, '1': {'start': ['before rule 0 end in meta:US']} } ), # open end range in middle ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', None), ('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2021-06-15T00:00:00+00:00') # noqa:E501 ], {'0': {'end': ['cannot be empty for non-last rule in meta:US']}} ), # start date must be set ( [ ('US', 'monetize', 'meta', None, '2022-05-15T00:00:00+00:00') ], {'0': {'start': ['must be set when policy != carveout']}} ), # invalid policy ( [('US', 'block', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'policy': [f"Must be one of: {', '.join(validation_constants.RULE_POLICIES)}."]}} # noqa:E501 ), # invalid territory ( [('USA', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'territory': [f"Must be one of: {', '.join(validation_constants.RULE_TERRITORIES)}."]}} # noqa:E501 ), # invalid service ( [('US', 'monetize', 'facebook', '2019-05-15T00:00:00+00:00', None)], {'0': {'service': [f"Must be one of: {', '.join(validation_constants.RULE_SERVICES)}."]}} # noqa:E501 ), # sixtyseconds valid for youtube ( [('US', 'sixtyseconds', 'youtube', '2019-06-15T00:00:00+00:00', None)], None ), # sixtyseconds invalid for non-youtube service ( [('US', 'sixtyseconds', 'meta', '2019-06-15T00:00:00+00:00', None)], {'0': {'policy': ['sixtyseconds policy is only valid for the youtube service']}} ), # sixtyseconds invalid for wildcard service ( [('US', 'sixtyseconds', '*', '2019-06-15T00:00:00+00:00', None)], {'0': {'policy': ['sixtyseconds policy is only valid for the youtube service']}} ) ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_body_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, input_data, output_message): """Test validation exceptions for fingerprint rule upsert body.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 2) request_body = [] for data in input_data: rule = { 'territory': data[0], 'policy': data[1], 'service': data[2], 'start': data[3], 'end': data[4] } request_body.append(rule) url = '/tracks/12345/rules' response = app_client.post(url, json=request_body) if not output_message: assert mock_logic.called # format dates and add schema defaults to verify data passed to logic correctly parsed_request = [] format_dates = ['start', 'end'] for rule in request_body: for key in format_dates: if rule[key]: rule[key] = datetime.datetime.fromisoformat(rule[key]) rule.setdefault('is_start_absolute', False) rule.setdefault('is_end_absolute', False) parsed_request.append(rule) assert mock_logic.call_args_list == [ call('Track', 12345, parsed_request, ('ContentProfile', 123)) ] assert response.json == { 'num_created': 1, 'num_deleted': 2 } assert mock_neo4j_driver.session.called else: assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called assert not mock_neo4j_driver.session.called @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_is_start_and_end_absolute_default_to_false( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test is_start_absolute and is_end_absolute default to False when omitted.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (0, 0) request_body = [{ 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2019-06-15T00:00:00+00:00', 'end': None, }] response = app_client.post('/tracks/12345/rules', json=request_body) assert response.status_code == 200 expected_rule = { 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': datetime.datetime.fromisoformat('2019-06-15T00:00:00+00:00'), 'end': None, 'is_start_absolute': False, 'is_end_absolute': False, } assert mock_logic.call_args_list == [ call('Track', 12345, [expected_rule], ('ContentProfile', 123)) ] @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_is_start_and_end_absolute_true_valid_when_dates_set( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test is_start_absolute and is_end_absolute can be set to True when dates are present.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (0, 0) request_body = [{ 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2019-06-15T00:00:00+00:00', 'end': '2020-06-15T00:00:00+00:00', 'is_start_absolute': True, 'is_end_absolute': True, }] response = app_client.post('/tracks/12345/rules', json=request_body) assert response.status_code == 200 expected_rule = { 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': datetime.datetime.fromisoformat('2019-06-15T00:00:00+00:00'), 'end': datetime.datetime.fromisoformat('2020-06-15T00:00:00+00:00'), 'is_start_absolute': True, 'is_end_absolute': True, } assert mock_logic.call_args_list == [ call('Track', 12345, [expected_rule], ('ContentProfile', 123)) ] @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_is_start_absolute_true_invalid_when_start_is_null( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test is_start_absolute=True is invalid when start is null.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) request_body = [{ 'territory': 'US', 'policy': 'carveout', 'service': 'meta', 'start': None, 'end': None, 'is_start_absolute': True, }] response = app_client.post('/tracks/12345/rules', json=request_body) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == { '0': {'is_start_absolute': ['can only be true if start is not null']} } assert not mock_logic.called @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_is_end_absolute_true_invalid_when_end_is_null( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test is_end_absolute=True is invalid when end is null.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) request_body = [{ 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2019-06-15T00:00:00+00:00', 'end': None, 'is_end_absolute': True, }] response = app_client.post('/tracks/12345/rules', json=request_body) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == { '0': {'is_end_absolute': ['can only be true if end is not null']} } assert not mock_logic.called @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_is_start_and_end_absolute_default_to_false( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test is_start_absolute and is_end_absolute default to False on the bulk endpoint.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (0, 0) request_body = {'12345': [{ 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2019-06-15T00:00:00+00:00', 'end': None, }]} response = app_client.post('/tracks/bulk/rules', json=request_body) assert response.status_code == 200 expected_rule = { 'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': datetime.datetime.fromisoformat('2019-06-15T00:00:00+00:00'), 'end': None, 'is_start_absolute': False, 'is_end_absolute': False, } assert mock_logic.call_args_list == [ call('Track', 12345, new_rules=[expected_rule], profile=('ContentProfile', 123), affected_services=None) ] @patch( 'sound_recordings.validation.constants.RULE_SERVICES', new_callable=PropertyMock(return_value=('youtube', 'meta', '*')) ) @patch( 'sound_recordings.validation.constants.RULE_TERRITORIES', new_callable=PropertyMock(return_value=('US', 'UK', 'FR', '*')) ) def test_fingerprint_rules_grouping(mock_territories, mock_services): """Test creating groups to apply validation rules to.""" rules = [ { 'service': 'meta', 'territory': 'US', 'id': 1 }, { 'service': 'meta', 'territory': 'UK', 'id': 2 }, { 'service': '*', 'territory': 'UK', 'id': 3 }, # add to all UK groups { 'service': 'youtube', 'territory': 'US', 'id': 4 }, { 'service': 'youtube', 'territory': 'UK', 'id': 5 }, { 'service': 'youtube', 'territory': '*', 'id': 6 }, # add to all youtube groups { 'service': '*', 'territory': '*', 'id': 7 } # add to all groups ] results = FingerprintRuleSchema()._group_rules(rules) # use rule ids to make sure they were bucketed as expected assert { key: sorted([x['id'] for x in value]) for key, value in results.items() } == { 'meta:US': [1, 7], 'meta:UK': [2, 3, 7], 'meta:FR': [7], 'youtube:US': [4, 6, 7], 'youtube:UK': [3, 5, 6, 7], 'youtube:FR': [6, 7] } @pytest.mark.parametrize( 'url, output_message', [ ('/tracks/12345/rules', None), ('/track/12345/rules', {'obj_type': [f"Must be one of: {', '.join(validation_constants.RULE_OBJECT_SCOPES)}."]}), # noqa:E501 ('/tracks/abc/rules', {'obj_id': ['Not a valid integer.']}) ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_url_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, url, output_message): """Test validation exceptions for fingerprint rule upsert url.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) response = app_client.post(url, json=[]) if not output_message: assert mock_logic.called assert mock_neo4j_driver.session.called else: assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['view_args'] == output_message assert not mock_logic.called assert not mock_neo4j_driver.session.called @pytest.mark.parametrize( 'obj_type, url, input_data, output_message', [ ( 'Track', '/tracks/rules', { 'ids': 12345 }, None ), ( 'Track', '/tracks/rules', { 'ids': 'abc' }, ['Not a valid integer.'] ), ( 'Vendor', '/vendors/rules', { 'ids': 12345 }, None ), ( 'Vendor', '/vendors/rules', { 'ids': 'abc' }, ['Not a valid integer.'] ), ( 'Subaccount', '/subaccounts/rules', { 'ids': 12345 }, None ), ( 'Subaccount', '/subaccounts/rules', { 'ids': 'abc' }, ['Not a valid integer.'] ) ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.match_bulk') def test_get_bulk_fingerprint_rules_url_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, obj_type, url, input_data, output_message): """Test validation exceptions for fingerprint rule match url.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) response = app_client.get(url, query_string=input_data) if not output_message: assert mock_logic.call_args_list == [ call(obj_type, [12345]) ] assert mock_neo4j_driver.session.called else: assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['query']['ids']['0'] == output_message assert not mock_logic.called assert not mock_neo4j_driver.session.called @pytest.mark.parametrize( 'obj_type, url, input_data, profile, has_access', [ ( 'Track', '/tracks/rules', { 'ids': 12345 }, ('ContentProfile', 123), True ), ( 'Track', '/tracks/rules', { 'ids': 12345 }, ('ContentProfile', 456), False ), ( 'Vendor', '/vendors/rules', { 'ids': 12345 }, ('ContentProfile', 123), True ), ( 'Vendor', '/vendors/rules', { 'ids': 12345 }, ('ContentProfile', 456), False ), ( 'Subaccount', '/subaccounts/rules', { 'ids': 12345 }, ('ContentProfile', 123), True ), ( 'Subaccount', '/subaccounts/rules', { 'ids': 12345 }, ('ContentProfile', 456), False ) ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.match_bulk') def test_get_bulk_fingerprint_rules_profile_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, obj_type, url, input_data, profile, has_access): """Test profile access validation for bulk fingerprint rule match url.""" mock_permissions.return_value = (has_access, profile) response = app_client.get(url, query_string=input_data) if has_access: assert mock_logic.call_args_list == [ call(obj_type, [12345]) ] assert mock_neo4j_driver.session.called else: assert response.status_code == 403 assert response.json['error'] == 'NotAuthorizedOperation' assert not mock_logic.called @pytest.mark.parametrize( 'profile, has_access', [ (('ContentProfile', 123), True), (('ContentProfile', 456), False), ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_rules_profile_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, profile, has_access): """Test profile access validation for fingerprint rule upsert url.""" mock_logic.return_value = (1, 2) mock_permissions.return_value = (has_access, profile) request_body = [{ 'territory': 'US', 'policy': 'carveout', 'service': 'meta', 'start': None, 'end': None }] response = app_client.post('/tracks/12345/rules', json=request_body) if has_access: expected_body = [{**request_body[0], 'is_start_absolute': False, 'is_end_absolute': False}] assert mock_logic.call_args_list == [ call('Track', 12345, expected_body, ('ContentProfile', 123)) ] assert mock_neo4j_driver.session.called else: assert response.status_code == 403 assert response.json['error'] == 'NotAuthorizedOperation' assert not mock_logic.called @pytest.mark.parametrize( 'profile, has_access', [ (('ContentProfile', 123), True), (('ContentProfile', 456), False), ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.fetch_history') def test_get_fingerprint_rules_history_profile_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, profile, has_access): """Test profile access validation for fingerprint rule history url.""" mock_logic.return_value = [ { 'territory': 'US', 'service': 'meta', 'policy': 'monetize', 'start': datetime.datetime(2019, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'end': datetime.datetime(2020, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'timestamp': datetime.datetime(2022, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'profile_type': 'ContentProfile', 'profile_id': '123', 'event_type': 'created' } ] mock_permissions.return_value = (has_access, profile) expected_result = { 'history': [ { 'territory': 'US', 'service': 'meta', 'policy': 'monetize', 'start': datetime.datetime(2019, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'end': datetime.datetime(2020, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'timestamp': datetime.datetime(2022, 6, 15, 0, 0, tzinfo=datetime.timezone.utc).isoformat(), # noqa:E501 'profile_type': 'ContentProfile', 'profile_id': '123', 'event_type': 'created' } ] } response = app_client.get('/vendors/123/rules/history?service=meta&policy=monetize') if has_access: assert mock_logic.call_args_list == [ call('Vendor', 123, {'service': 'meta', 'policy': 'monetize'}) ] assert mock_neo4j_driver.session.called assert response.json == expected_result else: assert response.status_code == 403 assert response.json['error'] == 'NotAuthorizedOperation' assert not mock_logic.called @patch('sound_recordings.handlers.g', spec=['log']) def test_exception_handler(mock_g): """Verify exception_Handler returns 500 status code and json payload.""" message = ( 'The server encountered an internal error ' 'and was unable to complete your request.' ) mock_error = MagicMock() server_response = handlers.exception_handler(mock_error) mock_g.log.exception.assert_called_with(mock_error) # assert status code is 500 assert server_response.status_code == 500 # assert json payload response_message = json.loads(server_response.data.decode()) assert response_message['message'] == message assert response_message['code'] == response.error.ERROR_CODE_INTERNAL_ERROR @patch('connector_neo4j.Neo4jSession.__enter__') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_body_required( mock_logic, mock_neo4j_session_enter, app_client): """Test required fields for a rule are checked.""" url = '/tracks/bulk/rules' output_message = {'_schema': ['Input must be a dictionary at the root level.']} response = app_client.post(url, json=[{}]) assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called assert not mock_neo4j_session_enter.called @pytest.mark.parametrize( 'input_data, output_message', [ # basic valid case ( [('US', 'monetize', 'meta', '2019-06-15T00:00:00+00:00', None)], None ), # non-allowed start ( [('US', 'carveout', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'start': ['cannot be set when policy == carveout']}} ), # not-allowed end ( [('US', 'carveout', 'meta', None, '2019-05-15T00:00:00+00:00')], {'0': {'end': ['cannot be set when policy == carveout']}} ), # start after end ( [('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2019-05-15T00:00:00+00:00')], # noqa:E501 {'0': {'end': ['cannot be before start'], 'start': ['cannot be after end']}} # noqa:E501 ), # start and end same ( [('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2019-05-15T00:00:00+00:00')], # noqa:E501 {'0': {'end': ['cannot equal start'], 'start': ['cannot equal end']}} ), # carveout policy mixed ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2020-05-15T00:00:00+00:00'), # noqa:E501 ('US', 'carveout', 'meta', None, None) ], {'1': {'policy': ['cannot mix carveout rules with other rules in meta:US']}} # noqa:E501 ), # more than one carveout ( [ ('US', 'carveout', 'meta', None, None), ('US', 'carveout', 'meta', None, None) ], { '0': {'policy': ['cannot set multiple carveout rules in meta:US']}, '1': {'policy': ['cannot set multiple carveout rules in meta:US']} } ), # bad timezone ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+05:00', '2020-05-15T00:00:00+05:00'), # noqa:E501 ], {'0': {'start': ['must be in UTC timezone'], 'end': ['must be in UTC timezone']}} # noqa:E501 ), # date overlap ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', '2022-06-15T00:00:00+00:00'), # noqa:E501 ('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2021-06-15T00:00:00+00:00') # noqa:E501 ], { '0': {'end': ['after rule 1 start in meta:US']}, '1': {'start': ['before rule 0 end in meta:US']} } ), # open end range in middle ( [ ('US', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', None), ('US', 'monetize', 'meta', '2020-05-15T00:00:00+00:00', '2021-06-15T00:00:00+00:00') # noqa:E501 ], {'0': {'end': ['cannot be empty for non-last rule in meta:US']}} ), # start date must be set ( [ ('US', 'monetize', 'meta', None, '2022-05-15T00:00:00+00:00') ], {'0': {'start': ['must be set when policy != carveout']}} ), # invalid policy ( [('US', 'block', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'policy': [f"Must be one of: {', '.join(validation_constants.RULE_POLICIES)}."]}} # noqa:E501 ), # invalid territory ( [('USA', 'monetize', 'meta', '2019-05-15T00:00:00+00:00', None)], {'0': {'territory': [f"Must be one of: {', '.join(validation_constants.RULE_TERRITORIES)}."]}} # noqa:E501 ), # invalid service ( [('US', 'monetize', 'facebook', '2019-05-15T00:00:00+00:00', None)], {'0': {'service': [f"Must be one of: {', '.join(validation_constants.RULE_SERVICES)}."]}} # noqa:E501 ), # sixtyseconds valid for youtube ( [('US', 'sixtyseconds', 'youtube', '2019-06-15T00:00:00+00:00', None)], None ), # sixtyseconds invalid for non-youtube service ( [('US', 'sixtyseconds', 'meta', '2019-06-15T00:00:00+00:00', None)], {'0': {'policy': ['sixtyseconds policy is only valid for the youtube service']}} ), # sixtyseconds invalid for wildcard service ( [('US', 'sixtyseconds', '*', '2019-06-15T00:00:00+00:00', None)], {'0': {'policy': ['sixtyseconds policy is only valid for the youtube service']}} ) ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_body_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, input_data, output_message): """Test validation exceptions for fingerprint rule upsert body.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 2) rules = [] for data in input_data: rule = { 'territory': data[0], 'policy': data[1], 'service': data[2], 'start': data[3], 'end': data[4] } rules.append(rule) url = '/tracks/bulk/rules' request_body = {'12345': rules} response = app_client.post(url, json=request_body) if not output_message: assert mock_logic.called # bulk schema now materializes FingerprintRuleSchema defaults (dates deserialized, # is_start_absolute/is_end_absolute defaulted to False) expected_rules = [] for rule in rules: expected_rules.append({ **rule, 'start': datetime.datetime.fromisoformat(rule['start']) if rule['start'] else None, 'end': datetime.datetime.fromisoformat(rule['end']) if rule['end'] else None, 'is_start_absolute': False, 'is_end_absolute': False, }) assert mock_logic.call_args_list == [ call('Track', 12345, new_rules=expected_rules, profile=('ContentProfile', 123), affected_services=None) ] assert response.json == { 'num_created': 1, 'num_deleted': 2 } assert mock_neo4j_driver.session.called else: assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['json'] == output_message assert not mock_logic.called assert not mock_neo4j_driver.session.called @pytest.mark.parametrize( 'url, output_message', [ ('/tracks/bulk/rules', None), ('/track/bulk/rules', {'obj_type': [f"Must be one of: {', '.join(validation_constants.RULE_OBJECT_SCOPES)}."]}), # noqa:E501 ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_url_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, url, output_message): """Test validation exceptions for fingerprint rule upsert url.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) response = app_client.post(url, json={12345: []}) if not output_message: assert mock_logic.called assert mock_neo4j_driver.session.called else: assert response.status_code == 422 assert response.json['code'] == 'bad_request' assert response.json['message']['view_args'] == output_message assert not mock_logic.called assert not mock_neo4j_driver.session.called @pytest.mark.parametrize( 'profile, has_access', [ (('ContentProfile', 123), True), (('ContentProfile', 456), False), ] ) @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_profile_validation( mock_logic, mock_neo4j_driver, mock_permissions, app_client, profile, has_access): """Test profile access validation for fingerprint rule upsert url.""" mock_logic.return_value = (1, 2) mock_permissions.return_value = (has_access, profile) rules = [{ 'territory': 'US', 'policy': 'carveout', 'service': 'meta', 'start': None, 'end': None }] request_body = {12345: rules} response = app_client.post('/tracks/bulk/rules', json=request_body) if has_access: expected_rules = [{**rules[0], 'is_start_absolute': False, 'is_end_absolute': False}] assert mock_logic.call_args_list == [ call('Track', 12345, new_rules=expected_rules, profile=('ContentProfile', 123), affected_services=None) ] assert mock_neo4j_driver.session.called else: assert response.status_code == 403 assert response.json['error'] == 'NotAuthorizedOperation' assert not mock_logic.called @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_affected_services( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test that affected_services only includes the requested services.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 0) rules = [ {'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, {'territory': 'US', 'policy': 'monetize', 'service': 'tiktok', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, ] request_body = {12345: rules} response = app_client.post( '/tracks/bulk/rules?services=meta,tiktok', json=request_body ) assert response.status_code == 200 affected_services = mock_logic.call_args_list[0].kwargs['affected_services'] assert 'meta' in affected_services assert 'tiktok' in affected_services assert 'youtube' not in affected_services @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_affected_services_rejects_out_of_scope( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test that a 422 is returned when rules contain services outside affected_services.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 0) rules = [ {'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, {'territory': 'US', 'policy': 'monetize', 'service': 'youtube', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, ] response = app_client.post( '/tracks/bulk/rules?services=meta,tiktok', json={12345: rules} ) assert response.status_code == 422 assert response.json['error'] == 'ServiceNotInAffectedServices' assert response.json['services'] == ['youtube'] assert not mock_logic.called @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_wildcard_allowed( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test that affected_services is treated like None when set to '*'.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 0) rules = [ {'territory': 'CA', 'policy': 'monetize', 'service': 'tiktok', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, {'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, ] response = app_client.post( '/tracks/bulk/rules?services=*', json={12345: rules} ) assert response.status_code == 200 assert mock_logic.call_args_list == [ call('Track', 12345, new_rules=[ {**r, 'start': datetime.datetime.fromisoformat(r['start']) if r['start'] else None, 'end': datetime.datetime.fromisoformat(r['end']) if r['end'] else None, 'is_start_absolute': False, 'is_end_absolute': False} for r in rules ], profile=('ContentProfile', 123), affected_services=None) ] @patch('sound_recordings.utils.request_access.check_access') @patch('connector_neo4j.neo4j_driver') @patch('sound_recordings.logic.fingerprint_rules.upsert') def test_post_fingerprint_bulk_rules_wildcard_allowed_in_list( mock_logic, mock_neo4j_driver, mock_permissions, app_client): """Test that affected_services is treated like None when any entry is '*'.""" mock_permissions.return_value = (True, ('ContentProfile', 123)) mock_logic.return_value = (1, 0) rules = [ {'territory': 'CA', 'policy': 'monetize', 'service': 'tiktok', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, {'territory': 'US', 'policy': 'monetize', 'service': 'meta', 'start': '2024-01-01T00:00:00+00:00', 'end': None}, ] response = app_client.post( '/tracks/bulk/rules?services=*,tiktok', json={12345: rules} ) assert response.status_code == 200 assert mock_logic.call_args_list == [ call('Track', 12345, new_rules=[ {**r, 'start': datetime.datetime.fromisoformat(r['start']) if r['start'] else None, 'end': datetime.datetime.fromisoformat(r['end']) if r['end'] else None, 'is_start_absolute': False, 'is_end_absolute': False} for r in rules ], profile=('ContentProfile', 123), affected_services=None) ]