1
0
mirror of https://github.com/haiwen/seahub.git synced 2025-09-04 08:28:11 +00:00

file participants update python

This commit is contained in:
sniper-py
2019-06-11 17:21:38 +08:00
committed by Michael An
parent e0e7bfc927
commit c29d990c67
11 changed files with 243 additions and 221 deletions

View File

@@ -8,54 +8,56 @@ from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response from rest_framework.response import Response
from rest_framework.views import APIView from rest_framework.views import APIView
from seaserv import seafile_api from seaserv import seafile_api
from django.utils.translation import ugettext as _
from seahub.api2.authentication import TokenAuthentication from seahub.api2.authentication import TokenAuthentication
from seahub.api2.permissions import IsRepoAccessible
from seahub.api2.throttling import UserRateThrottle from seahub.api2.throttling import UserRateThrottle
from seahub.api2.utils import api_error, get_user_common_info from seahub.api2.utils import api_error, get_user_common_info
from .models import FileParticipant from .models import FileParticipant
from seahub.utils import normalize_file_path, is_valid_username from seahub.utils import normalize_file_path, is_valid_username
from seahub.views import check_folder_permission from seahub.views import check_folder_permission
from pysearpc import SearpcError
from seahub.base.accounts import User from seahub.base.accounts import User
from seahub.tags.models import FileUUIDMap
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class FileParticipantsView(APIView): class FileParticipantsView(APIView):
authentication_classes = (TokenAuthentication, SessionAuthentication) authentication_classes = (TokenAuthentication, SessionAuthentication)
permission_classes = (IsAuthenticated, IsRepoAccessible) permission_classes = (IsAuthenticated,)
throttle_classes = (UserRateThrottle,) throttle_classes = (UserRateThrottle,)
def get(self, request, repo_id, format=None): def get(self, request, repo_id, format=None):
"""List all participants of a file. """List all participants of a file.
""" """
# argument check # argument check
path = request.GET.get('path', '/').rstrip('/') path = request.GET.get('path')
if not path: if not path:
return api_error(status.HTTP_400_BAD_REQUEST, 'Invalid path.') return api_error(status.HTTP_400_BAD_REQUEST, 'path invalid.')
path = normalize_file_path(path) path = normalize_file_path(path)
# resource check # resource check
try: file_id = seafile_api.get_file_id_by_path(repo_id, path)
file_id = seafile_api.get_file_id_by_path(repo_id, path)
except SearpcError as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal error.')
if not file_id: if not file_id:
return api_error(status.HTTP_404_NOT_FOUND, 'File not found.') return api_error(status.HTTP_404_NOT_FOUND, 'File %s not found.' % path)
# permission check # permission check
if not check_folder_permission(request, repo_id, '/'): if not check_folder_permission(request, repo_id, '/'):
return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.') return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.')
# main # main
participant_list = [] try:
participant_queryset = FileParticipant.objects.get_by_file_path(repo_id, path) file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(repo_id, path, False)
for participant in participant_queryset: participant_list = []
participant_info = get_user_common_info(participant.username) participant_queryset = FileParticipant.objects.get_participants(file_uuid)
participant_list.append(participant_info)
for participant in participant_queryset:
participant_info = get_user_common_info(participant.username)
participant_list.append(participant_info)
except Exception as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal Server Error.')
return Response({'participant_list': participant_list}) return Response({'participant_list': participant_list})
@@ -63,26 +65,19 @@ class FileParticipantsView(APIView):
"""Post a participant of a file. """Post a participant of a file.
""" """
# argument check # argument check
path = request.GET.get('path', '/').rstrip('/') path = request.data.get('path')
if not path: if not path:
return api_error(status.HTTP_400_BAD_REQUEST, 'Invalid path.') return api_error(status.HTTP_400_BAD_REQUEST, 'path invalid.')
path = normalize_file_path(path) path = normalize_file_path(path)
email = request.data.get('email', '').lower() email = request.data.get('email')
if not email: if not email or not is_valid_username(email):
email = request.user.username return api_error(status.HTTP_400_BAD_REQUEST, 'email invalid.')
if not is_valid_username(email):
return api_error(status.HTTP_400_BAD_REQUEST, 'Invalid email.')
# resource check # resource check
try: file_id = seafile_api.get_file_id_by_path(repo_id, path)
file_id = seafile_api.get_file_id_by_path(repo_id, path)
except SearpcError as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal error.')
if not file_id: if not file_id:
return api_error(status.HTTP_404_NOT_FOUND, 'File not found.') return api_error(status.HTTP_404_NOT_FOUND, 'File %s not found.' % path)
try: try:
user = User.objects.get(email=email) user = User.objects.get(email=email)
@@ -93,52 +88,62 @@ class FileParticipantsView(APIView):
if not check_folder_permission(request, repo_id, '/'): if not check_folder_permission(request, repo_id, '/'):
return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.') return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.')
request.user = user if not seafile_api.check_permission_by_path(repo_id, '/', user.username):
if not check_folder_permission(request, repo_id, '/'): return api_error(status.HTTP_403_FORBIDDEN, _('%s Permission denied.') % email)
return api_error(status.HTTP_403_FORBIDDEN, 'User %s permission denied.' % email)
# main # main
if FileParticipant.objects.get_by_file_path_and_username(repo_id, path, email): try:
return api_error(status.HTTP_400_BAD_REQUEST, 'Participant %s already exists.' % email) file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(repo_id, path, False)
FileParticipant.objects.add_by_file_path_and_username(repo_id, path, email) if FileParticipant.objects.get_participant(file_uuid, email):
participant = get_user_common_info(email) return api_error(status.HTTP_409_CONFLICT, _('Participant %s already exists.') % email)
FileParticipant.objects.add_participant(file_uuid, email)
participant = get_user_common_info(email)
except Exception as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal Server Error.')
return Response(participant, status=201) return Response(participant, status=201)
class FileParticipantView(APIView): class FileParticipantView(APIView):
authentication_classes = (TokenAuthentication, SessionAuthentication) authentication_classes = (TokenAuthentication, SessionAuthentication)
permission_classes = (IsAuthenticated, IsRepoAccessible) permission_classes = (IsAuthenticated,)
throttle_classes = (UserRateThrottle,) throttle_classes = (UserRateThrottle,)
def delete(self, request, repo_id, format=None): def delete(self, request, repo_id, format=None):
"""Delete a participant """Delete a participant
""" """
# argument check # argument check
path = request.GET.get('path', '/').rstrip('/') path = request.data.get('path')
if not path: if not path:
return api_error(status.HTTP_400_BAD_REQUEST, 'Invalid path.') return api_error(status.HTTP_400_BAD_REQUEST, 'path invalid.')
path = normalize_file_path(path) path = normalize_file_path(path)
email = request.data.get('email', '').lower() email = request.data.get('email')
if not email or not is_valid_username(email): if not email or not is_valid_username(email):
return api_error(status.HTTP_400_BAD_REQUEST, 'Invalid email.') return api_error(status.HTTP_400_BAD_REQUEST, 'email invalid.')
# resource check # resource check
try: file_id = seafile_api.get_file_id_by_path(repo_id, path)
file_id = seafile_api.get_file_id_by_path(repo_id, path)
except SearpcError as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal error.')
if not file_id: if not file_id:
return api_error(status.HTTP_404_NOT_FOUND, 'File not found.') return api_error(status.HTTP_404_NOT_FOUND, 'File %s not found.' % path)
# permission check # permission check
if not check_folder_permission(request, repo_id, '/'): if not check_folder_permission(request, repo_id, '/'):
return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.') return api_error(status.HTTP_403_FORBIDDEN, 'Permission denied.')
# main # main
FileParticipant.objects.delete_by_file_path_and_username(repo_id, path, email) try:
file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(repo_id, path, False)
if not FileParticipant.objects.get_participant(file_uuid, email):
return api_error(status.HTTP_404_NOT_FOUND, 'Participant %s not found.' % email)
FileParticipant.objects.delete_participant(file_uuid, email)
except Exception as e:
logger.error(e)
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, 'Internal Server Error.')
return Response({'success': True}) return Response({'success': True})

