Merge pull request #1024 from haiwen/file-shared-link

[api-v2.1] add file share links api

seahub/api2/endpoints/share_links.py

@@ -0,0 +1,245 @@
+import logging
+from constance import config
+from dateutil.relativedelta import relativedelta
+
+from rest_framework.authentication import SessionAuthentication
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework import status
+
+from django.utils import timezone
+from django.utils.translation import ugettext as _
+
+from seaserv import seafile_api
+from pysearpc import SearpcError
+
+from seahub.api2.utils import api_error
+from seahub.api2.authentication import TokenAuthentication
+from seahub.api2.throttling import UserRateThrottle
+
+from seahub.share.models import FileShare, OrgFileShare
+from seahub.utils import gen_shared_link, is_org_context
+
+logger = logging.getLogger(__name__)
+
+
+class ShareLinks(APIView):
+
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAuthenticated,)
+    throttle_classes = (UserRateThrottle, )
+
+    def _can_generate_shared_link(self, request):
+
+        return request.user.permissions.can_generate_shared_link()
+
+    def _generate_obj_id_and_type_by_path(self, repo_id, path):
+
+        file_id = seafile_api.get_file_id_by_path(repo_id, path)
+        if file_id:
+            return (file_id, 'f')
+
+        dir_id = seafile_api.get_dir_id_by_path(repo_id, path)
+        if dir_id:
+            return (dir_id, 'd')
+
+        return (None, None)
+
+    def _get_share_link_info(self, fileshare):
+        data = {}
+        token = fileshare.token
+
+        data['repo_id'] = fileshare.repo_id
+        data['path'] = fileshare.path
+        data['ctime'] = fileshare.ctime
+        data['view_cnt'] = fileshare.view_cnt
+        data['link'] = gen_shared_link(token, fileshare.s_type)
+        data['token'] = token
+        data['expire_date'] = fileshare.expire_date
+        data['is_expired'] = fileshare.is_expired()
+        data['username'] = fileshare.username
+
+        return data
+
+    def get(self, request):
+        """ get share links.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        # check if args invalid
+        repo_id = request.GET.get('repo_id', None)
+        if repo_id:
+            repo = seafile_api.get_repo(repo_id)
+            if not repo:
+                error_msg = 'Library %s not found.' % repo_id
+                return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        path = request.GET.get('path', None)
+        if path:
+            try:
+                obj_id, s_type = self._generate_obj_id_and_type_by_path(repo_id, path)
+            except SearpcError as e:
+                logger.error(e)
+                error_msg = 'Internal Server Error'
+                return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
+
+            if not obj_id:
+                if s_type == 'f':
+                    error_msg = 'file %s not found.' % path
+                elif s_type == 'd':
+                    error_msg = 'folder %s not found.' % path
+                else:
+                    error_msg = 'path %s not found.' % path
+
+                return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        username = request.user.username
+        fileshares = FileShare.objects.filter(username=username)
+
+        # filter result by args
+        if repo_id:
+            fileshares = filter(lambda fs: fs.repo_id == repo_id, fileshares)
+
+        if path:
+            if s_type == 'd' and path[-1] != '/':
+                path = path + '/'
+
+            fileshares = filter(lambda fs: fs.path == path, fileshares)
+
+        result = []
+        for fs in fileshares:
+            link_info = self._get_share_link_info(fs)
+            result.append(link_info)
+
+        if len(result) == 1:
+            result = result[0]
+
+        return Response(result)
+
+    def post(self, request):
+        """ create share link.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        repo_id = request.data.get('repo_id', None)
+        if not repo_id:
+            error_msg = 'repo_id invalid.'
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        repo = seafile_api.get_repo(repo_id)
+        if not repo:
+            error_msg = 'Library %s not found.' % repo_id
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        path = request.data.get('path', None)
+        if not path:
+            error_msg = 'path invalid.'
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        try:
+            obj_id, s_type = self._generate_obj_id_and_type_by_path(repo_id, path)
+        except SearpcError as e:
+            logger.error(e)
+            error_msg = 'Internal Server Error'
+            return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
+
+        if not obj_id:
+            if s_type == 'f':
+                error_msg = 'file %s not found.' % path
+            elif s_type == 'd':
+                error_msg = 'folder %s not found.' % path
+            else:
+                error_msg = 'path %s not found.' % path
+
+            return api_error(status.HTTP_404_NOT_FOUND, 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.')
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        try:
+            expire_days = int(request.data.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)
+
+        username = request.user.username
+        if s_type == 'f':
+            fs = FileShare.objects.get_file_link_by_path(username, repo_id, path)
+            if not fs:
+                fs = FileShare.objects.create_file_link(username, repo_id, path,
+                                                        password, expire_date)
+                if is_org_context(request):
+                    org_id = request.user.org.org_id
+                    OrgFileShare.objects.set_org_file_share(org_id, fs)
+
+        elif s_type == 'd':
+            fs = FileShare.objects.get_dir_link_by_path(username, repo_id, path)
+            if not fs:
+                fs = FileShare.objects.create_dir_link(username, repo_id, path,
+                                                        password, expire_date)
+                if is_org_context(request):
+                    org_id = request.user.org.org_id
+                    OrgFileShare.objects.set_org_file_share(org_id, fs)
+
+        link_info = self._get_share_link_info(fs)
+        return Response(link_info)
+
+class ShareLink(APIView):
+
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAuthenticated,)
+    throttle_classes = (UserRateThrottle, )
+
+    def _can_generate_shared_link(self, request):
+
+        return request.user.permissions.can_generate_shared_link()
+
+    def get(self, request, token):
+        try:
+            fs = FileShare.objects.get(token=token)
+        except FileShare.DoesNotExist:
+            error_msg = 'token %s not found.' % token
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        link_info = self._get_share_link_info(fs)
+        return Response(link_info)
+
+    def delete(self, request, token):
+        """ delete share link.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            fs = FileShare.objects.get(token=token)
+        except FileShare.DoesNotExist:
+            error_msg = 'token %s not found.' % token
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        username = request.user.username
+        if not fs.is_owner(username):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            fs.delete()
+            return Response({'success': True})
+        except Exception as e:
+            logger.error(e)
+            error_msg = 'Internal Server Error'
+            return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)

seahub/api2/endpoints/upload_links.py

@@ -0,0 +1,200 @@
+import logging
+from constance import config
+
+from rest_framework.authentication import SessionAuthentication
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+from rest_framework import status
+
+from django.utils.translation import ugettext as _
+
+from seaserv import seafile_api
+from pysearpc import SearpcError
+
+from seahub.api2.utils import api_error
+from seahub.api2.authentication import TokenAuthentication
+from seahub.api2.throttling import UserRateThrottle
+
+from seahub.share.models import UploadLinkShare
+from seahub.utils import gen_shared_upload_link
+from seahub.views import check_folder_permission
+
+logger = logging.getLogger(__name__)
+
+class UploadLinks(APIView):
+
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAuthenticated,)
+    throttle_classes = (UserRateThrottle, )
+
+    def _can_generate_shared_link(self, request):
+
+        return request.user.permissions.can_generate_shared_link()
+
+    def _get_upload_link_info(self, uls):
+        data = {}
+        token = uls.token
+
+        data['repo_id'] = uls.repo_id
+        data['path'] = uls.path
+        data['ctime'] = uls.ctime
+        data['link'] = gen_shared_upload_link(token)
+        data['token'] = token
+        data['username'] = uls.username
+
+        return data
+
+    def get(self, request):
+        """ get upload link.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        repo_id = request.GET.get('repo_id', None)
+        if repo_id:
+            repo = seafile_api.get_repo(repo_id)
+            if not repo:
+                error_msg = 'Library %s not found.' % repo_id
+                return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        path = request.GET.get('path', None)
+        if path:
+            try:
+                dir_id = seafile_api.get_dir_id_by_path(repo_id, path)
+            except SearpcError as e:
+                logger.error(e)
+                error_msg = 'Internal Server Error'
+                return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
+
+            if not dir_id:
+                error_msg = 'folder %s not found.' % path
+                return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        username = request.user.username
+        upload_link_shares = UploadLinkShare.objects.filter(username=username)
+
+        # filter result by args
+        if repo_id:
+            upload_link_shares = filter(lambda ufs: ufs.repo_id == repo_id, upload_link_shares)
+
+        if path:
+            if path[-1] != '/':
+                path = path + '/'
+
+            upload_link_shares = filter(lambda ufs: ufs.path == path, upload_link_shares)
+
+        result = []
+        for uls in upload_link_shares:
+            link_info = self._get_upload_link_info(uls)
+            result.append(link_info)
+
+        if len(result) == 1:
+            result = result[0]
+
+        return Response(result)
+
+    def post(self, request):
+        """ create upload link.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        repo_id = request.data.get('repo_id', None)
+        if not repo_id:
+            error_msg = 'repo_id invalid.'
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        repo = seafile_api.get_repo(repo_id)
+        if not repo:
+            error_msg = 'Library %s not found.' % repo_id
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        path = request.data.get('path', None)
+        if not path:
+            error_msg = 'path invalid.'
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        try:
+            dir_id = seafile_api.get_dir_id_by_path(repo_id, path)
+        except SearpcError as e:
+            logger.error(e)
+            error_msg = 'Internal Server Error'
+            return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
+
+        if not dir_id:
+            error_msg = 'folder %s not found.' % path
+            return api_error(status.HTTP_404_NOT_FOUND, 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.')
+            return api_error(status.HTTP_400_BAD_REQUEST, error_msg)
+
+        user_perm = check_folder_permission(request, repo_id, '/')
+        if user_perm != 'rw':
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        username = request.user.username
+        uls = UploadLinkShare.objects.get_upload_link_by_path(username, repo_id, path)
+        if not uls:
+            uls = UploadLinkShare.objects.create_upload_link_share(username,
+                repo_id, path, password)
+
+        link_info = self._get_upload_link_info(uls)
+        return Response(link_info)
+
+class UploadLink(APIView):
+
+    authentication_classes = (TokenAuthentication, SessionAuthentication)
+    permission_classes = (IsAuthenticated,)
+    throttle_classes = (UserRateThrottle, )
+
+    def _can_generate_shared_link(self, request):
+
+        return request.user.permissions.can_generate_shared_link()
+
+    def get(self, request, token):
+        """ get upload link info.
+        """
+
+        try:
+            uls = UploadLinkShare.objects.get(token=token)
+        except UploadLinkShare.DoesNotExist:
+            error_msg = 'token %s not found.' % token
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        link_info = self._get_upload_link_info(uls)
+        return Response(link_info)
+
+    def delete(self, request, token):
+        """ delete upload link.
+        """
+
+        if not self._can_generate_shared_link(request):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            uls = UploadLinkShare.objects.get(token=token)
+        except UploadLinkShare.DoesNotExist:
+            error_msg = 'token %s not found.' % token
+            return api_error(status.HTTP_404_NOT_FOUND, error_msg)
+
+        username = request.user.username
+        if not uls.is_owner(username):
+            error_msg = 'Permission denied.'
+            return api_error(status.HTTP_403_FORBIDDEN, error_msg)
+
+        try:
+            uls.delete()
+            return Response({'success': True})
+        except Exception as e:
+            logger.error(e)
+            error_msg = 'Internal Server Error'
+            return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)

seahub/urls.py

@@ -21,6 +21,8 @@ from seahub.views.sysadmin import *
 from seahub.views.ajax import *
 from seahub.api2.endpoints.groups import Groups, Group
 from seahub.api2.endpoints.group_members import GroupMembers, GroupMembersBulk, GroupMember
+from seahub.api2.endpoints.share_links import ShareLinks, ShareLink
+from seahub.api2.endpoints.upload_links import UploadLinks, UploadLink
 
 # Uncomment the next two lines to enable the admin:
 #from django.contrib import admin
@@ -195,6 +197,11 @@ urlpatterns = patterns(
     url(r'^api/v2.1/groups/(?P<group_id>\d+)/members/$', GroupMembers.as_view(), name='api-v2.1-group-members'),
     url(r'^api/v2.1/groups/(?P<group_id>\d+)/members/bulk/$', GroupMembersBulk.as_view(), name='api-v2.1-group-members-bulk'),
     url(r'^api/v2.1/groups/(?P<group_id>\d+)/members/(?P<email>[^/]+)/$', GroupMember.as_view(), name='api-v2.1-group-member'),
+    url(r'^api/v2.1/share-links/$', ShareLinks.as_view(), name='api-v2.1-share-links'),
+    url(r'^api/v2.1/share-link/(?P<token>[a-f0-9]{10})/$', ShareLink.as_view(), name='api-v2.1-share-link'),
+    url(r'^api/v2.1/upload-links/$', UploadLinks.as_view(), name='api-v2.1-upload-links'),
+    url(r'^api/v2.1/upload-link/(?P<token>[a-f0-9]{10})/$', UploadLink.as_view(), name='api-v2.1-upload-link'),
+
     (r'^avatar/', include('seahub.avatar.urls')),
     (r'^notification/', include('seahub.notifications.urls')),
     (r'^contacts/', include('seahub.contacts.urls')),

tests/api/endpoints/test_share_links.py

@@ -0,0 +1,156 @@
+# -*- coding: utf-8 -*-
+import json
+from mock import patch
+
+from django.core.urlresolvers import reverse
+from seahub.test_utils import BaseTestCase
+from seahub.share.models import FileShare
+from seahub.api2.endpoints.share_links import ShareLinks, ShareLink
+
+class ShareLinksTest(BaseTestCase):
+
+    def setUp(self):
+        self.repo_id = self.repo.id
+        self.file_path= self.file
+        self.folder_path= self.folder
+        self.url = reverse('api-v2.1-share-links')
+
+    def tearDown(self):
+        self.remove_repo()
+
+    def _add_file_share_link(self):
+        fs = FileShare.objects.create_file_link(self.user.username,
+            self.repo.id, self.file, None, None)
+
+        return fs.token
+
+    def _add_dir_share_link(self):
+        fs = FileShare.objects.create_dir_link(self.user.username,
+            self.repo.id, self.folder, None, None)
+
+        return fs.token
+
+    def _remove_share_link(self, token):
+        link = FileShare.objects.get(token=token)
+        link.delete()
+
+    # test file share link
+    def test_get_file_share_link(self):
+        self.login_as(self.user)
+        token = self._add_file_share_link()
+
+        resp = self.client.get(self.url + '?path=' + self.file_path + '&repo_id=' + self.repo_id)
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+        assert json_resp['is_expired'] is not None
+
+        assert token in json_resp['link']
+        assert 'f' in json_resp['link']
+
+        assert token == json_resp['token']
+
+        self._remove_share_link(token)
+
+    def test_create_file_share_link(self):
+        self.login_as(self.user)
+
+        resp = self.client.post(self.url, {'path': self.file_path, 'repo_id': self.repo_id})
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+        assert json_resp['is_expired'] is not None
+
+        assert json_resp['token'] in json_resp['link']
+        assert 'f' in json_resp['link']
+
+        self._remove_share_link(json_resp['token'])
+
+    def test_delete_file_share_link(self):
+        self.login_as(self.user)
+        token = self._add_file_share_link()
+
+        url = reverse('api-v2.1-share-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['success'] is True
+
+    # test dir share link
+    def test_get_dir_share_link(self):
+        self.login_as(self.user)
+        token = self._add_dir_share_link()
+
+        resp = self.client.get(self.url + '?path=' + self.folder_path + '&repo_id=' + self.repo_id)
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+        assert json_resp['is_expired'] is not None
+
+        assert token in json_resp['link']
+        assert 'd' in json_resp['link']
+
+        assert token == json_resp['token']
+
+        self._remove_share_link(token)
+
+    def test_create_dir_share_link(self):
+        self.login_as(self.user)
+
+        resp = self.client.post(self.url, {'path': self.folder_path, 'repo_id': self.repo_id})
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+        assert json_resp['is_expired'] is not None
+
+        assert json_resp['token'] in json_resp['link']
+        assert 'd' in json_resp['link']
+
+        self._remove_share_link(json_resp['token'])
+
+    def test_delete_dir_share_link(self):
+        self.login_as(self.user)
+        token = self._add_file_share_link()
+        url = reverse('api-v2.1-share-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['success'] is True
+
+    # test permission
+    def test_can_not_delete_link_if_not_owner(self):
+        self.login_as(self.admin)
+        token = self._add_file_share_link()
+        url = reverse('api-v2.1-share-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(403, resp.status_code)
+
+    @patch.object(ShareLinks, '_can_generate_shared_link')
+    def test_can_not_get_and_create_link_with_invalid_permission(self, mock_can_generate_shared_link):
+        self.login_as(self.user)
+        mock_can_generate_shared_link.return_value = False
+
+        resp = self.client.get(self.url)
+        self.assertEqual(403, resp.status_code)
+
+        resp = self.client.post(self.url)
+        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()
+        url = reverse('api-v2.1-share-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(403, resp.status_code)

tests/api/endpoints/test_upload_links.py

@@ -0,0 +1,100 @@
+# -*- coding: utf-8 -*-
+import json
+from mock import patch
+
+from django.core.urlresolvers import reverse
+from seahub.test_utils import BaseTestCase
+from seahub.share.models import UploadLinkShare
+from seahub.api2.endpoints.upload_links import UploadLinks, UploadLink
+
+class UploadLinksTest(BaseTestCase):
+
+    def setUp(self):
+        self.repo_id = self.repo.id
+        self.folder_path= self.folder
+        self.url = reverse('api-v2.1-upload-links')
+
+    def tearDown(self):
+        self.remove_repo()
+
+    def _add_upload_link(self):
+        upload_link = UploadLinkShare.objects.create_upload_link_share(self.user.username,
+            self.repo.id, self.folder, None, None)
+
+        return upload_link.token
+
+    def _remove_upload_link(self, token):
+        link = UploadLinkShare.objects.get(token=token)
+        link.delete()
+
+    def test_get_upload_link(self):
+        self.login_as(self.user)
+        token = self._add_upload_link()
+
+        resp = self.client.get(self.url + '?path=' + self.folder_path + '&repo_id=' + self.repo_id)
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+
+        assert token in json_resp['link']
+        assert 'u/d' in json_resp['link']
+
+        assert token == json_resp['token']
+
+        self._remove_upload_link(token)
+
+    def test_create_upload_link(self):
+        self.login_as(self.user)
+
+        resp = self.client.post(self.url, {'path': self.folder_path, 'repo_id': self.repo_id})
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['link'] is not None
+        assert json_resp['token'] is not None
+
+        assert json_resp['token'] in json_resp['link']
+        assert 'u/d' in json_resp['link']
+
+        self._remove_upload_link(json_resp['token'])
+
+    def test_delete_upload_link(self):
+        self.login_as(self.user)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-upload-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(200, resp.status_code)
+
+        json_resp = json.loads(resp.content)
+        assert json_resp['success'] is True
+
+    # test permission
+    def test_can_not_delete_link_if_not_owner(self):
+        self.login_as(self.admin)
+        token = self._add_upload_link()
+
+        url = reverse('api-v2.1-upload-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(403, resp.status_code)
+
+    @patch.object(UploadLinks, '_can_generate_shared_link')
+    def test_can_not_get_and_create_link_with_invalid_permission(self, mock_can_generate_shared_link):
+        self.login_as(self.user)
+        mock_can_generate_shared_link.return_value = False
+
+        resp = self.client.get(self.url)
+        self.assertEqual(403, resp.status_code)
+
+        resp = self.client.post(self.url)
+        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()
+        url = reverse('api-v2.1-upload-link', args=[token])
+        resp = self.client.delete(url, {}, 'application/x-www-form-urlencoded')
+        self.assertEqual(403, resp.status_code)