"""Test for the Stream logic.""" import datetime from unittest.mock import MagicMock, call, patch import neo4j from marshmallow import Schema, fields from marshmallow.validate import Equal, OneOf from owsresponse import response, status from segment import analytics from notifications import config from notifications.logic import stream @patch( 'notifications.models.stream.add_activity', return_value=response.Response(status=status.CREATED), ) def test_add_activity(mock_add_activity): """Test adding an activty to a feed.""" feed_name = 'label_track_placement' feed_id = 'vendor_123' payload = { 'actor': 'The Orchard Activity Detector', 'verb': 'Placed', 'object': 'Track', 'target': 'Playlist', 'track_id': 123, } result = stream.add_activity(feed_name, feed_id, payload) mock_add_activity.assert_called_with(feed_name, feed_id, payload) assert result.status == status.CREATED @patch('notifications.models.stream.add_activity', return_value=response.create_fatal_response()) def test_add_activity_error(mock_add_activity): """Test adding an activty to a feed when an error occurs.""" feed_name = 'label_track_placement' feed_id = 'vendor_123' payload = { 'actor': 'The Orchard Activity Detector', 'verb': 'Placed', 'object': 'Track', 'target': 'Playlist', 'track_id': 123, } expected_calls = [ call(feed_name, feed_id, payload), call(feed_name, feed_id, payload), call(feed_name, feed_id, payload), call(feed_name, feed_id, payload), call(feed_name, feed_id, payload), call(feed_name, feed_id, payload), ] result = stream.add_activity(feed_name, feed_id, payload) mock_add_activity.assert_has_calls(expected_calls) assert result.status == status.INTERNAL_ERROR # @patch('notifications.logic.stream.g') def test_add_label_activities_to_user(app_context, mocker): """Test add_label_activities_to_user.""" activity_type = 'Vendor' activity_id = 7123 subscription_feed_type = 'productApproval' payload = {'foo': 'bar'} mocker.patch( 'notifications.models.subscriptions.get_subscription_by_param', return_value=response.Response({'name': 'product_approval'}), ) mocker.patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response( message=[ {'profile_type': 'LabelProfile', 'profile_id': 123}, {'profile_type': 'LabelProfile', 'profile_id': 456}, {'profile_type': 'InsightsProfile', 'profile_id': 789}, ] ), ) mocker.patch( 'notifications.models.label.get_default_brand_for_label', return_value=response.Response(message={'name': 'theorchard'}), ) mocker.patch( 'notifications.models.label.get_name_for_label', return_value=response.Response(message={'name': 'Test Label'}), ) result = stream.add_label_activities_to_user( activity_type, activity_id, subscription_feed_type, payload ) assert result.status == 201 assert result.message == ['LabelProfile_123', 'LabelProfile_456'] @patch( 'notifications.models.subscriptions.get_subscription_by_param', return_value=response.create_not_found_response('No subscription found'), ) @patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response({'not': 'called'}), ) def test_add_label_activities_to_user_no_subscription( mock_subscribed_profiles, mock_subscription_by_type, app_context ): """Test add_label_activities_to_user.""" activity_type = 'Vendor' activity_id = 7123 subscription_feed_type = 'productApproval' payload = {'foo': 'bar'} result = stream.add_label_activities_to_user( activity_type=activity_type, activity_id=activity_id, subscription_feed_type=subscription_feed_type, payload=payload, ) assert result.status == 404 assert result.errors == {'code': 'not_found_error', 'message': 'No subscription found'} assert not mock_subscribed_profiles.called @patch('notifications.models.ows_users.get_user', return_value=response.Response()) @patch('notifications.models.stream.subscribe', return_value=response.Response()) def test_subscribe(mock_subscribe, mock_get_user, monkeypatch): """Test subscribing to a feed.""" monkeypatch.setattr(analytics, 'identify', MagicMock()) monkeypatch.setattr(analytics, 'track', MagicMock()) monkeypatch.setattr(config, 'SEGMENT_WRITE_KEY', MagicMock()) user_feed_name = 'user_email_notifications' user_id = 'alw:123' feed_name = 'label_track_placement' feed_id = 'vendor_123' result = stream.subscribe(user_feed_name, user_id, feed_name, feed_id, None) analytics.identify.assert_called_once_with(user_id) analytics.track.assert_called_once_with( user_id, 'Subscribed', {'feed_name': feed_name, 'feed_id': feed_id, 'user_id': user_id} ) mock_get_user.assert_called_with(user_id) mock_subscribe.assert_called_with(user_feed_name, user_id, feed_name, feed_id, None) assert result.status == status.OK @patch('notifications.models.ows_users.get_user', return_value=response.create_fatal_response()) @patch('notifications.models.stream.subscribe', return_value=response.Response()) def test_subscribe_invalid_user(mock_subscribe, mock_get_user): """Test subscribing to a feed when the user id is invalid.""" user_feed_name = 'user_email_notifications' user_id = '123' feed_name = 'label_track_placement' feed_id = 'vendor_123' result = stream.subscribe(user_feed_name, user_id, feed_name, feed_id) mock_get_user.assert_called_with(user_id) mock_subscribe.assert_not_called() assert result.status == status.INTERNAL_ERROR @patch('notifications.models.stream.unsubscribe', return_value=response.Response()) def test_unsubscribe(mock_unsubscribe, monkeypatch): """Test unsubscribing from a feed.""" monkeypatch.setattr(analytics, 'identify', MagicMock()) monkeypatch.setattr(analytics, 'track', MagicMock()) monkeypatch.setattr(config, 'SEGMENT_WRITE_KEY', MagicMock()) user_feed_name = 'user_email_notifications' user_id = 'alw:123' feed_name = 'label_track_placement' feed_id = 'vendor_123' result = stream.unsubscribe(user_feed_name, user_id, feed_name, feed_id, None) analytics.identify.assert_called_once_with(user_id) analytics.track.assert_called_once_with( user_id, 'Unsubscribed', {'feed_name': feed_name, 'feed_id': feed_id, 'user_id': user_id} ) mock_unsubscribe.assert_called_with(user_feed_name, user_id, feed_name, feed_id, None) assert result.status == status.OK @patch('notifications.models.stream.get_user_notifications', return_value=response.Response()) def test_get_user_notifications(mock_get_user_notifications): """Test getting a user's notifications.""" user_id = 'alw:123' user_feed_name = 'user_email_notifications' result = stream.get_user_notifications(user_id, user_feed_name) mock_get_user_notifications.assert_called_with(user_id, user_feed_name, None) assert result.status == status.OK @patch('notifications.models.stream.get_user_subscriptions', return_value=response.Response()) def test_get_user_subscriptions(mock_get_user_subscriptions): """Test getting a user's subscriptions.""" user_id = 'alw:123' user_feed_name = 'user_email_notifications' result = stream.get_user_subscriptions(user_id, user_feed_name) mock_get_user_subscriptions.assert_called_with(user_id, user_feed_name, None) assert result.status == status.OK @patch('notifications.models.stream.get_feed_subscribers', return_value=response.Response()) def test_get_feed_subscribers(mock_get_feed_subscribers): """Test getting a feed's subscribers.""" feed_name = 'label_analytics_digest' feed_id = 'vendor_123' result = stream.get_feed_subscribers(feed_name, feed_id) mock_get_feed_subscribers.assert_called_with(feed_name, feed_id) assert result.status == status.OK @patch( 'notifications.logic.stream.add_label_activities_to_user', return_value=response.Response(message={'profiles': 'affected'}), ) @patch( 'notifications.models.product.get_product_details_by_id', return_value=response.Response( message={ 'name': 'product name', 'upc': 'adasd33453sdf', 'artist': {'name': 'artist 1'}, 'contextType': 'digital', 'project': {'name': 'project 1', 'vendorId': 7123, 'id': 134}, } ), ) @patch( 'notifications.models.label.get_default_brand_for_label', return_value=response.Response(message={'name': 'theorchard'}), ) @patch( 'notifications.models.label.get_name_for_label', return_value=response.Response(message={'name': 'Test Label'}), ) @patch( 'notifications.models.ows_account.get_vendor_assigned_to', return_value=response.Response(message={'id': 21}), ) @patch( 'notifications.models.ows_users.get_user', return_value=response.Response(message={'email': 'testemail@theorchard.com'}), ) def test_add_product_rejection_activity( mock_get_user, mock_get_assigned_to, mock_get_label_name, mock_get_default_brand, mock_get_product, mock_add_activity, app_context, ): """Test add_product_rejection_activity.""" product_id = 13434 rejection_reasons = [{'some': 'error'}] result = stream.add_product_rejection_activity(product_id, rejection_reasons) assert result assert mock_get_product.call_count == 1 assert mock_add_activity.call_count == 1 assert mock_get_default_brand.call_count == 1 assert mock_get_label_name.call_count == 1 assert mock_get_assigned_to.call_count == 1 assert mock_get_user.call_count == 1 mock_add_activity.assert_called_with( 'Vendor', 7123, 'productRejection', { 'actor': 'Product', 'verb': 'Rejected', 'object': 'Audio Product', 'artist_name': 'artist 1', 'product_name': 'product name', 'upc': 'adasd33453sdf', 'project_id': 134, 'product_id': 13434, 'assigned_to_email': 'testemail@theorchard.com', 'context_type': 'digital', 'rejection_reasons': [{'some': 'error'}], 'template_name': 'digital_rejection', 'original_feed': 'email_notification_rejection:Vendor_7123', 'default_brand': {'name': 'theorchard'}, 'label_name': {'name': 'Test Label'}, 'label_id': 7123, }, ) assert result.message == mock_add_activity().message @patch( 'notifications.logic.stream.add_label_activities_to_user', return_value=response.Response(message={'profiles': 'affected'}), ) @patch( 'notifications.models.product.get_product_details_by_id', return_value=response.create_not_found_response(), ) def test_add_product_rejection_product_not_found(mock_get_product, mock_add_activity): """Test add_product_rejection_activity.""" product_id = 13434 rejection_reasons = [{'some': 'error'}] result = stream.add_product_rejection_activity(product_id, rejection_reasons) assert result.status == 404 assert mock_get_product.call_count == 1 assert mock_add_activity.call_count == 0 @patch( 'notifications.logic.stream.add_label_activities_to_user', return_value=response.Response(message={'profiles': 'affected'}), ) @patch( 'notifications.models.product.get_product_details_by_id', return_value=response.Response( message={ 'name': 'product name', 'upc': 'adasd33453sdf', 'artist': {'name': 'artist 1'}, 'contextType': 'digital', 'project': {'name': 'project 1', 'vendorId': 7123, 'id': 134}, 'saleStartDate': neo4j.time.Date(year=2022, month=8, day=11), } ), ) @patch( 'notifications.models.label.get_default_brand_for_label', return_value=response.Response(message={'name': 'theorchard'}), ) @patch( 'notifications.models.label.get_name_for_label', return_value=response.Response(message={'name': 'Test Label'}), ) def test_add_product_approval_activity( mock_get_label_name, mock_get_default_brand, mock_get_product, mock_add_activity, app_context ): """Test add_product_approval_activity.""" product_id = 13434 result = stream.add_product_approval_activity(product_id) assert result assert mock_get_product.call_count == 1 assert mock_add_activity.call_count == 1 assert mock_get_default_brand.call_count == 1 assert mock_get_label_name.call_count == 1 assert result.message == mock_add_activity().message @patch( 'notifications.logic.stream.add_label_activities_to_user', return_value=response.Response(message={'profiles': 'affected'}), ) @patch( 'notifications.models.product.get_product_details_by_id', return_value=response.create_not_found_response(), ) def test_add_product_approval_product_not_found(mock_get_product, mock_add_activity): """Test add_product_approval_activity.""" product_id = 13434 result = stream.add_product_approval_activity(product_id) assert result.status == 404 assert mock_get_product.call_count == 1 assert mock_add_activity.call_count == 0 @patch('notifications.logic.stream.g') def test_add_playlist_placement_activity(mock_g, app_context, mocker): """Test happy path playlist placement adding.""" mock_profiles = mocker.patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response( status=200, message=[ {'profile_type': 'InsightsProfile', 'profile_id': 123}, {'profile_type': 'InsightsProfile', 'profile_id': 456}, {'profile_type': 'InsightsProfile', 'profile_id': 789}, ], ), ) mock_details = mocker.patch( 'notifications.models.graphql_router.get_sound_recording_details', return_value=('aaa', 'bbb', ['ccc']), ) mock_add_activity = mocker.patch( 'notifications.models.stream.add_playlist_placement_activity', return_value=response.Response(status=201), ) result = stream.add_playlist_placement_activity( datetime.datetime(2019, 6, 15, 12, 30, 5), {'id': '987', 'dsp': 'spotify'}, { 'isrc': 'abc', 'tracks': [ {'id': 789, 'vendor_id': 123, 'subaccount_id': 456}, {'id': 987, 'vendor_id': 123, 'subaccount_id': 456}, {'id': 879, 'vendor_id': 321, 'subaccount_id': None}, ], }, ) assert result.status == 201 assert mock_details.call_args_list == [call('abc')] assert mock_profiles.call_args_list == [ call( ['InsightsProfile', 'ArtistProfile', 'LabelProfile'], 'GlobalSoundRecording', 'aaa', 'HAS_FOLLOWED', 'profile', None, ), call(['LabelProfile', 'InsightsProfile'], 'Vendor', 123, 'HAS_FOLLOWED', 'profile', None), call(['LabelProfile', 'InsightsProfile'], 'Vendor', 321, 'HAS_FOLLOWED', 'profile', None), call( ['LabelProfile', 'InsightsProfile'], 'Subaccount', 456, 'HAS_FOLLOWED', 'profile', None ), call( ['InsightsProfile', 'ArtistProfile', 'LabelProfile'], 'GlobalParticipant', 'ccc', 'HAS_FOLLOWED', 'profile', None, ), ] event_metadata = { 'sound_recording': {'isrc': 'abc', 'id': 'aaa', 'name': 'bbb'}, 'playlist': {'id': '987', 'dsp': 'spotify'}, } assert mock_add_activity.call_args_list == [ call( [ { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15, 12, 30, 5), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 123, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_123', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15, 12, 30, 5), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 456, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_456', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15, 12, 30, 5), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 789, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_789', }, ] ) ] @patch( 'notifications.models.graphql_router.get_sound_recording_details', return_value=(None, None, []) ) def test_add_playlist_placement_activity_failed_id(mock_details): """Test isrc not found.""" result = stream.add_playlist_placement_activity('2019-06-15 12:30:05', {}, {'isrc': 'abc'}) assert result.status == 404 @patch('notifications.logic.stream.g') def test_add_trending_track_activity(mock_g, app_context, mocker): """Test happy path trending track adding.""" mock_add_activity = mocker.patch( 'notifications.models.stream.add_trending_track_activity', return_value=response.Response(status=201), ) mock_details = mocker.patch( 'notifications.models.graphql_router.get_sound_recording_details', return_value=('aaa', 'bbb', ['123']), ) mock_profiles = mocker.patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response( status=200, message=[ {'profile_type': 'InsightsProfile', 'profile_id': 123}, {'profile_type': 'InsightsProfile', 'profile_id': 456}, {'profile_type': 'InsightsProfile', 'profile_id': 789}, ], ), ) result = stream.add_trending_track_activity( datetime.datetime(2019, 6, 15), 'spotify', 'USA', 100, 1000, {'id': 1234, 'isrc': 'xyz', 'vendor_id': 5678, 'subaccount_id': 9012}, ) assert result.status == 201 assert mock_details.call_args_list == [call('xyz')] assert mock_profiles.call_args_list == [ call( ['InsightsProfile', 'ArtistProfile', 'LabelProfile'], 'GlobalSoundRecording', 'aaa', 'HAS_FOLLOWED', 'profile', None, ), call(['LabelProfile', 'InsightsProfile'], 'Vendor', 5678, 'HAS_FOLLOWED', 'profile', None), call( ['LabelProfile', 'InsightsProfile'], 'Subaccount', 9012, 'HAS_FOLLOWED', 'profile', None ), call( ['InsightsProfile', 'ArtistProfile', 'LabelProfile'], 'GlobalParticipant', '123', 'HAS_FOLLOWED', 'profile', None, ), ] event_metadata = { 'track': { 'id': 1234, 'isrc': 'xyz', 'vendor_id': 5678, 'subaccount_id': 9012, 'name': 'bbb', }, 'day_streams': 1000, 'percent_diff': 100, 'dsp': 'spotify', 'region': 'USA', } assert mock_add_activity.call_args_list == [ call( [ { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 123, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_123', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 456, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_456', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant', 'sound_recording', 'sub_account', 'vendor'], 'profile_id': 789, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_789', }, ] ) ] @patch( 'notifications.models.graphql_router.get_sound_recording_details', return_value=(None, None, []) ) def test_add_trending_track_activity_failed_id(mock_details): """Test isrc not found.""" result = stream.add_trending_track_activity( '2019-06-15', 'spotify', 'USA', 100, 1000, {'id': 1234, 'isrc': 'xyz', 'label_id': 5678} ) assert result.status == 404 @patch('notifications.models.subscriptions.get_subscribed_profiles') def test_fan_out_empty_identifier(mock_get_profiles): """Test fan out with follow config missing identifier.""" results = stream._get_fanout_events( [('xyz', None)], datetime.datetime(2019, 6, 15, 12, 0, 0), {} ) assert not mock_get_profiles.called assert results == [] @patch('notifications.models.subscriptions.get_subscribed_profiles') def test_fan_out_empty_identifier_profile_cypher(mock_get_profiles): """Test fan out with follow config missing identifier.""" schema_structure = { 'profile_type': fields.Str(required=True, validate=OneOf(['LabelProfile'])), 'relationship': fields.Str(required=True, validate=Equal('HAS_AUTO_FOLLOWED')), 'entity_node_type': fields.Constant('Vendor', required=True), 'relationship_from': fields.Constant('identity', load_only=True), 'subscription_name': fields.Constant('approval', load_only=True), } test_schema = Schema.from_dict(schema_structure) stream._get_fanout_events( [(test_schema, 7123)], datetime.datetime(2019, 6, 15, 12, 0, 0), {}, ) assert mock_get_profiles.called mock_get_profiles.assert_called_once_with( ['LabelProfile'], 'Vendor', 7123, 'HAS_AUTO_FOLLOWED', 'identity', 'approval' ) @patch('notifications.logic.stream.g') def test_add_playlist_placement_no_profiles(mock_g, app_context, mocker): """Test no profiles subscribed.""" mocker.patch( 'notifications.models.graphql_router.get_sound_recording_details', return_value=('xyz', 'aaa', []), ) mocker.patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response(status=200, message=[]), ) result = stream.add_playlist_placement_activity( datetime.datetime(2019, 6, 15, 12, 30, 5), {}, {'isrc': 'abc', 'tracks': [{'id': 789, 'vendor_id': 123, 'subaccount_id': 456}]}, ) assert result.status == 200 @patch( 'notifications.logic.stream.g', ) def test_add_social_spike_activity(mock_g, app_context, mocker): """Test adding social activity with resolved id.""" mock_add_activity = mocker.patch( 'notifications.models.stream.add_social_spike_activity', return_value=response.Response(status=201), ) mock_get_id = mocker.patch( 'notifications.models.graphql_router.get_participant_by_chartmetric_id', return_value=('abcd', 'qwerty'), ) mock_profiles = mocker.patch( 'notifications.models.subscriptions.get_subscribed_profiles', return_value=response.Response( status=200, message=[ {'profile_type': 'InsightsProfile', 'profile_id': 123}, {'profile_type': 'InsightsProfile', 'profile_id': 456}, {'profile_type': 'InsightsProfile', 'profile_id': 789}, ], ), ) result = stream.add_social_spike_activity(datetime.datetime(2019, 6, 15), 'twitter', 100, 12345) assert mock_get_id.called assert mock_add_activity.called assert mock_profiles.call_args_list == [ call( ['InsightsProfile', 'ArtistProfile', 'LabelProfile'], 'GlobalParticipant', 'abcd', 'HAS_FOLLOWED', 'profile', None, ) ] event_metadata = {'id': 'abcd', 'name': 'qwerty', 'new_followers': 100, 'network': 'twitter'} assert mock_add_activity.call_args_list == [ call( [ { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant'], 'profile_id': 123, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_123', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant'], 'profile_id': 456, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_456', }, { 'metadata': event_metadata, 'time': datetime.datetime(2019, 6, 15), 'sources': ['participant'], 'profile_id': 789, 'profile_type': 'InsightsProfile', 'profile_string': 'InsightsProfile_789', }, ] ) ] assert result.status == 201 @patch( 'notifications.models.graphql_router.get_participant_by_chartmetric_id', return_value=(None, None), ) @patch('notifications.models.stream.add_social_spike_activity') def test_add_social_spike_activity_failed_id(mock_add_activity, mock_get_id): """Test adding social activity without resolved id.""" result = stream.add_social_spike_activity('2019-06-15 15:00:00', 'twitter', 100, 12345) assert mock_get_id.called assert not mock_add_activity.called assert result.status == 404 @patch( 'notifications.models.stream.add_activity', return_value=response.Response(status=status.CREATED), ) @patch('notifications.logic.stream.g') def test_add_streams_updated_activity(_mock_g, mock_add_activity, app_context): """Test adding streams updated activity.""" timestamp = '2025-08-11 17:35:00' store_id = 286 available_date = '2025-08-10' stream.add_streams_updated_activity(timestamp, store_id, available_date) mock_add_activity.assert_called_once_with( 'streams_updated', 'streams_updated_all', { 'payload': {'store_id': store_id}, 'actor': 'store', 'verb': 'streams_updated', 'foreign_id': f'streams_updated:{store_id}', 'object': { 'activity_sources': ['store'], 'available_date': available_date, 'store_id': store_id, }, 'time': timestamp, }, )