cr-connect-workflow/tests/test_approvals_api.py

import json
from tests.base_test import BaseTest

from crc import app, db, session
from crc.models.approval import ApprovalModel, ApprovalSchema, ApprovalStatus


class TestApprovals(BaseTest):
    def setUp(self):
        """Initial setup shared by all TestApprovals tests"""
        self.load_example_data()
        self.study = self.create_study()
        self.workflow = self.create_workflow('random_fact')
        # TODO: Move to base_test as a helper
        self.approval = ApprovalModel(
            study=self.study,
            workflow=self.workflow,
            approver_uid='arc93',
            status=ApprovalStatus.PENDING.value,
            version=1
        )
        session.add(self.approval)

        self.approval_2 = ApprovalModel(
            study=self.study,
            workflow=self.workflow,
            approver_uid='dhf8r',
            status=ApprovalStatus.PENDING.value,
            version=1
        )
        session.add(self.approval_2)

        session.commit()

    def test_list_approvals_per_approver(self):
        """Only approvals associated with approver should be returned"""
        approver_uid = self.approval_2.approver_uid
        rv = self.app.get(f'/v1.0/approval', headers=self.logged_in_headers())
        self.assert_success(rv)

        response = json.loads(rv.get_data(as_text=True))

        # Stored approvals are 2
        approvals_count = ApprovalModel.query.count()
        self.assertEqual(approvals_count, 2)

        # but Dan's approvals should be only 1
        self.assertEqual(len(response), 1)

        # Confirm approver UID matches returned payload
        approval = ApprovalSchema().load(response[0])
        self.assertEqual(approval.approver['uid'], approver_uid)

    def test_list_approvals_per_admin(self):
        """All approvals will be returned"""
        rv = self.app.get('/v1.0/approval?everything=true', headers=self.logged_in_headers())
        self.assert_success(rv)

        response = json.loads(rv.get_data(as_text=True))

        # Returned approvals should match what's in the db
        approvals_count = ApprovalModel.query.count()
        response_count = len(response)
        self.assertEqual(2, response_count)

        rv = self.app.get('/v1.0/approval', headers=self.logged_in_headers())
        self.assert_success(rv)
        response = json.loads(rv.get_data(as_text=True))
        response_count = len(response)
        self.assertEqual(1, response_count)

    def test_update_approval_fails_if_not_the_approver(self):
        approval = session.query(ApprovalModel).filter_by(approver_uid='arc93').first()
        data = {'id': approval.id,
                "approver_uid": "dhf8r",
                'message': "Approved.  I like the cut of your jib.",
                'status': ApprovalStatus.APPROVED.value}

        self.assertEqual(approval.status, ApprovalStatus.PENDING.value)

        rv = self.app.put(f'/v1.0/approval/{approval.id}',
                          content_type="application/json",
                          headers=self.logged_in_headers(),  # As dhf8r
                          data=json.dumps(data))
        self.assert_failure(rv)

    def test_accept_approval(self):
        approval = session.query(ApprovalModel).filter_by(approver_uid='dhf8r').first()
        data = {'id': approval.id,
                "approver_uid": "dhf8r",
                'message': "Approved.  I like the cut of your jib.",
                'status': ApprovalStatus.APPROVED.value}

        self.assertEqual(approval.status, ApprovalStatus.PENDING.value)

        rv = self.app.put(f'/v1.0/approval/{approval.id}',
                          content_type="application/json",
                          headers=self.logged_in_headers(),  # As dhf8r
                          data=json.dumps(data))
        self.assert_success(rv)

        session.refresh(approval)

        # Updated record should now have the data sent to the endpoint
        self.assertEqual(approval.message, data['message'])
        self.assertEqual(approval.status, ApprovalStatus.APPROVED.value)

    def test_decline_approval(self):
        approval = session.query(ApprovalModel).filter_by(approver_uid='dhf8r').first()
        data = {'id': approval.id,
                "approver_uid": "dhf8r",
                'message': "Approved.  I find the cut of your jib lacking.",
                'status': ApprovalStatus.DECLINED.value}

        self.assertEqual(approval.status, ApprovalStatus.PENDING.value)

        rv = self.app.put(f'/v1.0/approval/{approval.id}',
                          content_type="application/json",
                          headers=self.logged_in_headers(),  # As dhf8r
                          data=json.dumps(data))
        self.assert_success(rv)

        session.refresh(approval)

        # Updated record should now have the data sent to the endpoint
        self.assertEqual(approval.message, data['message'])
        self.assertEqual(approval.status, ApprovalStatus.DECLINED.value)