View File

@@ -1,7 +1,6 @@
# Copyright (c) 2012-2019 Seafile Ltd. # Copyright (c) 2012-2019 Seafile Ltd.
import os
import logging import logging
from django.db import models from django.db import models
from seahub.tags.models import FileUUIDMap from seahub.tags.models import FileUUIDMap
from seahub.base.fields import LowerCaseCharField from seahub.base.fields import LowerCaseCharField
@@ -12,43 +11,26 @@ logger = logging.getLogger(__name__)
class FileParticipantManager(models.Manager): class FileParticipantManager(models.Manager):
def _get_file_uuid_map(self, repo_id, file_path): def get_participants(self, uuid):
parent_path = os.path.dirname(file_path) objs = self.filter(uuid=uuid)
item_name = os.path.basename(file_path)
file_uuid_map = FileUUIDMap.objects.get_or_create_fileuuidmap( return objs
repo_id, parent_path, item_name, False)
return file_uuid_map def get_participant(self, uuid, username):
def add_by_file_path_and_username(self, repo_id, file_path, username):
uuid = self._get_file_uuid_map(repo_id, file_path)
if self.filter(uuid=uuid, username=username).exists(): if self.filter(uuid=uuid, username=username).exists():
return self.filter(uuid=uuid, username=username)[0] return self.filter(uuid=uuid, username=username)[0]
return None
def add_participant(self, uuid, username):
obj = self.model(uuid=uuid, username=username) obj = self.model(uuid=uuid, username=username)
obj.save(using=self._db) obj.save(using=self._db)
return obj return obj
def get_by_file_path_and_username(self, repo_id, file_path, username): def delete_participant(self, uuid, username):
uuid = self._get_file_uuid_map(repo_id, file_path)
try:
obj = self.get(uuid=uuid, username=username)
return obj
except self.model.DoesNotExist:
return None
def delete_by_file_path_and_username(self, repo_id, file_path, username):
uuid = self._get_file_uuid_map(repo_id, file_path)
self.filter(uuid=uuid, username=username).delete() self.filter(uuid=uuid, username=username).delete()
def get_by_file_path(self, repo_id, file_path):
uuid = self._get_file_uuid_map(repo_id, file_path)
objs = self.filter(uuid=uuid)
return objs
class FileParticipant(models.Model): class FileParticipant(models.Model):
""" """

