haiwen/seahub
1
0
Fork 0
mirror of https://github.com/haiwen/seahub.git synced 2025-09-09 10:50:24 +00:00
Code Issues Releases Wiki Activity

sysadmin reconstruct add invitations api (#4179)

Browse Source 
* sysadmin reconstruct add invitations api
This commit is contained in:
Leo
2019-10-24 11:30:12 +08:00
committed by lian
parent c611994bc0
commit 485f14b3ff
3 changed files with 171 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
seahub/api2/endpoints/admin/invitations.py
View File

@@ -1,5 +1,6 @@
# Copyright (c) 2012-2016 Seafile Ltd.
from django.contrib import messages
import logging
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response
@@ -10,13 +11,93 @@ from seahub.api2.authentication import TokenAuthentication
from seahub.api2.throttling import UserRateThrottle
from seahub.api2.utils import api_error
from seahub.invitations.models import Invitation
from seahub.settings import ENABLE_GUEST_INVITATION
from seahub.utils.timeutils import datetime_to_isoformat_timestr
from seahub.base.templatetags.seahub_tags import email2nickname, email2contact_email
logger = logging.getLogger(__name__)
class InvitationsView(APIView):
class AdminInvitations(APIView):
authentication_classes = (TokenAuthentication, SessionAuthentication)
throttle_classes = (UserRateThrottle, )
permission_classes = (IsAdminUser, )
def get(self, request):
""" List invitations
Permission checking:
1. only admin can perform this action.
"""
if not ENABLE_GUEST_INVITATION:
error_msg = 'invitation not enabled.'
return api_error(status.HTTP_404_NOT_FOUND, error_msg)
try:
current_page = int(request.GET.get('page', '1'))
per_page = int(request.GET.get('per_page', '100'))
except ValueError:
current_page = 1
per_page = 100
start = per_page * (current_page - 1)
invitations = Invitation.objects.all().order_by('-invite_time')[start:start + per_page]
count = Invitation.objects.count()
# Use dict to reduce memcache fetch cost in large for-loop.
inviter_email_set = set()
accepter_email_set = set()
for invitation in invitations:
inviter_email_set.add(invitation.inviter)
accepter_email_set.add(invitation.accepter)
inviter_nickname_dict = {}
inviter_contact_email_dict = {}
accepter_nickname_dict = {}
accepter_contact_email_dict = {}
for e in inviter_email_set:
if e not in inviter_nickname_dict:
inviter_nickname_dict[e] = email2nickname(e)
if e not in inviter_contact_email_dict:
inviter_contact_email_dict[e] = email2contact_email(e)
for e in accepter_email_set:
if e not in accepter_nickname_dict:
accepter_nickname_dict[e] = email2nickname(e)
if e not in accepter_contact_email_dict:
accepter_contact_email_dict[e] = email2contact_email(e)
invitations_info = []
for invitation in invitations:
data = {}
data['id'] = invitation.id
data['token'] = invitation.token
inviter_email = invitation.inviter
data['inviter_email'] = inviter_email
data['inviter_name'] = inviter_nickname_dict.get(inviter_email, '')
data['inviter_contact_email'] = inviter_contact_email_dict.get(inviter_email, '')
accepter_email = invitation.accepter
data['accepter_email'] = accepter_email
data['accepter_name'] = accepter_nickname_dict.get(accepter_email, '')
data['accepter_contact_email'] = accepter_contact_email_dict.get(accepter_email, '')
data['invite_type'] = invitation.invite_type
data['invite_time'] = datetime_to_isoformat_timestr(invitation.invite_time)
data['accept_time'] = datetime_to_isoformat_timestr(invitation.accept_time)
data['expire_time'] = datetime_to_isoformat_timestr(invitation.expire_time)
data['is_expired'] = invitation.is_expired()
invitations_info.append(data)
resp = {
'invitation_list': invitations_info,
'total_count': count
}
return Response(resp)
def delete(self, request):
_type = request.GET.get('type', '')
if _type == "" or _type not in ["expired"]:
@@ -25,3 +106,35 @@ class InvitationsView(APIView):
if _type == "expired":
Invitation.objects.delete_all_expire_invitation()
return Response(status.HTTP_200_OK)
class AdminInvitation(APIView):
authentication_classes = (TokenAuthentication, SessionAuthentication)
throttle_classes = (UserRateThrottle, )
permission_classes = (IsAdminUser, )
def delete(self, request, token):
""" delete a invitation
Permission checking:
1. only admin can perform this action.
"""
if not ENABLE_GUEST_INVITATION:
error_msg = 'invitation not enabled.'
return api_error(status.HTTP_404_NOT_FOUND, error_msg)
try:
invitation = Invitation.objects.get(token=token)
except Invitation.DoesNotExist:
return Response({'success': True})
try:
invitation.delete()
except Exception as e:
logger.error(e)
error_msg = 'Internal Server Error'
return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg)
return Response({'success': True})

