Merge branch 'trash-bin' into beta

kpi/serializers/v2/asset.py

@@ -3,7 +3,7 @@
 
 import json
 import re
-from distutils import util
+
 
 from constance import config
 from django.conf import settings
@@ -29,6 +29,7 @@
     ASSET_STATUS_PRIVATE,
     ASSET_STATUS_PUBLIC,
     ASSET_STATUS_SHARED,
+    ASSET_TYPE_SURVEY,
     ASSET_TYPES,
     ASSET_TYPE_COLLECTION,
     PERM_CHANGE_ASSET,
@@ -114,6 +115,7 @@ def validate_payload(self, payload: dict) -> dict:
             asset_uids = []
 
         self._has_perms(payload, asset_uids)
+        self._validate_asset_types(payload, asset_uids)
 
         return payload
 
@@ -256,6 +258,21 @@ def _validate_action(self, payload: dict):
         ):
             raise exceptions.PermissionDenied()
 
+    def _validate_asset_types(self, payload: dict, asset_uids: list[str]):
+        delete_request, put_back_ = self._get_action_type_and_direction(payload)
+
+        if put_back_ or delete_request or not asset_uids:
+            return
+
+        if Asset.objects.filter(
+            asset_type=ASSET_TYPE_SURVEY,
+            uid__in=asset_uids,
+            _deployment_data={},
+        ).exists():
+            raise serializers.ValidationError(
+                t('Draft projects cannot be archived')
+            )
+
     def _validate_confirm(self, payload: dict):
 
         if not payload.get('confirm'):

kpi/tests/api/v2/test_api_asset_delete.py → kpi/tests/api/v2/test_api_asset_bulk_actions.py

@@ -4,17 +4,19 @@
 from rest_framework.response import Response
 
 from kpi.constants import (
+    ASSET_TYPE_SURVEY,
     PERM_CHANGE_ASSET,
     PERM_MANAGE_ASSET,
-    PERM_VIEW_ASSET
+    PERM_VIEW_ASSET,
 )
 from kpi.models import Asset
 from kpi.tests.base_test_case import BaseTestCase
 from kpi.urls.router_api_v2 import URL_NAMESPACE as ROUTER_URL_NAMESPACE
 from kpi.utils.object_permission import get_anonymous_user
 
 
-class AssetBulkDeleteAPITestCase(BaseTestCase):
+class BaseAssetBulkActionsTestCase(BaseTestCase):
+
     fixtures = ['test_data']
     URL_NAMESPACE = ROUTER_URL_NAMESPACE
 
@@ -41,6 +43,7 @@ def _add_one_asset_for_someuser(self) -> Asset:
         asset = Asset.objects.create(
             owner=User.objects.get(username='someuser'),
             content=content,
+            asset_type=ASSET_TYPE_SURVEY
         )
         asset.deploy(backend='mock', active=True)
         return asset
@@ -95,6 +98,9 @@ def _login_user(self, userpass: str):
         self.client.logout()
         self.client.login(username=userpass, password=userpass)
 
+
+class AssetBulkArchiveAPITestCase(BaseAssetBulkActionsTestCase):
+
     def test_archive_all_with_confirm_true(self):
         # Create multiple assets
         self._login_user('someuser')
@@ -187,7 +193,7 @@ def test_other_user_cannot_archive_others_assets(self):
         assert detail_response.status_code == status.HTTP_200_OK
         assert detail_response.data['deployment__active'] is True
 
-    def test_anonymous_archive_public(self):
+    def test_anonymous_cannot_archive_public(self):
         asset = self._add_one_asset_for_someuser()
         anonymous = get_anonymous_user()
         asset.assign_perm(anonymous, PERM_VIEW_ASSET)
@@ -200,7 +206,56 @@ def test_anonymous_archive_public(self):
         assert detail_response.status_code == status.HTTP_200_OK
         assert detail_response.data['deployment__active'] is True
 
