repo commit revert API (#3656)

* repo commit revert API

* repo commit revert API test

seahub/api2/endpoints/repo_commit_revert.py

@@ -0,0 +1,69 @@
+# Copyright (c) 2012-2019 Seafile Ltd.
+# encoding: utf-8
+
+import logging
+
+from rest_framework.authentication import SessionAuthentication
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework import status
+
+from seahub.api2.throttling import UserRateThrottle
+from seahub.api2.authentication import TokenAuthentication
+from seahub.api2.utils import api_error
+from seahub.views import check_folder_permission
+from seaserv import seafile_api
+from seahub.utils.repo import is_repo_owner
+from seahub.constants import PERMISSION_READ_WRITE
+
+logger = logging.getLogger(__name__)
+
+
+class RepoCommitRevertView(APIView):
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAuthenticated,)
+    throttle_classes = (UserRateThrottle,)
+
+    def post(self, request, repo_id, commit_id, format=None):
+        """ revert commit in repo history
+
+        Permission checking:
+        1. only repo owner can perform this action.
+        """
+        username = request.user.username
+
+        # 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)
+
+        commit = seafile_api.get_commit(repo.id, repo.version, commit_id)
+        if not commit:
+            error_msg = 'Commit %s not found.' % commit
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        # permission check
+        if not is_repo_owner(request, repo_id, username) or \
+                check_folder_permission(request, repo_id, '/') != PERMISSION_READ_WRITE:
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        # main
+        if repo.encrypted:
+            ret = seafile_api.is_password_set(repo_id, username)
+            is_decrypted = False if ret == 0 else True
+
+            if not is_decrypted:
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            seafile_api.revert_repo(repo_id, commit_id, username)
+        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})

seahub/urls.py

@@ -55,6 +55,7 @@ from seahub.api2.endpoints.file_tag import FileTagView
 from seahub.api2.endpoints.file_tag import FileTagsView
 from seahub.api2.endpoints.repo_trash import RepoTrash
 from seahub.api2.endpoints.repo_commit_dir import RepoCommitDirView
+from seahub.api2.endpoints.repo_commit_revert import RepoCommitRevertView
 from seahub.api2.endpoints.deleted_repos import DeletedRepos
 from seahub.api2.endpoints.repo_history import RepoHistory
 from seahub.api2.endpoints.repo_set_password import RepoSetPassword
@@ -333,6 +334,7 @@ urlpatterns = [
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/file/new_history/$', NewFileHistoryView.as_view(), name='api-v2.1-new-file-history-view'),
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/dir/$', DirView.as_view(), name='api-v2.1-dir-view'),
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/commits/(?P<commit_id>[0-9a-f]{40})/dir/$', RepoCommitDirView.as_view(), name='api-v2.1-repo-commit-dir'),
+    url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/commits/(?P<commit_id>[0-9a-f]{40})/revert/$', RepoCommitRevertView.as_view(), name='api-v2.1-repo-commit-revert'),
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/dir/detail/$', DirDetailView.as_view(), name='api-v2.1-dir-detail-view'),
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/trash/$', RepoTrash.as_view(), name='api-v2.1-repo-trash'),
     url(r'^api/v2.1/repos/(?P<repo_id>[-0-9a-f]{36})/history/$', RepoHistory.as_view(), name='api-v2.1-repo-history'),

tests/api/endpoints/test_repo_commit_revert.py

@@ -0,0 +1,111 @@
+import os
+import json
+
+from django.core.urlresolvers import reverse
+
+from seaserv import seafile_api
+from seahub.test_utils import BaseTestCase
+
+
+class RepoCommitDirTest(BaseTestCase):
+
+    def setUp(self):
+        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.enc_repo_id = self.enc_repo.id
+        self.enc_repo_name = self.enc_repo
+
+        self.folder_path = self.folder
+        self.folder_name = os.path.basename(self.folder.rstrip('/'))
+
+    def tearDown(self):
+        self.remove_repo()
+        self.remove_group()
+
+    def test_post(self):
+        # delete a file first
+        seafile_api.del_file(self.repo_id, '/', self.file_name, self.user_name)
+
+        self.login_as(self.user)
+
+        # get commit id form trash
+        trash_url = reverse('api-v2.1-repo-trash', args=[self.repo_id])
+        trash_resp = self.client.get(trash_url)
+        self.assertEqual(200, trash_resp.status_code)
+        trash_json_resp = json.loads(trash_resp.content)
+        assert trash_json_resp['data'][0]['obj_name'] == self.file_name
+        assert not trash_json_resp['data'][0]['is_dir']
+        assert trash_json_resp['data'][0]['commit_id']
+        commit_id = trash_json_resp['data'][0]['commit_id']
+
+        dir_url = reverse('api-v2.1-dir-view', args=[self.repo_id])
+        url = reverse('api-v2.1-repo-commit-revert', args=[self.repo_id, commit_id])
+
+        # test can post
+        resp = self.client.get(dir_url)
+        self.assertEqual(200, resp.status_code)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['dirent_list']) == 1
+
+        resp = self.client.post(url)
+        self.assertEqual(200, resp.status_code)
+
+        resp = self.client.get(dir_url)
+        self.assertEqual(200, resp.status_code)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['dirent_list']) == 2
+
+        # test_can_not_post_with_invalid_repo_permission
+        self.logout()
+        self.login_as(self.admin)
+
+        resp = self.client.post(url)
+        self.assertEqual(403, resp.status_code)
+
+    def test_enc_repo_post(self):
+        # delete a file first
+        seafile_api.del_file(self.enc_repo_id, '/', self.file_name, self.user_name)
+
+        self.login_as(self.user)
+
+        # get commit id form trash
+        trash_url = reverse('api-v2.1-repo-trash', args=[self.enc_repo_id])
+        trash_resp = self.client.get(trash_url)
+        self.assertEqual(200, trash_resp.status_code)
+        trash_json_resp = json.loads(trash_resp.content)
+        assert trash_json_resp['data'][0]['obj_name'] == self.file_name
+        assert not trash_json_resp['data'][0]['is_dir']
+        assert trash_json_resp['data'][0]['commit_id']
+        commit_id = trash_json_resp['data'][0]['commit_id']
+
+        dir_url = reverse('api-v2.1-dir-view', args=[self.enc_repo_id])
+        url = reverse('api-v2.1-repo-commit-revert', args=[self.enc_repo_id, commit_id])
+
+        # test can not post without repo decrypted
+        resp = self.client.post(url)
+        self.assertEqual(403, resp.status_code)
+
+        resp = self.client.get(dir_url)
+        self.assertEqual(200, resp.status_code)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['dirent_list']) == 0
+
+        # test can post with repo decrypted
+        decrypted_url = reverse('api-v2.1-repo-set-password', args=[self.enc_repo_id])
+        resp = self.client.post(decrypted_url, data={'password': '123'})
+        self.assertEqual(200, resp.status_code)
+
+        resp = self.client.post(url)
+        self.assertEqual(200, resp.status_code)
+
+        resp = self.client.get(dir_url)
+        self.assertEqual(200, resp.status_code)
+        json_resp = json.loads(resp.content)
+        assert len(json_resp['dirent_list']) == 1