haiwen/seahub
1
0
Fork 0
mirror of https://github.com/haiwen/seahub.git synced 2025-04-28 03:10:45 +00:00
Code Issues Releases Wiki Activity

add_api_admin_notifications_with_test_cases (#3246)

Browse Source 
* add_api_admin_notifications_with_test_cases
This commit is contained in:
Leo 2019-04-13 18:15:40 +08:00 committed by lian
parent c613d4f248
commit 66b668fd21
4 changed files with 189 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

87
seahub/api2/endpoints/admin/notifications.py Normal file
View File

@ -0,0 +1,87 @@
import logging
from rest_framework.authentication import SessionAuthentication
from rest_framework.permissions import IsAdminUser, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework import status
from django.core.cache import cache
from seahub.api2.authentication import TokenAuthentication
from seahub.api2.throttling import UserRateThrottle
from seahub.notifications.models import UserNotification
from seahub.notifications.models import get_cache_key_of_unseen_notifications
from seahub.notifications.utils import update_notice_detail
from seahub.utils.timeutils import datetime_to_isoformat_timestr
from seahub.api2.utils import api_error
logger = logging.getLogger(__name__)
class AdminNotificationsView(APIView):
authentication_classes = (TokenAuthentication, SessionAuthentication)
throttle_classes = (UserRateThrottle,)
permission_classes = (IsAuthenticated, IsAdminUser)
def get(self, request):
"""
list all notifications / filt by user name
Permission checking:
1.login and is admin user.
"""
user_name = request.GET.get('username', '')
# argument check
# permission check
# permission_classes will return 403, if not admin user
# prepare default values
result = {}
try:
per_page = int(request.GET.get('per_page', ''))
page = int(request.GET.get('page', ''))
except ValueError:
per_page = 100
page = 1
start = (page - 1) * per_page
end = page * per_page
total_count = 0
# resource check
if user_name != '':
# return all notifications of a user given by name
total_count = UserNotification.objects.get_user_notifications(user_name).count()
notification_list = UserNotification.objects.get_user_notifications(user_name)[start:end]
else:
# return all notifications of all users
total_count = UserNotification.objects.get_all_notifications().count()
notification_list = UserNotification.objects.get_all_notifications()[start:end]
# notification does not exist, return an empty list
if not notification_list:
result['count'] = 0
result['notification_list'] = []
return Response(result)
result_notification = update_notice_detail(request, notification_list)
notification_list = []
for i in result_notification:
notification_info = {}
notification_info['id'] = i.id
notification_info['type'] = i.msg_type
notification_info['time'] = datetime_to_isoformat_timestr(i.timestamp)
if i.detail is not None:
notification_info['detail'] = i.detail
else:
notification_info['detail'] = {}
notification_list.append(notification_info)
result['count'] = total_count
result['notification_list'] = notification_list
return Response(result)

65
seahub/notifications/utils.py
View File

@ -9,7 +9,7 @@ from seaserv import ccnet_api, seafile_api
from seahub.notifications.models import Notification
from seahub.notifications.settings import NOTIFICATION_CACHE_TIMEOUT
from seahub.avatar.templatetags.avatar_tags import api_avatar_url
from seahub.base.templatetags.seahub_tags import email2nickname
from seahub.base.templatetags.seahub_tags import email2nickname, email2contact_email
logger = logging.getLogger(__name__)
@ -46,14 +46,14 @@ def update_notice_detail(request, notices):
notice.detail = None
else:
d.pop('org_id')
share_from_user_name = email2nickname(d['share_from'])
url, is_default, date_uploaded = api_avatar_url(d['share_from'], 32)
share_from_user_email = d.pop('share_from')
url, is_default, date_uploaded = api_avatar_url(share_from_user_email, 32)
d['repo_name'] = repo.name
d['repo_id'] = repo.id
d['share_from_user_email'] = d.pop('share_from')
d['share_from_user_name'] = share_from_user_name
d['share_from_user_name'] = email2nickname(share_from_user_email)
d['share_from_user_email'] = share_from_user_email
d['share_from_user_contact_email'] = email2contact_email(share_from_user_email)
d['share_from_user_avatar_url'] = request.build_absolute_uri(url)
notice.detail = d
except Exception as e:
@ -82,10 +82,11 @@ def update_notice_detail(request, notices):
notice.detail = None
else:
d.pop('org_id')
share_from_user_name = email2nickname(d['share_from'])
url, is_default, date_uploaded = api_avatar_url(d['share_from'], 32)
d['share_from_user_email'] = d.pop('share_from')
d['share_from_user_name'] = share_from_user_name
share_from_user_email = d.pop('share_from')
url, is_default, date_uploaded = api_avatar_url(share_from_user_email, 32)
d['share_from_user_name'] = email2nickname(share_from_user_email)
d['share_from_user_email'] = share_from_user_email
d['share_from_user_contact_email'] = email2contact_email(share_from_user_email)
d['share_from_user_avatar_url'] = request.build_absolute_uri(url)
d['repo_name'] = repo.name
d['repo_id'] = repo.id
@ -103,10 +104,11 @@ def update_notice_detail(request, notices):
if group is None:
notice.detail = None
else:
group_staff_name = email2nickname(d['group_staff'])
url, is_default, date_uploaded = api_avatar_url(d['group_staff'], 32)
d['group_staff_email'] = d.pop('group_staff')
d['group_staff_name'] = group_staff_name
group_staff_email = d.pop('group_staff')
url, is_default, date_uploaded = api_avatar_url(group_staff_email, 32)
d['group_staff_name'] = email2nickname(group_staff_email)
d['group_staff_email'] = group_staff_email
d['group_staff_contact_email'] = email2contact_email(group_staff_email)
d['group_staff_avatar_url'] = request.build_absolute_uri(url)
d['group_name'] = group.group_name
@ -117,10 +119,11 @@ def update_notice_detail(request, notices):
elif notice.is_draft_comment_msg():
try:
d = json.loads(notice.detail)
author_name = email2nickname(d['author'])
url, is_default, date_uploaded = api_avatar_url(d['author'], 32)
d['author_name'] = author_name
d['author_email'] = d.pop('author')
author_email = d.pop('author')
url, is_default, date_uploaded = api_avatar_url(author_email, 32)
d['author_name'] = email2nickname(author_email)
d['author_email'] = author_email
d['author_context_email'] = email2contact_email(author_email)
d['author_avatar_url'] = request.build_absolute_uri(url)
notice.detail = d
@ -136,10 +139,10 @@ def update_notice_detail(request, notices):
notice.detail = None
else:
d.pop('org_id')
repo_owner_email = d['repo_owner']
repo_owner_name = email2nickname(repo_owner_email)
d['transfer_from_user_email'] = d.pop('repo_owner')
d['transfer_from_user_name'] = repo_owner_name
repo_owner_email = d.pop('repo_owner')
d['transfer_from_user_name'] = email2nickname(repo_owner_email)
d['transfer_from_user_email'] = repo_owner_email
d['transfer_from_user_contact_email'] = email2contact_email(repo_owner_email)
url, is_default, date_uploaded = api_avatar_url(repo_owner_email, 32)
d['transfer_from_user_avatar_url'] = request.build_absolute_uri(url)
notice.detail = d
@ -151,10 +154,11 @@ def update_notice_detail(request, notices):
try:
d = json.loads(notice.detail)
d.pop('to_user')
request_user_name = email2nickname(d['from_user'])
url, is_default, date_uploaded = api_avatar_url(d['from_user'], 32)
d['request_user_name'] = request_user_name
d['request_user_email'] = d.pop('from_user')
request_user_email = d.pop('from_user')
url, is_default, date_uploaded = api_avatar_url(request_user_email, 32)
d['request_user_name'] = email2nickname(request_user_email)
d['request_user_email'] = request_user_email
d['request_user_contact_email'] = email2contact_email(request_user_email)
d['request_user_avatat_url'] = request.build_absolute_uri(url)
notice.detail = d
except Exception as e:
@ -211,12 +215,13 @@ def update_notice_detail(request, notices):
if repo is None or not seafile_api.get_file_id_by_path(repo.id, file_path):
notice.detail = None
else:
author_name = email2nickname(d['author'])
author_email = d.pop('author')
file_name = os.path.basename(file_path)
url, is_default, date_uploaded = api_avatar_url(d['author'], 32)
url, is_default, date_uploaded = api_avatar_url(author_email, 32)
d['author_avatar_url'] = request.build_absolute_uri(url)
d['author_email'] = d.pop('author')
d['author_name'] = author_name
d['author_name'] = email2nickname(author_email)
d['author_email'] = author_email
d['author_contact_email'] = email2contact_email(author_email)
d['file_name'] = file_name
notice.detail = d
except Exception as e:

4
seahub/urls.py
View File

@ -131,6 +131,7 @@ from seahub.api2.endpoints.admin.address_book.groups import AdminAddressBookGrou
from seahub.api2.endpoints.admin.group_owned_libraries import AdminGroupOwnedLibraries, \
AdminGroupOwnedLibrary
from seahub.api2.endpoints.admin.file_scan_records import AdminFileScanRecords
from seahub.api2.endpoints.admin.notifications import AdminNotificationsView
urlpatterns = [
url(r'^accounts/', include('seahub.base.registration_urls')),
@ -506,6 +507,9 @@ urlpatterns = [
## admin::file-scan-records
url(r'^api/v2.1/admin/file-scan-records/$', AdminFileScanRecords.as_view(), name='api-v2.1-admin-file-scan-records'),
## admin::notifications
url(r'^api/v2.1/admin/notifications/$', AdminNotificationsView.as_view(), name='api-2.1-admin-notifications'),
### system admin ###
url(r'^sysadmin/$', sysadmin, name='sysadmin'),
url(r'^sys/settings/$', sys_settings, name='sys_settings'),

63
tests/api/endpoints/admin/test_notifications.py Normal file
View File

@ -0,0 +1,63 @@
import json
from seahub.test_utils import BaseTestCase
from seahub.notifications.models import UserNotification
from seahub.base.accounts import UserManager
class AdminNotificationsTest(BaseTestCase):
def setUp(self):
self.endpoint = '/api/v2.1/admin/notifications/'
self.user_name = self.user.username
self.admin_name = self.admin.username
def test_get_all_nofitications_as_admin(self):
self.login_as(self.admin)
notice1 = UserNotification.objects.add_user_message(self.user_name, 'test1')
notice2 = UserNotification.objects.add_user_message(self.admin_name, 'test2')
resp = self.client.get(self.endpoint)
json_resp = json.loads(resp.content)
self.assertEqual(200, resp.status_code)
assert json_resp['count'] == 2
def test_get_a_user_notification_by_name_as_admin(self):
self.login_as(self.admin)
new_user1 = UserManager().create_user(email='1@1.com', password='1')
new_user2 = UserManager().create_user(email='2@2.com', password='2')
notice1 = UserNotification.objects.add_user_message(new_user1.username, 'test for user1')
notice2 = UserNotification.objects.add_user_message(new_user1.username, 'test for user1')
notice3 = UserNotification.objects.add_user_message(new_user2.username, 'test for user2')
notice4 = UserNotification.objects.add_user_message(new_user2.username, 'test for user2')
notice5 = UserNotification.objects.add_user_message(new_user2.username, 'test for user2')
resp_user1 = self.client.get(self.endpoint + '?username=' + new_user1.username)
self.assertEqual(200, resp_user1.status_code)
json_resp1 = json.loads(resp_user1.content)
assert json_resp1['count'] == 2
resp_user2 = self.client.get(self.endpoint + '?username=' + new_user2.username)
self.assertEqual(200, resp_user2.status_code)
json_resp2 = json.loads(resp_user2.content)
assert json_resp2['count'] == 3
new_user1.delete()
new_user2.delete()
def test_get_notifications_with_invalid_usre_permission(self):
self.login_as(self.user)
notice1 = UserNotification.objects.add_user_message(self.user_name, 'test1')
resp = self.client.get(self.endpoint)
self.assertEqual(403, resp.status_code)
def test_get_notifications_with_no_notification(self):
self.login_as(self.admin)
resp = self.client.get(self.endpoint)
obj_resp = json.loads(resp.content)
assert len(obj_resp['notification_list']) == 0
self.assertEqual(200, resp.status_code)