Merge pull request #2004 from haiwen/clean-repo-trash

add clean library trash api

seahub/api2/endpoints/repo_trash.py

@@ -13,6 +13,7 @@ from seahub.api2.authentication import TokenAuthentication
 from seahub.api2.utils import api_error
 
 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 seaserv import seafile_api
@@ -121,3 +122,39 @@ class RepoTrash(APIView):
         }
 
         return Response(result)
+
+    def delete(self, request, repo_id, format=None):
+        """ Clean library's trash.
+
+        Permission checking:
+        1. only repo owner can perform this action.
+        """
+
+        # argument check
+        try:
+            keep_days = int(request.data.get('keep_days', 0))
+        except ValueError:
+            error_msg = 'keep_days invalid.'
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        # resource check
+        repo = seafile_api.get_repo(repo_id)
+        if not repo:
+            error_msg = 'Library %s not found.' % repo_id
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        # permission check
+        username = request.user.username
+        repo_owner = get_repo_owner(request, repo_id)
+        if username != repo_owner:
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            seafile_api.clean_up_repo_history(repo_id, keep_days)
+        except Exception as e:
+            logger.error(e)
+            error_msg = 'Internal Server Error'
+            return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
+
+        return Response({'success': True})

tests/api/endpoints/test_repo_trash.py

@@ -6,11 +6,24 @@ from django.core.urlresolvers import reverse
 from seaserv import seafile_api
 from seahub.test_utils import BaseTestCase
 
+from tests.common.utils import randstring
+
 class RepoTrashTest(BaseTestCase):
 
     def setUp(self):
-        self.repo_id = self.repo.id
+
         self.user_name = self.user.username
+        self.admin_name = self.admin.username
+
+        self.repo_id = self.repo.id
+        self.repo_name = self.repo.repo_name
+
+        self.file_path = self.file
+        self.file_name = os.path.basename(self.file_path)
+
+        self.folder_path = self.folder
+        self.folder_name = os.path.basename(self.folder.rstrip('/'))
+
         self.url = reverse('api-v2.1-repo-trash', args=[self.repo_id])
 
     def tearDown(self):
@@ -18,13 +31,60 @@ class RepoTrashTest(BaseTestCase):
         self.remove_group()
 
     def test_can_get(self):
+
         # delete a file first
-        file_name = os.path.basename(self.file)
-        seafile_api.del_file(self.repo_id, '/', file_name, self.user_name)
+        seafile_api.del_file(self.repo_id, '/',
+                self.file_name, self.user_name)
 
         self.login_as(self.user)
+
         resp = self.client.get(self.url)
         self.assertEqual(200, resp.status_code)
+
         json_resp = json.loads(resp.content)
-        assert json_resp['data'][0]['obj_name'] == file_name
+        assert json_resp['data'][0]['obj_name'] == self.file_name
         assert json_resp['data'][0]['is_dir'] == False
+
+    def test_can_not_get_with_invalid_repo_permission(self):
+
+        self.login_as(self.admin)
+
+        resp = self.client.get(self.url)
+        self.assertEqual(403, resp.status_code)
+
+    def test_can_not_get_with_invalid_path_parameter(self):
+
+        invalid_path = randstring(6)
+
+        self.login_as(self.admin)
+
+        resp = self.client.get(self.url + '?path=%s' % invalid_path)
+        self.assertEqual(404, resp.status_code)
+
+    def test_can_clean_library_trash(self):
+
+        # delete a file first
+        seafile_api.del_file(self.repo_id, '/',
+                self.file_name, self.user_name)
+
+        self.login_as(self.user)
+
+        # get trash item count
+        resp = self.client.get(self.url)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['data']) > 0
+
+        # clean library trash
+        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)
+        resp = self.client.delete(self.url)
+        self.assertEqual(403, resp.status_code)