import uuid from unittest.mock import patch import json from models.campaign_approval import ApprovalRequest, ApprovalRequestCampaign from tests import BaseTestCase, FixtureLoader, mock_for_user from tests.media_plan.fixtures.media_plan_approval_fixture import fixture from models import models, UserProject, Project from db import db from models.campaign import Campaign APPROVAL_ROLE = 4 requesterMock = mock_for_user(3, "user-3@example.com", "sme-dna|9op0091d6cuhjp30fe6ffth34", name="User Admin") def mocked_email_send(*args, **kwargs): return [{"user-2@example.com": uuid.uuid4()}] def mocked_42_email_send(*args, **kwargs): return [{"suprauser42@example.com": uuid.uuid4()}] @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) class TestMediaPlanApproval(BaseTestCase): url = "/projects/{}/media-plans/{}/{}" def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) def test_get_users_list_for_approval(self): response = self.execute_get( self.url.format(500, 42, "approver-users"), headers=self.headers ) self.assertEqual(200, response.status_code) data = response.json self.assertEqual(len(data), 4) self.soft_assert(self.assertEqual, data[0], { "id": 1, "name": "First User", "email": "dev-marketing-user-1@example.com", "hasProjectAccess": True # label } ) self.soft_assert(self.assertEqual, data[1], { "id": 3, "name": "User Admin", "email": "user-3@example.com", "hasProjectAccess": True } ) self.soft_assert(self.assertEqual, data[2], { "id": 2, "name": "User2", "email": "user-2@example.com", "hasProjectAccess": True # have user project } ) self.soft_assert(self.assertEqual, data[3], { "id": 33, "name": "User3", "email": "user3@example.com", "hasProjectAccess": False # admin without label or user project } ) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request(self): data = { "campaigns": [ "34ed4c22-e1f9-4184-a06d-f04cbdb97455", ], "reviewers": [2, ], } response = self.execute_post(self.url.format(500, 42, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 200) data = response.json approval_request_id = data.get("approvalRequestId") approval_request = db.session.query(ApprovalRequest).filter(ApprovalRequest.id==approval_request_id).one_or_none() self.soft_assert(self.assertIsNotNone, approval_request) self.soft_assert_equal(approval_request.requester_id, 3) self.soft_assert_equal(approval_request.reviewer_id, 2) self.soft_assert_equal(approval_request.media_plan_id, 42) approval_request_campaigns = approval_request.approval_request_campaigns self.soft_assert_equal(len(approval_request_campaigns), 1) self.soft_assert_equal(approval_request_campaigns[0].campaign_uuid, uuid.UUID("34ed4c22-e1f9-4184-a06d-f04cbdb97455")) self.soft_assert_equal(approval_request_campaigns[0].is_approved, False) self.soft_assert_equal(approval_request_campaigns[0].is_deleted, False) campaign = approval_request_campaigns[0].campaign self.soft_assert_equal(campaign.workflow_status, 'Pending Approval') reviewer_user_project = db.session.query(UserProject).filter( UserProject.user_id == 2, UserProject.project_id == 500 ).all() self.soft_assert_equal(len(reviewer_user_project), 2) self.soft_assert_equal(reviewer_user_project[0].role, 3) self.soft_assert_equal(reviewer_user_project[1].role, 4) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_with_empty_campaigns(self): data = { "campaigns": [], "reviewers": [2, ], } response = self.execute_post(self.url.format(500, 42, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 422) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_with_invalid_campaign(self): data = { "campaigns": ["34ed4c22-e1f9-4184-a06d-f04cbdb97444", "34ed4c22-e1f9-4184-a06d-f04cbdb97451"], "reviewers": [2, ], } response = self.execute_post(self.url.format(500, 42, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 422) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_with_campaign_from_another_media_plan(self): data = { "campaigns": ["34ed4c22-e1f9-4184-a06d-f04cbdb97455", "34ed4c22-e1f9-4184-a06d-f04cbdb97451"], "reviewers": [2, ], } response = self.execute_post(self.url.format(500, 43, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 404) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_42_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_for_approver_exist_only_one_approver_role(self): data = { "campaigns": [ "34ed4c22-e1f9-4184-a06d-f04cbdb97451", ], "reviewers": [42, ], } response = self.execute_post(self.url.format(501, 44, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 200) data = response.json approval_request_id = data.get("approvalRequestId") approval_request = db.session.query(ApprovalRequest).filter( ApprovalRequest.id == approval_request_id).one_or_none() self.soft_assert(self.assertIsNotNone, approval_request) self.soft_assert_equal(approval_request.requester_id, 3) self.soft_assert_equal(approval_request.reviewer_id, 42) self.soft_assert_equal(approval_request.media_plan_id, 44) approval_request_campaigns = approval_request.approval_request_campaigns self.soft_assert_equal(len(approval_request_campaigns), 1) self.soft_assert_equal(approval_request_campaigns[0].campaign_uuid, uuid.UUID("34ed4c22-e1f9-4184-a06d-f04cbdb97451")) self.soft_assert_equal(approval_request_campaigns[0].is_approved, False) self.soft_assert_equal(approval_request_campaigns[0].is_deleted, False) campaign = approval_request_campaigns[0].campaign self.soft_assert_equal(campaign.workflow_status, 'Pending Approval') approval_user_project = db.session.query(UserProject).filter( UserProject.user_id == 42, UserProject.project_id == 501, UserProject.role == APPROVAL_ROLE ).all() self.soft_assert_equal(len(approval_user_project), 1) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_with_same_campaign_uuid(self): data = { "campaigns": ["34ed4c22-e1f9-4184-a06d-f04cbdb97451", "34ed4c22-e1f9-4184-a06d-f04cbdb97451"], "reviewers": [2, ], } response = self.execute_post(self.url.format(501, 44, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 400) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_for_pending_approval_campaign(self): data = { "campaigns": ["34ed4c22-e1f9-4184-a06d-f04cbdb97452", "34ed4c22-e1f9-4184-a06d-f04cbdb97451"], "reviewers": [2, ], } response = self.execute_post(self.url.format(501, 44, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 400) self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_for_disapproved_campaign(self): data = { "campaigns": ["34ed4c22-e1f9-4184-a06d-f04cbdb97333"], "reviewers": [2, ], } response = self.execute_post(self.url.format(501, 44, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 200) data = response.json approval_request_id = data.get("approvalRequestId") approval_request = db.session.query(ApprovalRequest).filter( ApprovalRequest.id == approval_request_id).one_or_none() self.soft_assert(self.assertIsNotNone, approval_request) self.soft_assert_equal(approval_request.requester_id, 3) self.soft_assert_equal(approval_request.reviewer_id, 2) self.soft_assert_equal(approval_request.media_plan_id, 44) approval_request_campaigns = approval_request.approval_request_campaigns self.soft_assert_equal(len(approval_request_campaigns), 1) self.soft_assert_equal(approval_request_campaigns[0].campaign_uuid, uuid.UUID("34ed4c22-e1f9-4184-a06d-f04cbdb97333")) self.soft_assert_equal(approval_request_campaigns[0].is_approved, False) self.soft_assert_equal(approval_request_campaigns[0].is_deleted, False) campaign = approval_request_campaigns[0].campaign self.soft_assert_equal(campaign.workflow_status, 'Pending Approval') self.assert_all() @patch("notifications.notifications_center.NotificationsCenter.notify", new=mocked_email_send) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_send_approval_request_should_update_project_last_edit_at(self): data = { "campaigns": [ "34ed4c22-e1f9-4184-a06d-f04cbdb97666", ], "reviewers": [2, ], } response = self.execute_post(self.url.format(500, 42, "approval-request"), data=json.dumps(data), headers=self.headers ) self.assertEqual(response.status_code, 200) project = db.session.query(Project).filter(Project.id == 500).one_or_none() self.assertNotEqual(project.last_edit_at, "2019-07-22 16:14:55.355226") @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) class TestCampaignWorkflowStatus(BaseTestCase): url = "/projects/{}/media-plans/{}/campaigns/{}" def setUp(self): super().setUp() FixtureLoader(models=models).import_as_sql(fixture) def get_campaign(self, uuid): return db.session.query(Campaign).filter(Campaign.uuid == uuid).one_or_none() def test_campaign_transition(self): campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97455") campaign.pending_approval_transition_handler.set() self.assertEqual(campaign.workflow_status, "Pending Approval") self.assert_all() def test_campaign_change_workflow_status(self): data = { "platforms": [2], "objective": 1, "territories": [3], "workflowStatus": "Booked" } response = self.execute_patch( self.url.format(501, 44, "34ed4c22-e1f9-4184-a06d-f04cbdb97453"), data=json.dumps(data), headers=self.headers ) data = response.json self.assertEqual(200, response.status_code) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97453") self.assertEqual(campaign.workflow_status, "Booked") self.assert_all() def test_campaign_change_workflow_status_from_draft_to_disapproved(self): data = { "platforms": [2], "objective": 1, "territories": [3], "workflowStatus": "Disapproved" } response = self.execute_patch( self.url.format(501, 44, "34ed4c22-e1f9-4184-a06d-f04cbdb97455"), data=json.dumps(data), headers=self.headers ) self.assertEqual(200, response.status_code) self.assert_all() def test_workflow_status_doesnt_change(self): data = { "platforms": [2], "objective": 1, "territories": [3], "workflowStatus": "Approved" } response = self.execute_patch( self.url.format(501, 44, "34ed4c22-e1f9-4184-a06d-f04cbdb97453"), data=json.dumps(data), headers=self.headers ) data = response.json self.assertEqual(200, response.status_code) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97453") self.assertEqual(campaign.workflow_status, "Approved") self.assert_all() @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_status_transition_from_pending_approval_to_draft_remove_approval_request(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97452", "columns": { "workflowStatus": "Draft", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) approval_request_campaign = ( db.session.query(ApprovalRequestCampaign) .filter(ApprovalRequestCampaign.id==4242).one_or_none() ) self.assertEqual(approval_request_campaign.is_live, False) @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_status_transition_from_pending_approval_to_booked_remove_approval_request(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97452", "columns": { "workflowStatus": "Booked", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) approval_request_campaign = ( db.session.query(ApprovalRequestCampaign) .filter(ApprovalRequestCampaign.id==4242).one_or_none() ) self.assertEqual(approval_request_campaign.is_live, False) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97452") self.assertEqual(campaign.workflow_status, "Booked") @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_status_transition_from_pending_approval_to_completed_remove_approval_request(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97452", "columns": { "workflowStatus": "Completed", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) approval_request_campaign = ( db.session.query(ApprovalRequestCampaign) .filter(ApprovalRequestCampaign.id==4242).one_or_none() ) self.assertEqual(approval_request_campaign.is_live, False) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97452") self.assertEqual(campaign.workflow_status, "Completed") @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_status_transition_from_pending_approval_to_paused_remove_approval_request(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97452", "columns": { "workflowStatus": "Paused", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) approval_request_campaign = ( db.session.query(ApprovalRequestCampaign) .filter(ApprovalRequestCampaign.id == 4242).one_or_none() ) self.assertEqual(approval_request_campaign.is_live, False) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97452") self.assertEqual(campaign.workflow_status, "Paused") @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_status_transition_from_pending_approval_to_live_remove_approval_request(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97452", "columns": { "workflowStatus": "Live", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) approval_request_campaign = ( db.session.query(ApprovalRequestCampaign) .filter(ApprovalRequestCampaign.id == 4242).one_or_none() ) self.assertEqual(approval_request_campaign.is_live, False) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97452") self.assertEqual(campaign.workflow_status, "Live") @patch("auth.atlas.atlas_token_decoder.jwt.decode", requesterMock.jwt_decode) def test_manual_approval_transitions(self): url = "/projects/501/media-plans/44/campaigns/bulk-action" data = { "update": [ { "campaignUuid": "34ed4c22-e1f9-4184-a06d-f04cbdb97455", "columns": { "workflowStatus": "Pending Approval", } } ], } response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97455") self.assertEqual(campaign.workflow_status, "Pending Approval") data["update"][0]["columns"]["workflowStatus"] = "Approved" response = self.execute_post(url, data=json.dumps(data), headers=self.headers) self.assertEqual(200, response.status_code) campaign = self.get_campaign("34ed4c22-e1f9-4184-a06d-f04cbdb97455") self.assertEqual(campaign.workflow_status, "Approved")