View File

@@ -1,12 +1,22 @@
import logging
from .models import FileParticipant from .models import FileParticipant
from seahub.tags.models import FileUUIDMap
logger = logging.getLogger(__name__)
def list_file_participants_username(repo_id, path): def list_file_participants(repo_id, path):
""" return participants username list """ return participants username list
""" """
username_list = [] username_list = []
file_participant_queryset = FileParticipant.objects.get_by_file_path(repo_id, path) try:
for participant in file_participant_queryset: file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(repo_id, path, False)
username_list.append(participant.username)
participant_queryset = FileParticipant.objects.get_participants(file_uuid)
for participant in participant_queryset:
username_list.append(participant.username)
except Exception as e:
logger.error(e)
return username_list return username_list

View File

@@ -25,7 +25,7 @@ from seahub.utils import normalize_cache_key
from seahub.utils.timeutils import datetime_to_isoformat_timestr from seahub.utils.timeutils import datetime_to_isoformat_timestr
from seahub.constants import HASH_URLS from seahub.constants import HASH_URLS
from seahub.drafts.models import DraftReviewer from seahub.drafts.models import DraftReviewer
from seahub.file_participants.utils import list_file_participants_username from seahub.file_participants.utils import list_file_participants
# Get an instance of a logger # Get an instance of a logger
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -988,7 +988,7 @@ def comment_file_successful_cb(sender, **kwargs):
comment = kwargs['comment'] comment = kwargs['comment']
author = kwargs['author'] author = kwargs['author']
notify_users = list_file_participants_username(repo.id, file_path) notify_users = list_file_participants(repo.id, file_path)
notify_users = [x for x in notify_users if x != author] notify_users = [x for x in notify_users if x != author]
for u in notify_users: for u in notify_users:
detail = file_comment_msg_to_json(repo.id, file_path, author, comment) detail = file_comment_msg_to_json(repo.id, file_path, author, comment)