-    def test_anonymous_delete_public(self):
+    def test_project_editor_cannot_archive_project(self):
+        editor = User.objects.get(username='anotheruser')
+        asset = self._add_one_asset_for_someuser()
+        asset.assign_perm(editor, PERM_CHANGE_ASSET)
+        self._login_user('anotheruser')
+        response = self._create_send_payload([asset.uid], 'archive')
+        assert response.status_code == status.HTTP_403_FORBIDDEN
+
+        asset.refresh_from_db()
+        assert asset.deployment.active is True
+        assert asset.pending_delete is False
+
+        # Another can still access the project
+        detail_response = self._get_asset_detail_results(asset.uid)
+        assert detail_response.status_code == status.HTTP_200_OK
+        assert detail_response.data['deployment__active'] is True
+
+    def test_project_manager_can_archive_project(self):
+        manager = User.objects.get(username='anotheruser')
+        asset = self._add_one_asset_for_someuser()
+        asset.assign_perm(manager, PERM_MANAGE_ASSET)
+        self._login_user('anotheruser')
+        response = self._create_send_payload([asset.uid], 'archive')
+        print('RESPONSE', response.data)
+        assert response.status_code == status.HTTP_200_OK
+
+        asset.refresh_from_db()
+        assert asset.deployment.active is False
+        assert asset.pending_delete is False
+
+        detail_response = self._get_asset_detail_results(asset.uid)
+        assert detail_response.status_code == status.HTTP_200_OK
+        assert detail_response.data['deployment__active'] is False
+
+    def test_user_cannot_archive_drafts(self):
+        self._login_user('someuser')
+        deployed_asset = self._add_one_asset_for_someuser()
+        asset = Asset.objects.create(
+            owner=User.objects.get(username='someuser'),
+            asset_type=ASSET_TYPE_SURVEY
+        )
+        response = self._create_send_payload(
+            [deployed_asset.uid, asset.uid], 'archive'
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+
+
+class AssetBulkDeleteAPITestCase(BaseAssetBulkActionsTestCase):
+
+    def test_anonymous_cannot_delete_public(self):
         asset = self._add_one_asset_for_someuser()
         anonymous = get_anonymous_user()
         asset.assign_perm(anonymous, PERM_VIEW_ASSET)
@@ -240,7 +295,7 @@ def test_delete_all_assets_with_confirm_true(self):
             detail_response = self._get_asset_detail_results(deleted_asset.uid)
             assert detail_response.status_code == status.HTTP_404_NOT_FOUND
 
-    def test_delete_all_assets_without_confirm_true(self):
+    def test_cannot_delete_all_assets_without_confirm_true(self):
         self._login_user('someuser')
         asset_uids = []
         for i in range(3):
@@ -301,39 +356,6 @@ def test_other_user_cannot_delete_others_assets(self):
         assert detail_response.status_code == status.HTTP_200_OK
         assert detail_response.data['deployment__active'] is True
 
-    def test_project_editor_cannot_archive_project(self):
-        editor = User.objects.get(username='anotheruser')
-        asset = self._add_one_asset_for_someuser()
-        asset.assign_perm(editor, PERM_CHANGE_ASSET)
-        self._login_user('anotheruser')
-        response = self._create_send_payload([asset.uid], 'archive')
-        assert response.status_code == status.HTTP_403_FORBIDDEN
-
-        asset.refresh_from_db()
-        assert asset.deployment.active is True
-        assert asset.pending_delete is False
-
-        # Another can still access the project
-        detail_response = self._get_asset_detail_results(asset.uid)
-        assert detail_response.status_code == status.HTTP_200_OK
-        assert detail_response.data['deployment__active'] is True
-
-    def test_project_manager_can_archive_project(self):
-        manager = User.objects.get(username='anotheruser')
-        asset = self._add_one_asset_for_someuser()
-        asset.assign_perm(manager, PERM_MANAGE_ASSET)
-        self._login_user('anotheruser')
-        response = self._create_send_payload([asset.uid], 'archive')
-        assert response.status_code == status.HTTP_200_OK
-
-        asset.refresh_from_db()
-        assert asset.deployment.active is False
-        assert asset.pending_delete is False
-
-        detail_response = self._get_asset_detail_results(asset.uid)
-        assert detail_response.status_code == status.HTTP_200_OK
-        assert detail_response.data['deployment__active'] is False
-
     def test_superuser_can_undelete(self):
         self._login_user('someuser')
         asset = self._add_one_asset_for_someuser()