5
seahub/urls.py
View File

@@ -136,7 +136,7 @@ from seahub.api2.endpoints.admin.org_stats import AdminOrgStatsTraffic
from seahub.api2.endpoints.admin.logo import AdminLogo
from seahub.api2.endpoints.admin.favicon import AdminFavicon
from seahub.api2.endpoints.admin.license import AdminLicense
from seahub.api2.endpoints.admin.invitations import InvitationsView as AdminInvitationsView
from seahub.api2.endpoints.admin.invitations import AdminInvitations, AdminInvitation
from seahub.api2.endpoints.admin.library_history import AdminLibraryHistoryLimit
from seahub.api2.endpoints.admin.login_bg_image import AdminLoginBgImage
from seahub.api2.endpoints.admin.admin_role import AdminAdminRole
@@ -553,7 +553,8 @@ urlpatterns = [
url(r'^api/v2.1/admin/login-background-image/$', AdminLoginBgImage.as_view(), name='api-v2.1-admin-login-background-image'),
## admin::invitations
url(r'^api/v2.1/admin/invitations/$', AdminInvitationsView.as_view(), name='api-v2.1-admin-invitations'),
url(r'^api/v2.1/admin/invitations/$', AdminInvitations.as_view(), name='api-v2.1-admin-invitations'),
url(r'^api/v2.1/admin/invitations/(?P<token>[a-f0-9]{32})/$', AdminInvitation.as_view(), name='api-v2.1-admin-invitation'),
url(r'^avatar/', include('seahub.avatar.urls')),
url(r'^notification/', include('seahub.notifications.urls')),

57
tests/api/endpoints/admin/test_invitations.py
View File

@@ -1,4 +1,5 @@
import time
import json
from mock import patch
from django.utils import timezone
@@ -13,8 +14,7 @@ from seahub.invitations import models
class InvitationsTest(BaseTestCase):
def setUp(self):
self.login_as(self.admin)
self.delete_url = reverse('api-v2.1-admin-invitations')
self.url = reverse('api-v2.1-admin-invitations')
@patch.object(CanInviteGuest, 'has_permission')
@patch.object(UserPermissions, 'can_invite_guest')
@@ -31,7 +31,7 @@ class InvitationsTest(BaseTestCase):
self.assertEqual(2, new_invitations_number-invitations_number)
time.sleep(2)
resp = self.client.delete(self.delete_url+"?type=expired")
resp = self.client.delete(self.url+"?type=expired")
self.assertEqual(200, resp.status_code)
self.assertEqual(invitations_number, len(Invitation.objects.all()))
@@ -43,7 +43,56 @@ class InvitationsTest(BaseTestCase):
expire_time=timezone.now())
entry.save()
def test_get_invitations(self):
self.login_as(self.admin)
resp = self.client.get(self.url)
self.assertEqual(200, resp.status_code)
json_resp = json.loads(resp.content)
assert type(json_resp['invitation_list']) is list
def test_get_invitations_permision_denied(self):
self.login_as(self.user)
resp = self.client.get(self.url)
self.assertEqual(403, resp.status_code)
def test_invalid_args(self):
resp = self.client.delete(self.delete_url+"?type=expired122")
self.login_as(self.admin)
resp = self.client.delete(self.url+"?type=expired122")
self.assertEqual(400, resp.status_code)
class InvitationTest(BaseTestCase):
def setUp(self):
pass
def _add_invitations(self, email):
token = models.gen_token(max_length=32)
entry = models.Invitation(token=token,
inviter=self.admin,
accepter=email,
invite_type=models.GUEST,
expire_time=timezone.now())
entry.save()
return token
def _remove_invitation(self, token):
invitation = Invitation.objects.get(token=token)
invitation.delete()
def test_can_delete(self):
self.login_as(self.admin)
token = self._add_invitations('test@noway.com')
url = reverse('api-v2.1-admin-invitation', args=[token])
resp = self.client.delete(url)
self.assertEqual(200, resp.status_code)
def test_delete_share_link_with_invalid_permission(self):
self.login_as(self.user)
token = self._add_invitations('test@noway.com')
url = reverse('api-v2.1-admin-invitation', args=[token])
resp = self.client.delete(url)
self.assertEqual(403, resp.status_code)
self._remove_invitation(token)