View File

@@ -1,11 +1,13 @@
# Copyright (c) 2012-2016 Seafile Ltd. # Copyright (c) 2012-2016 Seafile Ltd.
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
import os
import uuid import uuid
import hashlib import hashlib
from django.db import models from django.db import models
from seahub.base.fields import LowerCaseCharField from seahub.base.fields import LowerCaseCharField
from seahub.utils import normalize_file_path,normalize_dir_path
########## Manager ########## Manager
@@ -61,6 +63,19 @@ class FileUUIDMapManager(models.Manager):
) )
return uuids return uuids
def get_or_create_fileuuidmap_by_path(self, repo_id, path, is_dir):
if is_dir:
path = normalize_dir_path(path)
else:
path = normalize_file_path(path)
path = path.rstrip('/')
parent_path = os.path.dirname(path)
obj_name = os.path.basename(path)
return self.get_or_create_fileuuidmap(repo_id, parent_path, obj_name, is_dir)
class TagsManager(models.Manager): class TagsManager(models.Manager):
def get_or_create_tag(self, tagname): def get_or_create_tag(self, tagname):

View File

@@ -143,7 +143,7 @@ from seahub.api2.endpoints.admin.file_scan_records import AdminFileScanRecords
from seahub.api2.endpoints.admin.notifications import AdminNotificationsView from seahub.api2.endpoints.admin.notifications import AdminNotificationsView
from seahub.api2.endpoints.admin.work_weixin import AdminWorkWeixinDepartments, \ from seahub.api2.endpoints.admin.work_weixin import AdminWorkWeixinDepartments, \
AdminWorkWeixinDepartmentMembers, AdminWorkWeixinUsersBatch AdminWorkWeixinDepartmentMembers, AdminWorkWeixinUsersBatch
from seahub.file_participants.views import FileParticipantsView, FileParticipantView from seahub.file_participants.api import FileParticipantsView, FileParticipantView
urlpatterns = [ urlpatterns = [
url(r'^accounts/', include('seahub.base.registration_urls')), url(r'^accounts/', include('seahub.base.registration_urls')),

View File

@@ -8,6 +8,7 @@ from seahub.base.models import FileComment
from seahub.notifications.models import UserNotification from seahub.notifications.models import UserNotification
from seahub.test_utils import BaseTestCase from seahub.test_utils import BaseTestCase
from seahub.file_participants.models import FileParticipant from seahub.file_participants.models import FileParticipant
from seahub.tags.models import FileUUIDMap
class FileCommentsTest(BaseTestCase): class FileCommentsTest(BaseTestCase):
def setUp(self): def setUp(self):
@@ -92,8 +93,8 @@ class FileCommentsTest(BaseTestCase):
# share repo and add participant # share repo and add participant
seafile_api.share_repo(self.repo.id, self.user.username, self.admin.username, 'rw') seafile_api.share_repo(self.repo.id, self.user.username, self.admin.username, 'rw')
FileParticipant.objects.add_by_file_path_and_username( file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(self.repo.id, self.file, False)
repo_id=self.repo.id, file_path=self.file, username=self.admin.username) FileParticipant.objects.add_participant(file_uuid, self.admin.username)
resp = self.client.post(self.endpoint, { resp = self.client.post(self.endpoint, {
'comment': 'new comment' 'comment': 'new comment'

View File

@@ -1,48 +0,0 @@
import json
from seaserv import seafile_api
from django.core.urlresolvers import reverse
from seahub.file_participants.models import FileParticipant
from seahub.test_utils import BaseTestCase
from tests.common.utils import randstring
class FileParticipantTest(BaseTestCase):
def setUp(self):
self.tmp_user = self.create_user()
self.login_as(self.user)
self.url = reverse('api-v2.1-file-participant', args=[self.repo.id]) + '?path=' + self.file
# share repo and add participant
seafile_api.share_repo(self.repo.id, self.user.username, self.tmp_user.username, 'rw')
FileParticipant.objects.add_by_file_path_and_username(
repo_id=self.repo.id, file_path=self.file, username=self.tmp_user.username)
def tearDown(self):
self.remove_repo()
self.remove_user(self.tmp_user.email)
def test_can_delete(self):
data = 'email=' + self.tmp_user.username
resp = self.client.delete(self.url, data, 'application/x-www-form-urlencoded')
self.assertEqual(200, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['success']
def test_can_not_delete_by_not_exists_path(self):
invalid_path = randstring(5)
data = 'email=' + self.tmp_user.username
resp = self.client.delete(self.url + invalid_path, data, 'application/x-www-form-urlencoded')
self.assertEqual(404, resp.status_code)
def test_can_not_delete_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
data = 'email=' + self.tmp_user.username
resp = self.client.delete(self.url, data, 'application/x-www-form-urlencoded')
self.assertEqual(403, resp.status_code)

View File

@@ -1,88 +0,0 @@
import json
from seaserv import seafile_api
from django.core.urlresolvers import reverse
from seahub.file_participants.models import FileParticipant
from seahub.test_utils import BaseTestCase
from tests.common.utils import randstring
class FileParticipantsTest(BaseTestCase):
def setUp(self):
self.tmp_user = self.create_user()
self.login_as(self.user)
self.url = reverse('api-v2.1-file-participants', args=[self.repo.id]) + '?path=' + self.file
# share repo and add participant
seafile_api.share_repo(self.repo.id, self.user.username, self.tmp_user.username, 'rw')
FileParticipant.objects.add_by_file_path_and_username(
repo_id=self.repo.id, file_path=self.file, username=self.user.username)
def tearDown(self):
self.remove_repo()
self.remove_user(self.tmp_user.email)
def test_can_list(self):
resp = self.client.get(self.url)
self.assertEqual(200, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['participant_list']
assert json_resp['participant_list'][0]
assert json_resp['participant_list'][0]['email'] == self.user.username
assert json_resp['participant_list'][0]['avatar_url']
assert json_resp['participant_list'][0]['contact_email']
assert json_resp['participant_list'][0]['name']
def test_can_not_list_by_not_exists_path(self):
invalid_path = randstring(5)
resp = self.client.get(self.url + invalid_path)
self.assertEqual(404, resp.status_code)
def test_can_not_list_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
resp = self.client.get(self.url)
self.assertEqual(403, resp.status_code)
def test_can_post(self):
resp = self.client.post(self.url, {
'email': self.tmp_user.username
})
self.assertEqual(201, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['email'] == self.tmp_user.username
assert json_resp['avatar_url']
assert json_resp['contact_email']
assert json_resp['name']
def test_can_not_post_by_not_exists_path(self):
invalid_path = randstring(5)
resp = self.client.post(self.url + invalid_path, {
'email': self.tmp_user.username
})
self.assertEqual(404, resp.status_code)
def test_can_not_post_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
resp = self.client.post(self.url, {
'email': self.tmp_user.username
})
self.assertEqual(403, resp.status_code)
def test_can_not_post_by_not_exists_user(self):
invalid_email = randstring(5) + '@seafile.com'
resp = self.client.post(self.url, {
'email': invalid_email
})
self.assertEqual(404, resp.status_code)
def test_can_not_post_by_invalid_email_permission(self):
resp = self.client.post(self.url, {
'email': self.admin.username
})
self.assertEqual(403, resp.status_code)

View File

@@ -0,0 +1,145 @@
import json
from seaserv import seafile_api
from django.core.urlresolvers import reverse
from seahub.file_participants.models import FileParticipant
from seahub.test_utils import BaseTestCase
from tests.common.utils import randstring
from seahub.tags.models import FileUUIDMap
class FileParticipantsTest(BaseTestCase):
def setUp(self):
self.tmp_user = self.create_user()
self.login_as(self.user)
self.url = reverse('api-v2.1-file-participants', args=[self.repo.id])
# share repo and add participant
seafile_api.share_repo(self.repo.id, self.user.username, self.tmp_user.username, 'rw')
self.file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(self.repo.id, self.file, False)
FileParticipant.objects.add_participant(self.file_uuid, self.user.username)
def tearDown(self):
self.remove_repo()
self.remove_user(self.tmp_user.email)
def test_can_list(self):
resp = self.client.get(self.url + '?path=' + self.file)
self.assertEqual(200, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['participant_list']
assert json_resp['participant_list'][0]
assert json_resp['participant_list'][0]['email'] == self.user.username
assert json_resp['participant_list'][0]['avatar_url']
assert json_resp['participant_list'][0]['contact_email']
assert json_resp['participant_list'][0]['name']
def test_can_not_list_by_not_exists_path(self):
invalid_path = self.file + randstring(5)
resp = self.client.get(self.url + '?path=' + invalid_path)
self.assertEqual(404, resp.status_code)
def test_can_not_list_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
resp = self.client.get(self.url + '?path=' + self.file)
self.assertEqual(403, resp.status_code)
def test_can_post(self):
resp = self.client.post(self.url, {
'email': self.tmp_user.username,
'path': self.file,
})
self.assertEqual(201, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['email'] == self.tmp_user.username
assert json_resp['avatar_url']
assert json_resp['contact_email']
assert json_resp['name']
assert FileParticipant.objects.filter(
uuid=self.file_uuid, username=self.tmp_user.username).count() == 1
def test_can_not_post_by_not_exists_path(self):
invalid_path = self.file + randstring(5)
resp = self.client.post(self.url, {
'email': self.tmp_user.username,
'path': invalid_path,
})
self.assertEqual(404, resp.status_code)
def test_can_not_post_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
resp = self.client.post(self.url, {
'email': self.tmp_user.username,
'path': self.file,
})
self.assertEqual(403, resp.status_code)
def test_can_not_post_by_not_exists_user(self):
invalid_email = randstring(5) + '@seafile.com'
resp = self.client.post(self.url, {
'email': invalid_email,
'path': self.file,
})
self.assertEqual(404, resp.status_code)
def test_can_not_post_by_invalid_email_permission(self):
resp = self.client.post(self.url, {
'email': self.admin.username,
'path': self.file,
})
self.assertEqual(403, resp.status_code)
class FileParticipantTest(BaseTestCase):
def setUp(self):
self.tmp_user = self.create_user()
self.login_as(self.user)
self.url = reverse('api-v2.1-file-participant', args=[self.repo.id])
# share repo and add participant
seafile_api.share_repo(self.repo.id, self.user.username, self.tmp_user.username, 'rw')
self.file_uuid = FileUUIDMap.objects.get_or_create_fileuuidmap_by_path(self.repo.id, self.file, False)
FileParticipant.objects.add_participant(self.file_uuid, self.user.username)
FileParticipant.objects.add_participant(self.file_uuid, self.tmp_user.username)
def tearDown(self):
self.remove_repo()
self.remove_user(self.tmp_user.email)
def test_can_delete(self):
assert FileParticipant.objects.filter(
uuid=self.file_uuid, username=self.tmp_user.username).count() == 1
data = 'email=' + self.tmp_user.username + '&path=' + self.file
resp = self.client.delete(self.url, data, 'application/x-www-form-urlencoded')
self.assertEqual(200, resp.status_code)
json_resp = json.loads(resp.content)
assert json_resp['success']
assert FileParticipant.objects.filter(
uuid=self.file_uuid, username=self.tmp_user.username).count() == 0
def test_can_not_delete_by_not_exists_path(self):
invalid_path = self.file + randstring(5)
data = 'email=' + self.tmp_user.username + '&path=' + invalid_path
resp = self.client.delete(self.url + invalid_path, data, 'application/x-www-form-urlencoded')
self.assertEqual(404, resp.status_code)
def test_can_not_delete_by_invalid_user_permission(self):
self.logout()
self.login_as(self.admin)
data = 'email=' + self.tmp_user.username + '&path=' + self.file
resp = self.client.delete(self.url, data, 'application/x-www-form-urlencoded')
self.assertEqual(403, resp.status_code)