Merge pull request #1215 from haiwen/share-link

Share link

seahub/api2/endpoints/share_links.py

@@ -20,6 +20,7 @@ from seahub.api2.throttling import UserRateThrottle
 
 from seahub.share.models import FileShare, OrgFileShare
 from seahub.utils import gen_shared_link, is_org_context
+from seahub.views import check_folder_permission
 
 logger = logging.getLogger(__name__)
 
@@ -78,6 +79,11 @@ class ShareLinks(APIView):
                 error_msg = 'Library %s not found.' % repo_id
                 return api_error(status.HTTP_404_NOT_FOUND, error_msg)
 
+            # repo level permission check
+            if not check_folder_permission(request, repo_id, '/'):
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
         path = request.GET.get('path', None)
         if path:
             try:
@@ -97,6 +103,11 @@ class ShareLinks(APIView):
 
                 return api_error(status.HTTP_404_NOT_FOUND, error_msg)
 
+            # folder/path permission check
+            if not check_folder_permission(request, repo_id, path):
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
         username = request.user.username
         fileshares = FileShare.objects.filter(username=username)
 
@@ -160,6 +171,11 @@ class ShareLinks(APIView):
 
             return api_error(status.HTTP_404_NOT_FOUND, error_msg)
 
+        # permission check
+        if not check_folder_permission(request, repo_id, path):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
         password = request.data.get('password', None)
         if password and len(password) < config.SHARE_LINK_PASSWORD_MIN_LENGTH:
             error_msg = _('Password is too short.')

seahub/api2/endpoints/upload_links.py

@@ -60,6 +60,11 @@ class UploadLinks(APIView):
                 error_msg = 'Library %s not found.' % repo_id
                 return api_error(status.HTTP_404_NOT_FOUND, error_msg)
 
+            # repo level permission check
+            if not check_folder_permission(request, repo_id, '/'):
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
         path = request.GET.get('path', None)
         if path:
             try:
@@ -73,6 +78,11 @@ class UploadLinks(APIView):
                 error_msg = 'folder %s not found.' % path
                 return api_error(status.HTTP_404_NOT_FOUND, error_msg)
 
+            # folder permission check
+            if not check_folder_permission(request, repo_id, path):
+                error_msg = 'Permission denied.'
+                return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
         username = request.user.username
         upload_link_shares = UploadLinkShare.objects.filter(username=username)
 

seahub/group/views.py

@@ -54,7 +54,7 @@ from seahub.utils import render_error, render_permission_error, string2list, \
     calc_file_path_hash, is_valid_username, send_html_email, is_org_context
 from seahub.utils.file_types import IMAGE
 from seahub.utils.paginator import Paginator
-from seahub.views import is_registered_user
+from seahub.views import is_registered_user, check_folder_permission
 from seahub.views.modules import get_enabled_mods_by_group, MOD_GROUP_WIKI, \
     enable_mod_for_group, disable_mod_for_group, get_available_mods_by_group, \
     get_wiki_enabled_group_list
@@ -533,6 +533,10 @@ def group_wiki_use_lib(request, group):
         messages.error(request, _('Failed to set wiki library.'))
         return HttpResponseRedirect(next)
 
+    if check_folder_permission(request, repo_id, '/') != 'rw':
+        messages.error(request, _('Permission denied.'))
+        return HttpResponseRedirect(next)
+
     GroupWiki.objects.save_group_wiki(group_id=group.id, repo_id=repo_id)
 
     # create home page if not exist

seahub/share/views.py

@@ -927,77 +927,119 @@ def ajax_get_upload_link(request):
     content_type = 'application/json; charset=utf-8'
 
     if request.method == 'GET':
