Department clear trash (#3727)

* department admin clear repo trash

* department admin clear repo trash API test

seahub/api2/endpoints/repo_trash.py

@@ -16,6 +16,8 @@ from seahub.signals import clean_up_repo_trash
 from seahub.utils.timeutils import timestamp_to_isoformat_timestr
 from seahub.utils.repo import get_repo_owner
 from seahub.views import check_folder_permission
+from seahub.group.utils import is_group_admin
+from seahub.api2.endpoints.group_owned_libraries import get_group_id_by_repo_owner
 
 from seaserv import seafile_api
 from pysearpc import SearpcError
@@ -129,7 +131,8 @@ class RepoTrash(APIView):
         """ Clean library's trash.
 
         Permission checking:
-        1. only repo owner can perform this action.
+        1. repo owner can perform this action.
+        2. is group admin.
         """
 
         # argument check
@@ -148,7 +151,17 @@ class RepoTrash(APIView):
         # permission check
         username = request.user.username
         repo_owner = get_repo_owner(request, repo_id)
-        if username != repo_owner or not config.ENABLE_USER_CLEAN_TRASH:
+        if not config.ENABLE_USER_CLEAN_TRASH:
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        if '@seafile_group' in repo_owner:
+            group_id = get_group_id_by_repo_owner(repo_owner)
+            if not is_group_admin(group_id, username):
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+        else:
+            if username != repo_owner:
                 error_msg = 'Permission denied.'
                 return api_error(status.HTTP_403_FORBIDDEN, error_msg)
 

tests/api/endpoints/test_repo_trash.py

@@ -3,9 +3,10 @@ import json
 
 from django.core.urlresolvers import reverse
 
-from seaserv import seafile_api
+from seaserv import seafile_api, ccnet_api
 from seahub.test_utils import BaseTestCase
 
+from seahub.group.utils import is_group_admin
 from tests.common.utils import randstring
 
 class RepoTrashTest(BaseTestCase):
@@ -26,9 +27,13 @@ class RepoTrashTest(BaseTestCase):
 
         self.url = reverse('api-v2.1-repo-trash', args=[self.repo_id])
 
+        self.tmp_user = self.create_user(
+            'user_%s@test.com' % randstring(4), is_staff=False)
+
     def tearDown(self):
         self.remove_repo()
         self.remove_group()
+        self.remove_user(self.tmp_user.username)
 
     def test_can_get(self):
 
@@ -83,6 +88,56 @@ class RepoTrashTest(BaseTestCase):
         json_resp = json.loads(resp.content)
         assert len(json_resp['data']) == 0
 
+    def test_can_clean_department_repo_trash(self):
+        # create a department
+        group_id = ccnet_api.create_group('department_test', 'system admin', parent_group_id=-1)
+        seafile_api.set_group_quota(group_id, -2)
+        repo_id = seafile_api.add_group_owned_repo(group_id, 'dep_test', 'rw')
+        repo_owner = seafile_api.get_repo_owner(repo_id)
+        assert '@seafile_group' in repo_owner
+        group_repos = seafile_api.get_repos_by_group(group_id)
+        assert len(group_repos) == 1
+        group = ccnet_api.get_group(group_id)
+
+        # department add user
+        ccnet_api.group_add_member(group_id, group.creator_name, self.user_name)
+        ccnet_api.group_add_member(group_id, group.creator_name, self.tmp_user.username)
+        ccnet_api.group_set_admin(group_id, self.user_name)
+        ccnet_api.group_unset_admin(group_id, self.tmp_user.username)
+        assert is_group_admin(group_id, self.user_name)
+        assert not is_group_admin(group_id, self.tmp_user.username)
+
+        file_name = 'dep_test.txt'
+        self.create_file(
+            repo_id=repo_id, parent_dir='/', filename=file_name, username=self.user_name)
+
+        # delete a file first
+        seafile_api.del_file(repo_id, '/', file_name, self.user_name)
+
+        # get trash item count
+        self.login_as(self.user)
+        resp = self.client.get(reverse('api-v2.1-repo-trash', args=[repo_id]))
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['data']) > 0
+
+        # department member can not clean trash
+        self.logout()
+        self.login_as(self.tmp_user)
+        resp = self.client.delete(self.url)
+        self.assertEqual(403, resp.status_code)
+
+        # department admin can clean library trash
+        self.logout()
+        self.login_as(self.user)
+        ccnet_api.group_set_admin(group_id, self.user_name)
+        resp = self.client.delete(self.url)
+        self.assertEqual(200, resp.status_code)
+
+        # get trash item count again
+        resp = self.client.get(self.url)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['data']) == 0
+
     def test_can_not_clean_with_invalid_user_permission(self):
 
         self.login_as(self.admin)