seahub/tests/api/endpoints/test_repos_batch.py

import os
import json
import posixpath
from seaserv import seafile_api, ccnet_api
from django.core.urlresolvers import reverse
from tests.common.utils import randstring
from seahub.test_utils import BaseTestCase
from seahub.utils import normalize_dir_path

try:
    from seahub.settings import LOCAL_PRO_DEV_ENV
except ImportError:
    LOCAL_PRO_DEV_ENV = False


class ReposBatchViewTest(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.repo_id = self.repo.id
        self.group_id = self.group.id

        self.url = reverse('api-v2.1-repos-batch')

    def tearDown(self):
        self.remove_repo()
        self.remove_group()

    def test_can_share_repos_to_user(self):
        tmp_repo_id = self.create_new_repo(self.user_name)

        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'user',
            'username': self.admin_name,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)
        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 1
        assert len(json_resp['failed']) == 0

        # share repo again will failed
        data = {
            'operation': 'share',
            'share_type': 'user',
            'username': self.admin_name,
            'repo_id': [self.repo_id, tmp_repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)
        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 1
        assert len(json_resp['failed']) == 1
        assert self.repo_id in json_resp['failed'][0]['repo_id']

        self.remove_repo(tmp_repo_id)

    def test_can_share_repos_to_group(self):
        tmp_repo_id = self.create_new_repo(self.user_name)

        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'group',
            'group_id': self.group_id,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)
        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 1
        assert len(json_resp['failed']) == 0

        # share repo again will failed
        data = {
            'operation': 'share',
            'share_type': 'group',
            'group_id': self.group_id,
            'repo_id': [self.repo_id, tmp_repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)
        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 1
        assert len(json_resp['failed']) == 1
        assert self.repo_id in json_resp['failed'][0]['repo_id']

        self.remove_repo(tmp_repo_id)

    def test_share_with_invalid_operation(self):
        self.login_as(self.user)

        data = {
            'operation': 'invalid_operation',
            'share_type': 'user',
            'username': self.admin_name,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

        data = {
            'operation': 'invalid_operation',
            'share_type': 'group',
            'group_id': self.group_id,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

    def test_share_with_invalid_share_type(self):
        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'invalid_share_type',
            'username': self.admin_name,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

        data = {
            'operation': 'share',
            'share_type': 'invalid_share_type',
            'group_id': self.group_id,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

    def test_share_with_invalid_permisson(self):
        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'user',
            'permission': 'invalid_permission',
            'username': self.admin_name,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

        data = {
            'operation': 'share',
            'share_type': 'group',
            'permission': 'invalid_permission',
            'group_id': self.group_id,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(400, resp.status_code)

    def test_share_with_invalid_user(self):
        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'user',
            'username': 'invalid@user.com',
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(404, resp.status_code)

    def test_share_with_not_exist_group(self):
        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'group',
            'group_id': -1,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(404, resp.status_code)

    def test_share_with_not_group_member(self):
        tmp_group_id = ccnet_api.create_group(randstring(10), self.admin_name)

        self.login_as(self.user)

        data = {
            'operation': 'share',
            'share_type': 'group',
            'group_id': tmp_group_id,
            'repo_id': [self.repo_id]
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(403, resp.status_code)


class ReposBatchCopyDirView(BaseTestCase):

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username
        self.repo_id = self.repo.id
        self.url = reverse('api-v2.1-repos-batch-copy-dir')

    def tearDown(self):
        self.remove_repo()
        self.remove_group()

    def get_random_path(self):
        return '/%s/%s/%s/' % (randstring(2), \
                randstring(2), randstring(2))

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def test_copy_dir(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # create two folders in src repo
        src_folder_1 = self.get_random_path()
        src_folder_2 = self.get_random_path()
        for path in [src_folder_1, src_folder_2]:
            seafile_api.mkdir_with_parents(self.repo_id,
                    '/', path.strip('/'), self.user_name)

        # share admin's tmp repo to user
        tmp_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(tmp_repo_id, self.admin_name,
                self.user_name, 'rw')

        # create two folders as parent dirs in dst repo for admin user
        dst_folder_1 = self.get_random_path()
        seafile_api.mkdir_with_parents(tmp_repo_id,
                '/', dst_folder_1.strip('/'), self.admin_name)

        dst_folder_2 = '/'

        # copy folders
        data = {
            "src_repo_id": self.repo_id,
            "dst_repo_id": tmp_repo_id,
            "paths": [
                {"src_path": src_folder_1, "dst_path": dst_folder_1},
                {"src_path": src_folder_2, "dst_path": dst_folder_2},
            ]
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 2
        assert len(json_resp['failed']) == 0

        def folder_exist(src_folder, dst_repo_id, dst_folder):
            src_obj_name = os.path.basename(src_folder.rstrip('/'))
            full_dst_folder_path = posixpath.join(dst_folder.strip('/'),
                    src_obj_name.strip('/'))
            full_dst_folder_path = normalize_dir_path(full_dst_folder_path)
            return seafile_api.get_dir_id_by_path(dst_repo_id,
                    full_dst_folder_path) is not None

        assert folder_exist(src_folder_1, tmp_repo_id, dst_folder_1)
        assert folder_exist(src_folder_2, tmp_repo_id, dst_folder_2)

        self.remove_repo(tmp_repo_id)

    def test_copy_dir_with_invalid_repo_permisson(self):

        self.login_as(self.user)

        # create two folders as parent dirs in dst repo for admin user
        tmp_repo_id = self.create_new_repo(self.admin_name)

        # copy folders
        data = {
            "src_repo_id": self.repo_id,
            "dst_repo_id": tmp_repo_id,
            "paths": [
                {"src_path": '/', "dst_path": '/'},
                {"src_path": '/', "dst_path": '/'},
            ]
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(403, resp.status_code)

    def test_copy_dir_with_src_path_is_root_folder(self):

        self.login_as(self.user)

        # create two folders as parent dirs in dst repo for admin user
        tmp_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(tmp_repo_id, self.admin_name,
                self.user_name, 'rw')

        # copy folders
        data = {
            "src_repo_id": self.repo_id,
            "dst_repo_id": tmp_repo_id,
            "paths": [
                {"src_path": '/', "dst_path": '/'},
            ]
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 0
        assert len(json_resp['failed']) == 1

        assert json_resp['failed'][0]['error_msg'] == \
                "The source path can not be '/'."

        self.remove_repo(tmp_repo_id)


class ReposBatchCreateDirViewTest(BaseTestCase):

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username
        self.repo_id = self.repo.id
        self.url = reverse('api-v2.1-repos-batch-create-dir')

    def tearDown(self):
        self.remove_repo()
        self.remove_group()

    def get_random_path(self):
        return '/%s/%s/%s/' % (randstring(2), \
                randstring(2), randstring(2))

    def test_create_dir(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        path_1 = self.get_random_path()
        path_2 = self.get_random_path()
        path_3 = self.get_random_path()

        self.login_as(self.user)

        data = {
            'repo_id': self.repo_id,
            'paths': [path_1, path_2, path_3],
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 3
        assert len(json_resp['failed']) == 0

        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_1) is not None
        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_2) is not None
        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_3) is not None

    def test_create_dir_with_invalid_repo_permission(self):

        # admin has NO permission for user's repo
        self.login_as(self.admin)

        data = {
            'repo_id': self.repo_id,
            'paths': 'path',
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(403, resp.status_code)

    def test_create_dir_with_invalid_folder_permission(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        path_1 = self.get_random_path()
        path_2 = self.get_random_path()
        path_3 = self.get_random_path()

        self.login_as(self.user)

        data = {
            'repo_id': self.repo_id,
            'paths': [path_1, path_2, path_3],
        }
        resp = self.client.post(self.url, data)
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        assert len(json_resp['success']) == 3
        assert len(json_resp['failed']) == 0

        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_1) is not None
        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_2) is not None
        assert seafile_api.get_dir_id_by_path(self.repo_id,
                path_3) is not None


class ReposAsyncBatchCopyItemView(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.src_repo_id = self.repo.id
        self.dst_repo_id = self.create_new_repo(self.user_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)

        self.url = reverse('api-v2.1-repos-async-batch-copy-item')

    def tearDown(self):
        self.remove_repo(self.src_repo_id)
        self.remove_repo(self.dst_repo_id)

    def test_can_copy(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],

            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        task_id = json_resp['task_id']
        assert len(task_id) == 36

#        progress_url = reverse('api-v2.1-query-copy-move-progress') + '?task_id=%s' % task_id
#        count = 1
#        while True:
#            count += 1
#            resp = self.client.get(progress_url)
#            json_resp = json.loads(resp.content)
#            if json_resp['done'] == 1 or count == 10:
#                break
#
#        # items remain in src folder
#        assert seafile_api.get_dir_id_by_path(self.src_repo_id, self.folder_path) is not None
#        assert seafile_api.get_file_id_by_path(self.src_repo_id, self.file_path) is not None
#
#        # items in dst folder
#        assert seafile_api.get_file_id_by_path(self.dst_repo_id, self.file_path) is not None
#        assert seafile_api.get_dir_id_by_path(self.dst_repo_id, self.folder_path) is not None

    def test_copy_with_invalid_parameter(self):

        self.login_as(self.user)

        data = {
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

    def test_copy_with_repo_not_exist(self):

        self.login_as(self.user)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": invalid_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": invalid_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_copy_with_folder_not_exist(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": 'invalid_folder',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": 'invalid_folder',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_copy_with_invalid_repo_permission(self):

        tmp_repo_id = self.create_new_repo(self.admin_name)

        self.login_as(self.user)

        data = {
            "src_repo_id": tmp_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": tmp_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_copy_with_invalid_src_folder_permission(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preivew' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

    def test_copy_with_invalid_dst_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preview' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preivew')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)


class ReposAsyncBatchMoveItemView(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.src_repo_id = self.repo.id
        self.dst_repo_id = self.create_new_repo(self.user_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)

        self.url = reverse('api-v2.1-repos-async-batch-move-item')

    def tearDown(self):
        self.remove_repo(self.src_repo_id)
        self.remove_repo(self.dst_repo_id)

    def test_can_move(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],

            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        json_resp = json.loads(resp.content)
        task_id = json_resp['task_id']
        assert len(task_id) == 36

#        progress_url = reverse('api-v2.1-query-copy-move-progress') + '?task_id=%s' % task_id
#        count = 1
#        while True:
#            count += 1
#            resp = self.client.get(progress_url)
#            json_resp = json.loads(resp.content)
#            if json_resp['done'] == 1 or count == 10:
#                break
#
#        # items NOT in src folder
#        assert seafile_api.get_dir_id_by_path(self.src_repo_id, self.folder_path) is None
#        assert seafile_api.get_file_id_by_path(self.src_repo_id, self.file_path) is None
#
#        # items in dst folder
#        assert seafile_api.get_file_id_by_path(self.dst_repo_id, self.file_path) is not None
#        assert seafile_api.get_dir_id_by_path(self.dst_repo_id, self.folder_path) is not None

    def test_move_with_invalid_parameter(self):

        self.login_as(self.user)

        data = {
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

    def test_move_with_repo_not_exist(self):

        self.login_as(self.user)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": invalid_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": invalid_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_move_with_folder_not_exist(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": 'invalid_folder',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": 'invalid_folder',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_move_with_invalid_repo_permission(self):

        tmp_repo_id = self.create_new_repo(self.admin_name)

        self.login_as(self.user)

        data = {
            "src_repo_id": tmp_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": tmp_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_move_with_invalid_src_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preivew' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

    def test_move_with_invalid_dst_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        # share admin's tmp repo to user with 'preview' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_move_with_locked_file(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin lock file
        admin_file_name = randstring(6)
        seafile_api.post_empty_file(admin_repo_id, '/', admin_file_name,
                self.admin_name)
        seafile_api.lock_file(admin_repo_id, admin_file_name, self.admin_name, 0)

        # user move locked file
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [admin_file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == 'File %s is locked.' % admin_file_name

    def test_move_with_r_permission_sub_folder(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'rw' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin set 'r' sub folder permission
        admin_folder_name = randstring(6)
        seafile_api.post_dir(admin_repo_id, '/', admin_folder_name, self.admin_name)
        seafile_api.add_folder_user_perm(admin_repo_id, '/' +
                admin_folder_name, 'r', self.user_name)

        # user move r permission folder
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [admin_folder_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == "Can't move folder %s, please check its permission." % admin_folder_name


class ReposSyncBatchCopyItemView(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.src_repo_id = self.repo.id
        self.dst_repo_id = self.create_new_repo(self.user_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)

        self.url = reverse('api-v2.1-repos-sync-batch-copy-item')

    def tearDown(self):
        self.remove_repo(self.src_repo_id)
        self.remove_repo(self.dst_repo_id)

    def test_can_copy(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],

            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        # items remain in src folder
        assert seafile_api.get_dir_id_by_path(self.src_repo_id, self.folder_path) is not None
        assert seafile_api.get_file_id_by_path(self.src_repo_id, self.file_path) is not None

        # items in dst folder
        assert seafile_api.get_file_id_by_path(self.dst_repo_id, self.file_path) is not None
        assert seafile_api.get_dir_id_by_path(self.dst_repo_id, self.folder_path) is not None

    def test_copy_with_invalid_parameter(self):

        self.login_as(self.user)

        data = {
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

    def test_copy_with_repo_not_exist(self):

        self.login_as(self.user)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": invalid_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": invalid_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_copy_with_folder_not_exist(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": 'invalid_folder',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": 'invalid_folder',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_copy_with_invalid_repo_permission(self):

        tmp_repo_id = self.create_new_repo(self.admin_name)

        self.login_as(self.user)

        data = {
            "src_repo_id": tmp_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": tmp_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_copy_with_invalid_src_folder_permission(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preivew' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

    def test_copy_with_invalid_dst_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preview' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preivew')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)


class ReposSyncBatchMoveItemView(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.src_repo_id = self.repo.id
        self.dst_repo_id = self.create_new_repo(self.user_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)

        self.url = reverse('api-v2.1-repos-sync-batch-move-item')

    def tearDown(self):
        self.remove_repo(self.src_repo_id)
        self.remove_repo(self.dst_repo_id)

    def test_can_move(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],

            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }

        resp = self.client.post(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        # items NOT in src folder
        assert seafile_api.get_dir_id_by_path(self.src_repo_id, self.folder_path) is None
        assert seafile_api.get_file_id_by_path(self.src_repo_id, self.file_path) is None

        # items in dst folder
        assert seafile_api.get_file_id_by_path(self.dst_repo_id, self.file_path) is not None
        assert seafile_api.get_dir_id_by_path(self.dst_repo_id, self.folder_path) is not None

    def test_move_with_invalid_parameter(self):

        self.login_as(self.user)

        data = {
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

    def test_move_with_repo_not_exist(self):

        self.login_as(self.user)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": invalid_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": invalid_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_move_with_folder_not_exist(self):

        self.login_as(self.user)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": 'invalid_folder',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": 'invalid_folder',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_move_with_invalid_repo_permission(self):

        tmp_repo_id = self.create_new_repo(self.admin_name)

        self.login_as(self.user)

        data = {
            "src_repo_id": tmp_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": tmp_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_move_with_invalid_src_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

        # share admin's tmp repo to user with 'preivew' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": self.src_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

    def test_move_with_invalid_dst_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        if not LOCAL_PRO_DEV_ENV:
            return

        # share admin's tmp repo to user with 'cloud-edit' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'cloud-edit')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

        # share admin's tmp repo to user with 'preview' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'preview')
        data = {
            "src_repo_id": self.src_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [self.folder_name, self.file_name],
            "dst_repo_id": admin_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_move_with_locked_file(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'rw' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin lock file
        admin_file_name = randstring(6)
        seafile_api.post_empty_file(admin_repo_id, '/', admin_file_name,
                self.admin_name)
        seafile_api.lock_file(admin_repo_id, admin_file_name, self.admin_name, 0)

        # user move locked file
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [admin_file_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == 'File %s is locked.' % admin_file_name

    def test_move_with_r_permission_sub_folder(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'rw' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin set 'r' sub folder permission
        admin_folder_name = randstring(6)
        seafile_api.post_dir(admin_repo_id, '/', admin_folder_name, self.admin_name)
        seafile_api.add_folder_user_perm(admin_repo_id, '/' +
                admin_folder_name, 'r', self.user_name)

        # user move r permission folder
        data = {
            "src_repo_id": admin_repo_id,
            "src_parent_dir": '/',
            "src_dirents": [admin_folder_name],
            "dst_repo_id": self.dst_repo_id,
            "dst_parent_dir": '/',
        }
        resp = self.client.post(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == "Can't move folder %s, please check its permission." % admin_folder_name

class ReposBatchDeleteItemView(BaseTestCase):

    def create_new_repo(self, username):
        new_repo_id = seafile_api.create_repo(name=randstring(10),
                desc='', username=username, passwd=None)

        return new_repo_id

    def setUp(self):
        self.user_name = self.user.username
        self.admin_name = self.admin.username

        self.repo_id = self.repo.id

        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)

        self.url = reverse('api-v2.1-repos-batch-delete-item')

    def tearDown(self):
        self.remove_repo(self.repo_id)

    def test_can_delete(self):

        # items in parent folder
        assert seafile_api.get_dir_id_by_path(self.repo_id, self.folder_path) != None
        assert seafile_api.get_file_id_by_path(self.repo_id, self.file_path) != None

        self.login_as(self.user)

        data = {
            "repo_id": self.repo_id,
            "parent_dir": '/',
            "dirents": [self.folder_name, self.file_name],
        }

        resp = self.client.delete(self.url, json.dumps(data),
                'application/json')
        self.assertEqual(200, resp.status_code)

        # items NOT in parent folder
        assert seafile_api.get_dir_id_by_path(self.repo_id, self.folder_path) is None
        assert seafile_api.get_file_id_by_path(self.repo_id, self.file_path) is None

    def test_delete_with_invalid_parameter(self):

        self.login_as(self.user)

        data = {
            "parent_dir": '/',
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "repo_id": self.repo_id,
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

        data = {
            "repo_id": self.repo_id,
            "parent_dir": '/',
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(400, resp.status_code)

    def test_delete_with_repo_not_exist(self):

        self.login_as(self.user)

        invalid_repo_id = 'd53fe97e-919a-42f9-a29f-042d285ba6fb'
        data = {
            "repo_id": invalid_repo_id,
            "parent_dir": '/',
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_delete_with_folder_not_exist(self):

        self.login_as(self.user)

        data = {
            "repo_id": self.repo_id,
            "parent_dir": 'invalid_folder',
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(404, resp.status_code)

    def test_delete_with_invalid_repo_permission(self):

        tmp_repo_id = self.create_new_repo(self.admin_name)

        self.login_as(self.user)

        data = {
            "repo_id": tmp_repo_id,
            "parent_dir": '/',
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)

    def test_delete_with_invalid_parent_folder_permission(self):

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'r')
        data = {
            "repo_id": admin_repo_id,
            "parent_dir": '/',
            "dirents": [self.folder_name, self.file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        self.remove_repo(admin_repo_id)

    def test_delete_with_locked_file(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin lock file
        admin_file_name = randstring(6)
        seafile_api.post_empty_file(admin_repo_id, '/', admin_file_name,
                self.admin_name)
        seafile_api.lock_file(admin_repo_id, admin_file_name, self.admin_name, 0)

        # user move locked file
        data = {
            "repo_id": admin_repo_id,
            "parent_dir": '/',
            "dirents": [admin_file_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == 'File %s is locked.' % admin_file_name

    def test_delete_with_r_permission_sub_folder(self):

        if not LOCAL_PRO_DEV_ENV:
            return

        self.login_as(self.user)

        # share admin's tmp repo to user with 'r' permission
        admin_repo_id = self.create_new_repo(self.admin_name)
        seafile_api.share_repo(admin_repo_id, self.admin_name,
                self.user_name, 'rw')

        # admin set 'r' sub folder permission
        admin_folder_name = randstring(6)
        seafile_api.post_dir(admin_repo_id, '/', admin_folder_name, self.admin_name)
        seafile_api.add_folder_user_perm(admin_repo_id, '/' +
                admin_folder_name, 'r', self.user_name)

        # user move locked file
        data = {
            "repo_id": admin_repo_id,
            "parent_dir": '/',
            "dirents": [admin_folder_name],
        }
        resp = self.client.delete(self.url, json.dumps(data), 'application/json')
        self.assertEqual(403, resp.status_code)
        json_resp = json.loads(resp.content)
        assert json_resp['error_msg'] == "Can't delete folder %s, please check its permission." % admin_folder_name