Merge pull request #2147 from uktrade/uat

Production release

api/applications/caseworker/permissions.py

@@ -0,0 +1,27 @@
+from rest_framework import permissions
+
+from api.core.constants import GovPermissions
+from api.staticdata.statuses.enums import CaseStatusEnum
+from lite_routing.routing_rules_internal.enums import TeamIdEnum
+
+
+# TODO: Review where django-rules can help simplify this
+class CaseStatusCaseworkerChangeable(permissions.BasePermission):
+    def has_object_permission(self, request, view, application):
+        new_status = request.data.get("status")
+        original_status = application.status.status
+        user = request.user.govuser
+
+        if new_status == CaseStatusEnum.FINALISED:
+            lu_user = str(user.team.id) == TeamIdEnum.LICENSING_UNIT
+            if lu_user and user.has_permission(GovPermissions.MANAGE_LICENCE_FINAL_ADVICE):
+                return True
+            return False
+
+        if new_status == CaseStatusEnum.APPLICANT_EDITING:
+            return False
+
+        if CaseStatusEnum.is_terminal(original_status) and not user.has_permission(GovPermissions.REOPEN_CLOSED_CASES):
+            return False
+
+        return True

api/applications/caseworker/serializers.py

@@ -0,0 +1,9 @@
+from rest_framework import serializers
+
+from api.staticdata.statuses.enums import CaseStatusEnum
+
+
+class ApplicationChangeStatusSerializer(serializers.Serializer):
+
+    status = serializers.ChoiceField(choices=CaseStatusEnum.all())
+    note = serializers.CharField(required=False, allow_blank=True, allow_null=True, max_length=2000)

api/applications/caseworker/tests/test_permissions.py

@@ -0,0 +1,93 @@
+from unittest import mock
+from parameterized import parameterized
+
+from api.applications.caseworker.permissions import CaseStatusCaseworkerChangeable
+from api.applications.tests.factories import StandardApplicationFactory
+from api.core.constants import GovPermissions
+from api.staticdata.statuses.enums import CaseStatusEnum
+from api.staticdata.statuses.models import CaseStatus
+from api.teams.models import Team
+from api.teams.enums import TeamIdEnum
+from api.users.models import Role
+
+
+from test_helpers.clients import DataTestClient
+
+permitted_statuses = list(
+    set(CaseStatusEnum.all()) - set(CaseStatusEnum.terminal_statuses()) - set([CaseStatusEnum.APPLICANT_EDITING])
+)
+
+
+class TestChangeStatusCaseworkerChangeable(DataTestClient):
+
+    def setUp(self):
+        super().setUp()
+        self.application = StandardApplicationFactory(status=CaseStatus.objects.get(status=CaseStatusEnum.SUBMITTED))
+        self.permission_obj = CaseStatusCaseworkerChangeable()
+
+    @parameterized.expand(permitted_statuses)
+    def test_has_object_permission_permitted(self, case_status):
+        mock_request = mock.Mock()
+        mock_request.data = {"status": case_status}
+        mock_request.user = self.gov_user.baseuser_ptr
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is True
+
+    @parameterized.expand(CaseStatusEnum.terminal_statuses())
+    def test_has_object_permission_original_status_terminal_no_user_permission(self, original_status):
+        self.application.status = CaseStatus.objects.get(status=original_status)
+        self.application.save()
+
+        role = Role.objects.create(name="test")
+        role.permissions.set([])
+        self.gov_user.role = role
+        self.gov_user.save()
+
+        mock_request = mock.Mock()
+        mock_request.user = self.gov_user.baseuser_ptr
+        mock_request.data = {"status": CaseStatusEnum.OGD_ADVICE}
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is False
+
+    @parameterized.expand(CaseStatusEnum.terminal_statuses())
+    def test_has_object_permission_original_status_terminal_user_permitted(self, original_status):
+        self.application.status = CaseStatus.objects.get(status=original_status)
+        self.application.save()
+
+        role = Role.objects.create(name="test")
+        role.permissions.set([GovPermissions.REOPEN_CLOSED_CASES.name])
+        self.gov_user.role = role
+        self.gov_user.save()
+
+        mock_request = mock.Mock()
+        mock_request.user = self.gov_user.baseuser_ptr
+        mock_request.data = {"status": CaseStatusEnum.OGD_ADVICE}
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is True
+
+    def test_has_object_permission_new_status_applicant_editing(self):
+        mock_request = mock.Mock()
+        mock_request.user = self.gov_user.baseuser_ptr
+        mock_request.data = {"status": CaseStatusEnum.APPLICANT_EDITING}
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is False
+
+    def test_has_object_permission_new_status_finalised_user_permitted(self):
+        self.gov_user.team = Team.objects.get(id=TeamIdEnum.LICENSING_UNIT)
+        role = Role.objects.create(name="test")
+        role.permissions.set([GovPermissions.MANAGE_LICENCE_FINAL_ADVICE.name])
+        self.gov_user.role = role
+        self.gov_user.save()
+
+        mock_request = mock.Mock()
+        mock_request.user = self.gov_user.baseuser_ptr
+        mock_request.data = {"status": CaseStatusEnum.FINALISED}
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is True
+
+    def test_has_object_permission_new_status_finalised_user_not_permitted(self):
+        self.gov_user.team = Team.objects.get(id=TeamIdEnum.LICENSING_UNIT)
+        role = Role.objects.create(name="test")
+        role.permissions.set([])
+        self.gov_user.role = role
+        self.gov_user.save()
+
+        mock_request = mock.Mock()
+        mock_request.user = self.gov_user.baseuser_ptr
+        mock_request.data = {"status": CaseStatusEnum.FINALISED}
+        assert self.permission_obj.has_object_permission(mock_request, None, self.application) is False