-        repo_id = request.GET.get('repo_id', '')
-        path = request.GET.get('p', '')
-        username = request.user.username
+        repo_id = request.GET.get('repo_id', None)
+        path = request.GET.get('p', None)
+
+        # augument check
+        if not repo_id:
+            data = json.dumps({'error': 'repo_id invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if not path:
+            data = json.dumps({'error': 'p invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        # resource check
+        try:
+            repo = seafile_api.get_repo(repo_id)
+        except Exception as e:
+            logger.error(e)
+            data = json.dumps({'error': 'Internal Server Error'})
+            return HttpResponse(data, status=500, content_type=content_type)
+
+        if not repo:
+            data = json.dumps({'error': 'Library %s not found.' % repo_id})
+            return HttpResponse(data, status=404, content_type=content_type)
 
         if not path.endswith('/'):
             path = path + '/'
 
+        if not seafile_api.get_dir_id_by_path(repo_id, path):
+            data = json.dumps({'error': 'Folder %s not found.' % path})
+            return HttpResponse(data, status=404, content_type=content_type)
+
+        # permission check
+        if not check_folder_permission(request, repo_id, path):
+            data = json.dumps({'error': 'Permission denied.'})
+            return HttpResponse(data, status=403, content_type=content_type)
+
+        # get upload link
+        username = request.user.username
         l = UploadLinkShare.objects.filter(repo_id=repo_id).filter(
             username=username).filter(path=path)
+
+        data = {}
         if len(l) > 0:
             token = l[0].token
-            data = {
-                    'upload_link': gen_shared_upload_link(token),
-                    'token': token,
-                   }
-        else:
-            data = {}
+            data['upload_link'] = gen_shared_upload_link(token)
+            data['token'] = token
 
         return HttpResponse(json.dumps(data), content_type=content_type)
 
     elif request.method == 'POST':
-
-        if not request.user.permissions.can_generate_shared_link():
-            err = _('You do not have permission to generate shared link')
-            data = json.dumps({'error': err})
-            return HttpResponse(data, status=403, content_type=content_type)
-
-        repo_id = request.POST.get('repo_id', '')
-        path = request.POST.get('p', '')
+        repo_id = request.POST.get('repo_id', None)
+        path = request.POST.get('p', None)
         use_passwd = True if int(request.POST.get('use_passwd', '0')) == 1 else False
         passwd = request.POST.get('passwd') if use_passwd else None
 
-        if not (repo_id and path):
-            err = _('Invalid arguments')
-            data = json.dumps({'error': err})
+        # augument check
+        if not repo_id:
+            data = json.dumps({'error': 'repo_id invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if not path:
+            data = json.dumps({'error': 'p invalid.'})
             return HttpResponse(data, status=400, content_type=content_type)
 
         if passwd and len(passwd) < config.SHARE_LINK_PASSWORD_MIN_LENGTH:
-            err = _('Password is too short')
-            data = json.dumps({'error': err})
+            data = json.dumps({'error': _('Password is too short')})
             return HttpResponse(data, status=400, content_type=content_type)
 
-        if path[-1] != '/': # append '/' at end of path
-            path += '/'
+        # resource check
+        try:
+            repo = seafile_api.get_repo(repo_id)
+        except Exception as e:
+            logger.error(e)
+            data = json.dumps({'error': 'Internal Server Error'})
+            return HttpResponse(data, status=500, content_type=content_type)
 
-        repo = seaserv.get_repo(repo_id)
         if not repo:
-            err = 'Library %s not found.' % repo_id
-            data = json.dumps({'error': err})
+            data = json.dumps({'error': 'Library %s not found.' % repo_id})
             return HttpResponse(data, status=404, content_type=content_type)
 
-        user_perm = check_folder_permission(request, repo_id, '/')
-        if user_perm == 'rw':
-            l = UploadLinkShare.objects.filter(repo_id=repo_id).filter(
-                username=request.user.username).filter(path=path)
-            if len(l) > 0:
-                upload_link = l[0]
-                token = upload_link.token
-            else:
-                username = request.user.username
-                uls = UploadLinkShare.objects.create_upload_link_share(
-                    username, repo_id, path, passwd)
-                token = uls.token
+        if not path.endswith('/'):
+            path = path + '/'
 
-            shared_upload_link = gen_shared_upload_link(token)
+        if not seafile_api.get_dir_id_by_path(repo_id, path):
+            data = json.dumps({'error': 'Folder %s not found.' % path})
+            return HttpResponse(data, status=404, content_type=content_type)
 
-            data = json.dumps({'token': token, 'upload_link': shared_upload_link})
-            return HttpResponse(data, content_type=content_type)
+        # permission check
+        # normal permission check & default/guest user permission check
+        if check_folder_permission(request, repo_id, path) != 'rw' or \
+            not request.user.permissions.can_generate_shared_link():
+            data = json.dumps({'error': 'Permission denied.'})
+            return HttpResponse(data, status=403, content_type=content_type)
+
+        # generate upload link
+        l = UploadLinkShare.objects.filter(repo_id=repo_id).filter(
+            username=request.user.username).filter(path=path)
+
+        if len(l) > 0:
+            # if already exist
+            upload_link = l[0]
+            token = upload_link.token
         else:
-            return HttpResponse(json.dumps({'error': _(u'Permission denied')}),
-                status=403, content_type=content_type)
+            # generate new
+            username = request.user.username
+            uls = UploadLinkShare.objects.create_upload_link_share(
+                username, repo_id, path, passwd)
+            token = uls.token
+
+        shared_upload_link = gen_shared_upload_link(token)
+        data = json.dumps({'token': token, 'upload_link': shared_upload_link})
+
+        return HttpResponse(data, content_type=content_type)
 
 @login_required_ajax
 def ajax_get_download_link(request):
@@ -1007,60 +1049,133 @@ def ajax_get_download_link(request):
     content_type = 'application/json; charset=utf-8'
 
     if request.method == 'GET':
-        repo_id = request.GET.get('repo_id', '')
-        share_type = request.GET.get('type', 'f')  # `f` or `d`
-        path = request.GET.get('p', '')
+        repo_id = request.GET.get('repo_id', None)
+        share_type = request.GET.get('type', 'f')
+        path = request.GET.get('p', None)
+
+        # augument check
+        if not repo_id:
+            data = json.dumps({'error': 'repo_id invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if not path:
+            data = json.dumps({'error': 'p invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if share_type not in ('f', 'd'):
+            data = json.dumps({'error': 'type invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        # resource check
+        try:
+            repo = seafile_api.get_repo(repo_id)
+        except Exception as e:
+            logger.error(e)
+            data = json.dumps({'error': 'Internal Server Error'})
+            return HttpResponse(data, status=500, content_type=content_type)
+
+        if not repo:
+            data = json.dumps({'error': 'Library %s not found.' % repo_id})
+            return HttpResponse(data, status=404, content_type=content_type)
+
+        if share_type == 'f':
+            if not seafile_api.get_file_id_by_path(repo_id, path):
+                data = json.dumps({'error': 'File %s not found.' % path})
+                return HttpResponse(data, status=404, content_type=content_type)
+
+        if share_type == 'd':
+            if not path.endswith('/'):
+                path = path + '/'
+
+            if not seafile_api.get_dir_id_by_path(repo_id, path):
+                data = json.dumps({'error': 'Folder %s not found.' % path})
+                return HttpResponse(data, status=404, content_type=content_type)
+
+        # permission check
+        if not check_folder_permission(request, repo_id, path):
+            data = json.dumps({'error': 'Permission denied.'})
+            return HttpResponse(data, status=403, content_type=content_type)
+
+        # get download link
         username = request.user.username
-
-        if share_type == 'd' and not path.endswith('/'):
-            path = path + '/'
-
         l = FileShare.objects.filter(repo_id=repo_id).filter(
             username=username).filter(path=path)
+
+        data = {}
         if len(l) > 0:
             token = l[0].token
-            data = {
-                    'download_link': gen_shared_link(token, l[0].s_type),
-                    'token': token,
-                    'is_expired': l[0].is_expired(),
-                   }
-        else:
-            data = {}
+            data['download_link'] = gen_shared_link(token, l[0].s_type)
+            data['token'] = token
+            data['is_expired'] = l[0].is_expired()
 
         return HttpResponse(json.dumps(data), content_type=content_type)
 
     elif request.method == 'POST':
-
-        if not request.user.permissions.can_generate_shared_link():
-            err = _('You do not have permission to generate shared link')
-            data = json.dumps({'error': err})
-            return HttpResponse(data, status=403, content_type=content_type)
-
-        repo_id = request.POST.get('repo_id', '')
-        share_type = request.POST.get('type', 'f')  # `f` or `d`
-        path = request.POST.get('p', '')
+        repo_id = request.POST.get('repo_id', None)
+        path = request.POST.get('p', None)
+        share_type = request.POST.get('type', 'f')
         use_passwd = True if int(request.POST.get('use_passwd', '0')) == 1 else False
         passwd = request.POST.get('passwd') if use_passwd else None
 
-        if not (repo_id and path):
-            err = _('Invalid arguments')
-            data = json.dumps({'error': err})
+        # augument check
+        if not repo_id:
+            data = json.dumps({'error': 'repo_id invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if not path:
+            data = json.dumps({'error': 'p invalid.'})
+            return HttpResponse(data, status=400, content_type=content_type)
+
+        if share_type not in ('f', 'd'):
+            data = json.dumps({'error': 'type invalid.'})
             return HttpResponse(data, status=400, content_type=content_type)
 
         if passwd and len(passwd) < config.SHARE_LINK_PASSWORD_MIN_LENGTH:
-            err = _('Password is too short')
-            data = json.dumps({'error': err})
+            data = json.dumps({'error': _('Password is too short')})
             return HttpResponse(data, status=400, content_type=content_type)
 
         try:
             expire_days = int(request.POST.get('expire_days', 0))
         except ValueError:
             expire_days = 0
+
         if expire_days <= 0:
             expire_date = None
         else:
             expire_date = timezone.now() + relativedelta(days=expire_days)
 
+        # resource check
+        try:
+            repo = seafile_api.get_repo(repo_id)
+        except Exception as e:
+            logger.error(e)
+            data = json.dumps({'error': 'Internal Server Error'})
+            return HttpResponse(data, status=500, content_type=content_type)
+
+        if not repo:
+            data = json.dumps({'error': 'Library %s not found.' % repo_id})
+            return HttpResponse(data, status=404, content_type=content_type)
+
+        if share_type == 'f':
+            if not seafile_api.get_file_id_by_path(repo_id, path):
+                data = json.dumps({'error': 'File %s not found.' % path})
+                return HttpResponse(data, status=404, content_type=content_type)
+
+        if share_type == 'd':
+            if not path.endswith('/'):
+                path = path + '/'
+
+            if not seafile_api.get_dir_id_by_path(repo_id, path):
+                data = json.dumps({'error': 'Folder %s not found.' % path})
+                return HttpResponse(data, status=404, content_type=content_type)
+
+        # permission check
+        # normal permission check & default/guest user permission check
+        if check_folder_permission(request, repo_id, path) != 'rw' or \
+            not request.user.permissions.can_generate_shared_link():
+            data = json.dumps({'error': 'Permission denied.'})
+            return HttpResponse(data, status=403, content_type=content_type)
+
         username = request.user.username
         if share_type == 'f':
             fs = FileShare.objects.get_file_link_by_path(username, repo_id, path)

seahub/views/wiki.py

@@ -37,6 +37,7 @@ from seahub.wiki import get_personal_wiki_page, get_personal_wiki_repo, \
 from seahub.wiki.forms import WikiCreateForm, WikiNewPageForm
 from seahub.wiki.utils import clean_page_name, page_name_to_file_name
 from seahub.utils import render_error
+from seahub.views import check_folder_permission
 
 # Get an instance of a logger
 logger = logging.getLogger(__name__)
@@ -166,6 +167,7 @@ def personal_wiki_create(request):
 def personal_wiki_use_lib(request):
     if request.method != 'POST':
         raise Http404
+
     repo_id = request.POST.get('dst_repo', '')
     username = request.user.username
     next = reverse('personal_wiki', args=[])
@@ -174,6 +176,10 @@ def personal_wiki_use_lib(request):
         messages.error(request, _('Failed to set wiki library.'))
         return HttpResponseRedirect(next)
 
+    if check_folder_permission(request, repo_id, '/') != 'rw':
+        messages.error(request, _('Permission denied.'))
+        return HttpResponseRedirect(next)
+
     PersonalWiki.objects.save_personal_wiki(username=username, repo_id=repo_id)
 
     # create home page if not exist

tests/api/endpoints/test_share_links.py

@@ -148,6 +148,20 @@ class ShareLinksTest(BaseTestCase):
         resp = self.client.post(self.url)
         self.assertEqual(403, resp.status_code)
 
+        self.logout()
+
+        # login with another user to test repo permission
+        self.login_as(self.admin)
+        mock_can_generate_shared_link.return_value = True
+
+        args = '?repo_id=%s' % self.repo_id
+        resp = self.client.get(self.url + args)
+        self.assertEqual(403, resp.status_code)
+
+        data = {'path': self.file_path, 'repo_id': self.repo_id}
+        resp = self.client.post(self.url, data)
+        self.assertEqual(403, resp.status_code)
+
     @patch.object(ShareLink, '_can_generate_shared_link')
     def test_can_not_delete_link_with_invalid_permission(self, mock_can_generate_shared_link):
         token = self._add_file_share_link()

tests/api/endpoints/test_upload_links.py

@@ -92,6 +92,20 @@ class UploadLinksTest(BaseTestCase):
         resp = self.client.post(self.url)
         self.assertEqual(403, resp.status_code)
 
+        self.logout()
+
+        # login with another user to test repo permission
+        self.login_as(self.admin)
+        mock_can_generate_shared_link.return_value = True
+
+        args = '?repo_id=%s' % self.repo_id
+        resp = self.client.get(self.url + args)
+        self.assertEqual(403, resp.status_code)
+
+        data = {'path': self.folder_path, 'repo_id': self.repo_id}
+        resp = self.client.post(self.url, data)
+        self.assertEqual(403, resp.status_code)
+
     @patch.object(UploadLink, '_can_generate_shared_link')
     def test_can_not_delete_link_with_invalid_permission(self, mock_can_generate_shared_link):
         token = self._add_upload_link()

tests/seahub/share/views/test_ajax_get_download_link.py

@@ -0,0 +1,122 @@
+import json
+
+from django.core.urlresolvers import reverse
+
+from seahub.share.models import FileShare
+from seahub.test_utils import BaseTestCase
+
+
+class AjaxGetDownloadLinkTest(BaseTestCase):
+    def setUp(self):
+
+        self.url = reverse('ajax_get_download_link')
+
+        self.user_repo_id = self.repo.id
+        self.user_dir_path = self.folder
+        self.user_file_path = self.file
+
+    def test_can_generate_file_share_link(self):
+        self.login_as(self.user)
+
+        url = self.url
+        data = {
+            'repo_id': self.user_repo_id,
+            'p': self.user_file_path,
+            'type': 'f',
+        }
+        resp = self.client.post(url, data, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert '/f/' in json_resp['download_link']
+
+    def test_can_generate_dir_share_link(self):
+        self.login_as(self.user)
+
+        url = self.url
+        data = {
+            'repo_id': self.user_repo_id,
+            'p': self.user_dir_path,
+            'type': 'd',
+        }
+        resp = self.client.post(url, data, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert '/d/' in json_resp['download_link']
+
+    def test_can_get_file_share_link(self):
+        fs = FileShare.objects.create_file_link(self.user.username,
+                self.user_repo_id, self.user_file_path)
+
+        self.login_as(self.user)
+
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert fs.token in json_resp['download_link']
+
+    def test_can_get_dir_share_link(self):
+        fs = FileShare.objects.create_dir_link(self.user.username,
+                self.user_repo_id, self.user_dir_path)
+
+        self.login_as(self.user)
+
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_dir_path, 'd')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert fs.token in json_resp['download_link']
+
+    def test_can_not_get_if_not_login(self):
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(401, resp.status_code)
+
+    def test_invalid_args(self):
+        self.login_as(self.user)
+
+        # invalid type
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'invalid_type')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(400, resp.status_code)
+
+        # invalid repo_id
+        args = '?invalid_repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(400, resp.status_code)
+
+        # invalid path
+        args = '?repo_id=%s&invalid_path=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(400, resp.status_code)
+
+    def test_invalid_recourse(self):
+        self.login_as(self.user)
+
+        # invalid repo_id
+        args = '?repo_id=%s&p=%s&type=%s' % ('invalid repo id', self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(500, resp.status_code)
+
+        # invalid repo_id
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id[:30] + '123456', self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(404, resp.status_code)
+
+        # invalid path
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, 'invalid path', 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(404, resp.status_code)
+
+    def test_invalid_permission(self):
+        self.login_as(self.admin)
+
+        args = '?repo_id=%s&p=%s&type=%s' % (self.user_repo_id, self.user_file_path, 'f')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(403, resp.status_code)

tests/seahub/share/views/test_ajax_get_upload_link.py

@@ -0,0 +1,89 @@
+import json
+
+from django.core.urlresolvers import reverse
+
+from seahub.share.models import UploadLinkShare
+from seahub.test_utils import BaseTestCase
+
+
+class AjaxGetUploadLinkTest(BaseTestCase):
+    def setUp(self):
+        self.url = reverse('ajax_get_upload_link')
+
+        self.user_repo_id = self.repo.id
+        self.user_dir_path = self.folder
+        self.user_file_path = self.file
+
+    def test_can_generate(self):
+        self.login_as(self.user)
+
+        url = self.url
+        data = {
+            'repo_id': self.user_repo_id,
+            'p': self.user_dir_path,
+        }
+        resp = self.client.post(url, data, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert '/u/d/' in json_resp['upload_link']
+
+    def test_can_get(self):
+        upload_link = UploadLinkShare.objects.create_upload_link_share(
+            self.user.username, self.user_repo_id, self.user_dir_path)
+
+        self.login_as(self.user)
+
+        args = '?repo_id=%s&p=%s' % (self.user_repo_id, self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        json_resp = json.loads(resp.content)
+        assert upload_link.token in json_resp['upload_link']
+
+    def test_unlogin_user(self):
+        args = '?repo_id=%s&p=%s' % (self.user_repo_id, self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(401, resp.status_code)
+
+    def test_invalid_args(self):
+        self.login_as(self.user)
+
+        # invalid repo_id
+        args = '?invalid_repo_id=%s&p=%s' % (self.user_repo_id, self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(400, resp.status_code)
+
+        # invalid path
+        args = '?repo_id=%s&invalid_path=%s' % (self.user_repo_id, self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(400, resp.status_code)
+
+    def test_invalid_recourse(self):
+        self.login_as(self.user)
+
+        # invalid repo_id
+        args = '?repo_id=%s&p=%s' % ('invalid repo id', self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(500, resp.status_code)
+
+        # invalid repo_id
+        args = '?repo_id=%s&p=%s' % (self.user_repo_id[:30] + '123456', self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(404, resp.status_code)
+
+        # invalid path
+        args = '?repo_id=%s&p=%s' % (self.user_repo_id, 'invalid path')
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(404, resp.status_code)
+
+    def test_invalid_permission(self):
+        self.login_as(self.admin)
+
+        args = '?repo_id=%s&p=%s' % (self.user_repo_id, self.user_dir_path)
+        url = self.url + args
+        resp = self.client.get(url, HTTP_X_REQUESTED_WITH='XMLHttpRequest')
+        self.assertEqual(403, resp.status_code)

tests/seahub/views/wiki/test_personal_wiki.py

@@ -21,7 +21,22 @@ class PersonalWikiTest(BaseTestCase):
 
         resp = self.client.get(reverse('personal_wiki'))
         self.assertEqual(302, resp.status_code)
-        self.assertRedirects(resp, reverse('personal_wiki', args=['home']))
+
+    def test_invalid_permisison(self):
+        self.login_as(self.admin)
+
+        data = {'dst_repo': self.repo.id}
+        resp = self.client.post(reverse('personal_wiki_use_lib'), data)
+        assert 'Permission denied.' in str(resp.cookies)
+        self.assertEqual(302, resp.status_code)
+
+    def test_invalid_repo(self):
+        self.login_as(self.user)
+
+        data = {'dst_repo': self.repo.id[:30] + '123456'}
+        resp = self.client.post(reverse('personal_wiki_use_lib'), data)
+        assert 'Failed to set wiki library.' in str(resp.cookies)
+        self.assertEqual(302, resp.status_code)
 
     def test_home_page(self):
         self.login_as(self.user)