"""Test for AdAction logic tier.""" import datetime from unittest.mock import MagicMock from unittest.mock import patch from botocore.exceptions import ClientError from owsresponse import response import pytest from pythonfeatures import pythonfeatures from podcast.logic import ad_action from podcast.logic import email from podcast.models import ad_action as ad_action_model from podcast.models.api_campaign import ApiCampaign from podcast.utils.exc import OwsError DUE_IN_7_DAYS = 'due_in_7_days' ALL_TIME = 'all_time' @patch('podcast.logic.ad_action.g', spec=['log', 'request_context']) def test_create_ad_action(mock_g, mocker, monkeypatch, ad_read_fixture, mock_current_admin_user): """Test create ad action.""" monkeypatch.setattr(ad_action_model, 'create_ad_action', MagicMock(return_value='cow')) mocker.patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='enabled')) lambda_client = MagicMock() lambda_client.invoke.return_value = {'ResponseMetadata': {'HTTPStatusCode': 200}} monkeypatch.setattr(ad_action, '_get_lambda_client', MagicMock(return_value=lambda_client)) s3_client = MagicMock() s3_client.head_object.return_value = { 'ContentType': 'application/pdf', 'Metadata': { 'original_filename': 'test-ad-action-script.pdf' } } resource = MagicMock() resource.meta.client.copy.return_value = True monkeypatch.setattr(ad_action, '_get_s3_objects', MagicMock(return_value=(s3_client, resource))) result = ad_action.create_ad_action(ad_read_fixture) ad_action_model.create_ad_action.assert_called_once_with(ad_read_fixture) assert result == 'cow' resource.meta.client.copy.assert_called_once_with( { 'Bucket': 'test-orcd-podcast-output-bucket', 'Key': 'fake/file.pdf' }, 'test-orcd-podcast-output-bucket', 'fake/file.pdf', ExtraArgs={ 'ContentType': 'application/pdf', 'ContentDisposition': 'attachment; filename="test-ad-action-script-{}.pdf"'.format( datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')), 'Metadata': {'original_filename': 'test-ad-action-script.pdf'}, 'MetadataDirective': 'REPLACE' }, SourceClient=s3_client ) def test_create_ad_action_duplicate(ad_read_fixture, mock_current_admin_user): """Test create ad action duplicate.""" ad_read_fixture['advertisement_id'] = 'mpad1' assert ad_action.does_ad_action_exist(ad_read_fixture) == {'exists': True} def test_create_ad_action_no_access_for_network_admin(monkeypatch, ad_read_fixture, mock_current_network_admin_user): """Test fails when network admin doesn't have access.""" with pytest.raises(OwsError) as err: ad_action.create_ad_action({'order_id': 'mp2'}) assert err.value.status == 403 def test_create_ad_action_not_admin(monkeypatch, mock_current_user): """Test create ad action when not admbin.""" with pytest.raises(OwsError) as err: ad_action.create_ad_action({}) assert err.value.status == 403 @patch('podcast.logic.ad_action.g', spec=['log', 'request_context']) def test_create_ad_action_av_lambda_error(mock_g, mocker, monkeypatch, ad_read_fixture, mock_current_admin_user): """Test create ad action.""" monkeypatch.setattr(ad_action_model, 'create_ad_action', MagicMock(return_value='cow')) mock_error = { 'Error': { 'Code': 'ResourceNotFoundException', 'Message': 'Function not found: arn:aws:lambda:us-east-1:111111111111:function:lambda-virus-scan' }, 'ResponseMetadata': { 'HTTPStatusCode': 404, } } mocker.patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='enabled')) client = MagicMock() client.invoke.side_effect = ClientError(mock_error, 'invoke') monkeypatch.setattr(ad_action, '_get_lambda_client', MagicMock(return_value=client)) s3_client = MagicMock() s3_client.head_object.return_value = { 'ContentType': 'application/pdf', 'Metadata': { 'original_filename': 'test-ad-action-script.pdf' } } resource = MagicMock() resource.meta.client.copy.return_value = True monkeypatch.setattr(ad_action, '_get_s3_objects', MagicMock(return_value=(s3_client, resource))) with pytest.raises(OwsError) as err: ad_action.create_ad_action(ad_read_fixture) assert err.value.status == 404 assert err.value.message == 'ResourceNotFoundException: Error invoking anti virus lambda, Function not found: ' \ 'arn:aws:lambda:us-east-1:111111111111:function:lambda-virus-scan' def test_create_ad_action_skip_virus_scan(monkeypatch, ad_read_fixture, mock_current_automation_admin_user): """Test create ad action by skipping virus scan.""" ad_read_fixture['is_skip_virus_scan'] = True ad_read_fixture['due_date'] = datetime.datetime(2019, 1, 1) result = ad_action.create_ad_action(ad_read_fixture) assert result['is_virus_free'] is True def test_get_ad_actions(monkeypatch): """Test get ad actions.""" monkeypatch.setattr(ad_action_model, 'get_ad_actions', MagicMock(return_value='cow')) result = ad_action.get_ad_actions(True, True, 10, 0, ALL_TIME) ad_action_model.get_ad_actions.assert_called_once_with(True, True, 10, 0, None, None) assert result == 'cow' @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_get_ad_actions_date_filters(mock_get_feature_flag, monkeypatch, mock_current_network_admin_user): """Test get ad action date filters.""" ad_action.update_ad_action(1, {'due_date': datetime.datetime.utcnow() + datetime.timedelta(weeks=2)}) result = ad_action.get_ad_actions(False, False, 9999, 0, ALL_TIME) assert len(result['items']) == 3 assert result['pagination']['total_records'] == 3 result = ad_action.get_ad_actions(False, False, 9999, 0, DUE_IN_7_DAYS) assert len(result['items']) == 2 assert result['pagination']['total_records'] == 2 @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_get_ad_actions_archived_filter(mock_get_feature_flag, monkeypatch, mock_current_user): """Test get ad action archived filters.""" result = ad_action.get_ad_actions(False, False, 9999, 0, ALL_TIME) assert result['pagination']['total_records'] == 3 assert result['items'][0]['id'] == 1 result = ad_action.get_ad_actions(True, False, 9999, 0, ALL_TIME) assert result['pagination']['total_records'] == 1 assert result['items'][0]['id'] == 5 @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_get_ad_actions_mine_filter(mock_get_feature_flag, monkeypatch, mock_current_user): """Test get ad action only mine filters.""" ad_action.set_virus_free('copies/unique-ad_copy-filename.pdf') result = ad_action.get_ad_actions(False, True, 9999, 0, ALL_TIME) assert len(result['items']) == 3 assert result['items'][0]['id'] == 1 result = ad_action.get_ad_actions(False, False, 9999, 0, ALL_TIME) assert len(result['items']) == 4 assert result['pagination']['total_records'] == 4 def test_delete_ad_action_sends_email(monkeypatch, mock_current_admin_user): """Test delete ad action sends email to assignees when status is other then completed.""" monkeypatch.setattr(email, 'send_ad_action_assigned_to_me_deleted', MagicMock()) result = ad_action.delete_ad_action(1) assert result['is_deleted'] is True assert result['updated_by'] == 2 email.send_ad_action_assigned_to_me_deleted.assert_called_once() def test_delete_ad_action_does_not_send_email(monkeypatch, mock_current_admin_user): """Test delete ad action does not send email to assignees when status is completed.""" monkeypatch.setattr(email, 'send_ad_action_assigned_to_me_deleted', MagicMock()) result = ad_action.delete_ad_action(5) assert result['is_deleted'] is True assert result['updated_by'] == 2 email.send_ad_action_assigned_to_me_deleted.assert_not_called() def test_delete_ad_action_not_admin(monkeypatch, mock_current_user): """Test delete ad action when not admbin.""" with pytest.raises(OwsError) as err: ad_action.delete_ad_action(1) assert err.value.status == 403 def test_update_reject_ad_action_not_admin(monkeypatch, mock_current_user): """Test delete ad action when not admin.""" with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'status': 'rejected'}) assert err.value.status == 403 def test_update_sends_reject_email(monkeypatch, mock_current_admin_user): """Test update sends reject email.""" result = ad_action.update_ad_action(1, {'status': 'rejected', 'rejection_reason': 'Cow'}) email._send_sqs.assert_called() assert result['updated_by'] == 2 def test_update_ad_action_fails_read_only(mock_current_read_only_user): """Test update ad action fails when user is read only.""" with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'is_archived_early': True}) assert err.value.status == 403 def test_reject_with_no_reason(monkeypatch, mock_current_admin_user): """Test can reject without rejection reason.""" result = ad_action.update_ad_action(1, {'status': 'rejected'}) assert result['status'] == 'rejected' @patch('podcast.logic.user_v2.current_user_owns_ad_action_or_raise') @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=True) def test_reject_with_no_ia_restructure_on(mock_get_feature_flag, mock_current_user_owns_ad_action_or_raise, monkeypatch, mock_current_network_admin_user): """Test can reject without rejection reason with ia_restructure FF turned on.""" result = ad_action.update_ad_action(1, {'status': 'rejected'}) assert result['status'] == 'rejected' mock_current_user_owns_ad_action_or_raise.assert_called_once() def test_update_complete_ad_action_producer_requires_approval_no_access(monkeypatch, mock_current_user, mock_update_advertisement, mock_audio_assets): """Test update fails action when ad action requires approval.""" with pytest.raises(OwsError) as err: ad_action.update_ad_action(3, {'status': 'completed'}) assert err.value.status == 403 ApiCampaign.update_advertisement.assert_not_called() @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_complete_ad_action_producer_not_requires_approval(mock_get_feature_flag, monkeypatch, mock_current_admin_user, mock_update_advertisement, mock_audio_assets): """Test update succeeds action when ad action does not require approval.""" ad_action.update_ad_action(3, {'requires_approval': False}) resp = ad_action.update_ad_action(3, {'status': 'completed'}) assert resp['status'] == 'completed' email._send_sqs.assert_called() ApiCampaign.update_advertisement.assert_called_with('1', 'mp2', 'mpad1', {'backgroundAudioFileUrl': 'mockurl.mp3'}) @patch('podcast.logic.user_v2.current_user_owns_ad_action_or_raise') @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=True) def test_update_complete_ad_action_producer_ia_restrcuture_on(mock_get_feature_flag, mock_current_user_owns_ad_action_or_raise, monkeypatch, mock_current_admin_user, mock_update_advertisement, mock_audio_assets): """Test update succeeds action when ad action does not require approval with ia_restructure FF turned on.""" ad_action.update_ad_action(3, {'requires_approval': False}) resp = ad_action.update_ad_action(3, {'status': 'completed'}) assert resp['status'] == 'completed' email._send_sqs.assert_called() mock_current_user_owns_ad_action_or_raise.assert_called_once() ApiCampaign.update_advertisement.assert_called_with('1', 'mp2', 'mpad1', {'backgroundAudioFileUrl': 'mockurl.mp3'}) @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_submit_ad_action_not_admin(mock_get_feature_flag, monkeypatch, mock_current_user): """Test update submit ad action when not admin.""" with pytest.raises(OwsError) as err: ad_action.update_ad_action(3, {'status': 'submitted'}) assert err.value.status == 403 def test_update_complete_ad_action_producer_requires_approval(monkeypatch, mock_current_admin_user, mock_update_advertisement, mock_audio_assets): """Test update succeeds action when ad action does not require approval.""" resp = ad_action.update_ad_action(3, {'status': 'completed'}) email._send_sqs.assert_called() assert resp['status'] == 'completed' ApiCampaign.update_advertisement.assert_called_with('1', 'mp2', 'mpad1', {'backgroundAudioFileUrl': 'mockurl.mp3'}) @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_submit_ad_action(mock_get_feature_flag, monkeypatch, mock_current_user): """Test delete ad action when not admin.""" result = ad_action.update_ad_action(1, {'status': 'submitted'}) email._send_sqs.assert_called() assert result['updated_by'] == 1 def test_update_archive_ad_action(monkeypatch, mock_current_admin_user): """Test archive ad action.""" resp = ad_action.update_ad_action(3, {'is_archived_early': True}) assert resp['is_archived_early'] is True def test_update_ad_action_by_filename(monkeypatch, mock_current_admin_user): """Test update_ad_action_by_filename method.""" monkeypatch.setattr(email, 'send_ad_action_assigned_to_me', MagicMock()) result = ad_action.set_virus_free('copies/unique-ad_copy-filename.pdf') assert result['assignee_ids'] == [2] email.send_ad_action_assigned_to_me.assert_called_once() def test_update_ad_action_assignees_and_due_date(monkeypatch, mock_current_admin_user): """Test update ad action assignees and due date by org admin.""" monkeypatch.setattr(email, 'send_assignees_updated_on_ad_action_assigned_to_me', MagicMock()) monkeypatch.setattr(email, 'send_due_date_updated_on_ad_action_assigned_to_me', MagicMock()) new_due_date = datetime.datetime(2019, 1, 2) result = ad_action.update_ad_action(1, {'assignee_ids': [2, 3], 'due_date': new_due_date}) assert result['assignee_ids'] == [2, 3] assert result['due_date'] == new_due_date assert result['updated_by'] == 2 email.send_assignees_updated_on_ad_action_assigned_to_me.assert_called_once() email.send_due_date_updated_on_ad_action_assigned_to_me.assert_called_once() @patch('podcast.logic.user.current_user_owns_ad_action_or_raise') def test_update_ad_action_assignees( mock_current_user_owns_ad_action_or_raise, monkeypatch, mock_current_network_admin_user_all_access): """Test update ad action assignees by network admin and no extra access checks done.""" monkeypatch.setattr(email, 'send_assignees_updated_on_ad_action_assigned_to_me', MagicMock()) result = ad_action.update_ad_action(1, {'assignee_ids': [2, 3]}) assert result['assignee_ids'] == [2, 3] assert result['updated_by'] == 13 mock_current_user_owns_ad_action_or_raise.assert_not_called() @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_ad_action_assignees_failure( mock_get_feature_flag, monkeypatch, mock_current_network_admin_user_second): """Test update ad action assignees for network admin with no access.""" monkeypatch.setattr(email, 'send_assignees_updated_on_ad_action_assigned_to_me', MagicMock()) with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'assignee_ids': [2, 3]}) assert err.value.status == 403 email.send_assignees_updated_on_ad_action_assigned_to_me.assert_not_called() @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) @patch('podcast.logic.user.current_user_owns_ad_action_or_raise') def test_update_ad_action_due_date( mock_get_feature_flag, mock_current_user_owns_ad_action_or_raise, monkeypatch, mock_current_network_admin_user): """Test update ad action due date by network admin and extra access check done.""" monkeypatch.setattr(email, 'send_due_date_updated_on_ad_action_assigned_to_me', MagicMock()) new_due_date = datetime.datetime(2019, 1, 2) result = ad_action.update_ad_action(1, {'due_date': new_due_date}) assert result['due_date'] == new_due_date assert result['updated_by'] == 6 email.send_due_date_updated_on_ad_action_assigned_to_me.assert_called_once() mock_current_user_owns_ad_action_or_raise.assert_called_once() @patch('podcast.logic.user_v2.current_user_owns_ad_action_or_raise') @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=True) def test_update_ad_action_due_date_with_ia_restructure_on( mock_get_feature_flag, mock_current_user_owns_ad_action_or_raise, monkeypatch, mock_current_network_admin_user): """Test update ad action due date by network admin and extra access check done with ia_restructure FF turned on.""" monkeypatch.setattr(email, 'send_due_date_updated_on_ad_action_assigned_to_me', MagicMock()) new_due_date = datetime.datetime(2019, 1, 2) result = ad_action.update_ad_action(1, {'due_date': new_due_date}) assert result['due_date'] == new_due_date assert result['updated_by'] == 6 email.send_due_date_updated_on_ad_action_assigned_to_me.assert_called_once() mock_current_user_owns_ad_action_or_raise.assert_called_once() def test_update_ad_action_due_date_failure(monkeypatch, mock_current_user): """Test update ad action due date failure for producer.""" monkeypatch.setattr(email, 'send_due_date_updated_on_ad_action_assigned_to_me', MagicMock()) new_due_date = datetime.datetime(2019, 1, 2) with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'due_date': new_due_date}) assert err.value.status == 403 email.send_due_date_updated_on_ad_action_assigned_to_me.assert_not_called() @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_ad_action_script(mock_get_feature_flag, monkeypatch, mock_current_network_admin_user): """Test update ad action script.""" s3_client = MagicMock() s3_client.head_object.return_value = { 'ContentType': 'application/pdf', 'Metadata': { 'original_filename': 'test-ad-action-script-update.pdf' } } resource = MagicMock() resource.meta.client.copy.return_value = True monkeypatch.setattr(ad_action, '_get_s3_objects', MagicMock(return_value=(s3_client, resource))) result = ad_action.update_ad_action(1, {'copy_url_path': 'fake/file.pdf'}) assert result['copy_url_path'] == 'https://None/fake/file.pdf?' \ 'Policy=policy&Signature=signature&Key-Pair-Id=key-pair' resource.meta.client.copy.assert_called_once_with( { 'Bucket': 'test-orcd-podcast-output-bucket', 'Key': 'fake/file.pdf' }, 'test-orcd-podcast-output-bucket', 'fake/file.pdf', ExtraArgs={ 'ContentType': 'application/pdf', 'ContentDisposition': 'attachment; filename="test-ad-action-script-update-{}.pdf"'.format( datetime.datetime.now().strftime('%Y-%m-%d-%H-%M-%S')), 'Metadata': {'original_filename': 'test-ad-action-script-update.pdf'}, 'MetadataDirective': 'REPLACE' }, SourceClient=s3_client ) def test_set_virus_free_on_create(monkeypatch, mock_current_admin_user): """Test set virus free for script upload on create ad action by org admin.""" monkeypatch.setattr(email, 'send_ad_action_assigned_to_me', MagicMock()) result = ad_action.set_virus_free('copies/unique-ad_copy-filename.pdf') assert result['is_virus_free'] is True email.send_ad_action_assigned_to_me.assert_called_once() def test_set_virus_free_on_script_reupload(monkeypatch, mock_current_admin_user): """Test set virus free for script re-upload on ad action by org admin.""" monkeypatch.setattr(email, 'send_script_updated_on_ad_action_assigned_to_me', MagicMock()) result = ad_action.set_virus_free('fake/path.pdf') assert result['is_virus_free'] is True email.send_script_updated_on_ad_action_assigned_to_me.assert_called_once() def test_update_ad_action_script_fails_for_producer(monkeypatch, mock_current_user): """Test update ad action script failure for producer.""" with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'copy_url_path': 'copy/path.pdf'}) assert err.value.status == 403 @patch('podcast.utils.feature_flag_utils.get_feature_flag', return_value=False) def test_update_ad_action_script_fails_for_network_admin( mock_get_feature_flag, monkeypatch, mock_current_network_admin_user_second): """Test update ad action script failure for network admin with no access.""" monkeypatch.setattr(email, 'send_script_updated_on_ad_action_assigned_to_me', MagicMock()) with pytest.raises(OwsError) as err: ad_action.update_ad_action(1, {'copy_url_path': 'copy/path.pdf'}) assert err.value.status == 403 email.send_script_updated_on_ad_action_assigned_to_me.assert_not_called()