api/applications/caseworker/urls.py

@@ -0,0 +1,9 @@
+from django.urls import path
+
+from api.applications.caseworker.views import applications
+
+app_name = "caseworker_applications"
+
+urlpatterns = [
+    path("<uuid:pk>/status/", applications.ApplicationChangeStatus.as_view(), name="change_status"),
+]

api/applications/caseworker/views/applications.py

@@ -0,0 +1,51 @@
+from django.core.exceptions import ObjectDoesNotExist
+from django.http import Http404, JsonResponse
+from django.db import transaction
+from rest_framework.generics import GenericAPIView
+from rest_framework import status
+
+from api.applications.caseworker.permissions import CaseStatusCaseworkerChangeable
+from api.applications.caseworker.serializers import ApplicationChangeStatusSerializer
+from api.applications.helpers import get_application_view_serializer
+from api.applications.libraries.get_applications import get_application
+from api.core.exceptions import NotFoundError
+from api.core.authentication import GovAuthentication
+from api.core.permissions import CaseInCaseworkerOperableStatus
+from api.staticdata.statuses.libraries.get_case_status import get_case_status_by_status
+
+
+class ApplicationChangeStatus(GenericAPIView):
+    authentication_classes = (GovAuthentication,)
+    permission_classes = [
+        CaseInCaseworkerOperableStatus,
+        CaseStatusCaseworkerChangeable,
+    ]
+    serializer_class = ApplicationChangeStatusSerializer
+
+    def setup(self, request, *args, **kwargs):
+        super().setup(request, *args, **kwargs)
+        try:
+            self.application = get_application(self.kwargs["pk"])
+        except (ObjectDoesNotExist, NotFoundError):
+            raise Http404()
+
+    def get_object(self):
+        self.check_object_permissions(self.request, self.application)
+        return self.application
+
+    def get_case(self):
+        return self.application
+
+    @transaction.atomic
+    def post(self, request, pk):
+        application = self.get_object()
+        serializer = self.serializer_class(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        data = serializer.data
+        application.change_status(request.user, get_case_status_by_status(data["status"]), data["note"])
+
+        response_data = get_application_view_serializer(application)(
+            application, context={"user_type": request.user.type}
+        ).data
+
+        return JsonResponse(data=response_data, status=status.HTTP_200_OK)

api/applications/caseworker/views/tests/test_applications.py

@@ -0,0 +1,86 @@
+from parameterized import parameterized
+
+from django.urls import reverse
+from rest_framework import status
+
+from api.audit_trail.models import Audit
+from api.audit_trail.enums import AuditType
+from api.applications.tests.factories import StandardApplicationFactory
+from api.staticdata.statuses.enums import CaseStatusEnum
+from api.staticdata.statuses.models import CaseStatus
+
+from test_helpers.clients import DataTestClient
+
+
+class TestChangeStatus(DataTestClient):
+
+    def setUp(self):
+        super().setUp()
+        self.application = StandardApplicationFactory(organisation=self.organisation)
+        self.url = reverse(
+            "caseworker_applications:change_status",
+            kwargs={
+                "pk": str(self.application.pk),
+            },
+        )
+
+    @parameterized.expand(
+        list(set(CaseStatusEnum.caseworker_operable_statuses()) - set(CaseStatusEnum.terminal_statuses()))
+    )
+    def test_change_status_success(self, case_status):
+        self.application.status = CaseStatus.objects.get(status=case_status)
+        self.application.save()
+        response = self.client.post(self.url, **self.gov_headers, data={"status": CaseStatusEnum.SUBMITTED})
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        application_id = response.json().get("id")
+        self.assertEqual(application_id, str(self.application.id))
+        self.application.refresh_from_db()
+        self.assertEqual(self.application.status.status, CaseStatusEnum.SUBMITTED)
+
+    def test_change_status_success_with_note(self):
+        self.application.status = CaseStatus.objects.get(status=CaseStatusEnum.SUBMITTED)
+        self.application.save()
+        response = self.client.post(
+            self.url, **self.gov_headers, data={"status": CaseStatusEnum.INITIAL_CHECKS, "note": "some reason"}
+        )
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        application_id = response.json().get("id")
+        self.assertEqual(application_id, str(self.application.id))
+        self.application.refresh_from_db()
+        self.assertEqual(self.application.status.status, CaseStatusEnum.INITIAL_CHECKS)
+        audit_entry = Audit.objects.get(verb=AuditType.UPDATED_STATUS)
+        assert audit_entry.payload == {
+            "additional_text": "some reason",
+            "status": {
+                "new": CaseStatusEnum.INITIAL_CHECKS,
+                "old": CaseStatusEnum.SUBMITTED,
+            },
+        }
+
+    def test_change_status_not_permitted_status(self):
+        self.application.status = CaseStatus.objects.get(status=CaseStatusEnum.SUBMITTED)
+        self.application.save()
+        response = self.client.post(self.url, **self.gov_headers, data={"status": CaseStatusEnum.APPLICANT_EDITING})
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.application.refresh_from_db()
+        self.assertEqual(self.application.status.status, CaseStatusEnum.SUBMITTED)
+
+    def test_change_status_not_caseworker_operable(self):
+        self.application.status = CaseStatus.objects.get(status=CaseStatusEnum.SUPERSEDED_BY_EXPORTER_EDIT)
+        self.application.save()
+        response = self.client.post(self.url, **self.gov_headers, data={"status": CaseStatusEnum.OGD_ADVICE})
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
+        self.application.refresh_from_db()
+        self.assertEqual(self.application.status.status, CaseStatusEnum.SUPERSEDED_BY_EXPORTER_EDIT)
+
+    def test_change_status_application_not_found(self):
+        self.application.delete()
+        response = self.client.post(self.url, **self.gov_headers, data={"status": CaseStatusEnum.APPLICANT_EDITING})
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+    def test_change_status_exporter_not_permitted(self):
+        self.application.organisation = self.exporter_user.organisation
+        self.application.save()
+
+        response = self.client.post(self.url, **self.exporter_headers, data={"status": CaseStatusEnum.SUBMITTED})
+        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

api/applications/exporter/permissions.py

@@ -0,0 +1,21 @@
+from rest_framework import permissions
+
+from api.staticdata.statuses.enums import CaseStatusEnum
+
+
+# TODO: Review where django-rules can help simplify this
+class CaseStatusExporterChangeable(permissions.BasePermission):
+    def has_object_permission(self, request, view, application):
+        new_status = request.data.get("status")
+        original_status = application.status.status
+
+        if new_status == CaseStatusEnum.WITHDRAWN and not CaseStatusEnum.is_terminal(original_status):
+            return True
+
+        if new_status == CaseStatusEnum.SURRENDERED and original_status == CaseStatusEnum.FINALISED:
+            return True
+
+        if new_status == CaseStatusEnum.APPLICANT_EDITING and CaseStatusEnum.can_invoke_major_edit(original_status):
+            return True
+
+        return False

api/applications/exporter/serializers.py

@@ -0,0 +1,9 @@
+from rest_framework import serializers
+
+from api.staticdata.statuses.enums import CaseStatusEnum
+
+
+class ApplicationChangeStatusSerializer(serializers.Serializer):
+
+    status = serializers.ChoiceField(choices=CaseStatusEnum.all())
+    note = serializers.CharField(required=False, allow_blank=True, allow_null=True, max_length=2000)