diff --git a/seahub/api2/endpoints/admin/address_book/__init__.py b/seahub/api2/endpoints/admin/address_book/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/seahub/api2/endpoints/admin/address_book/group.py b/seahub/api2/endpoints/admin/address_book/group.py new file mode 100644 index 0000000000..f56d90a3ed --- /dev/null +++ b/seahub/api2/endpoints/admin/address_book/group.py @@ -0,0 +1,133 @@ +import logging + +from rest_framework.authentication import SessionAuthentication +from rest_framework.permissions import IsAdminUser +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, ccnet_api +from pysearpc import SearpcError + +from seahub.api2.utils import to_python_boolean +from seahub.avatar.settings import AVATAR_DEFAULT_SIZE +from seahub.base.accounts import User +from seahub.utils import is_valid_username +from seahub.utils.timeutils import timestamp_to_isoformat_timestr +from seahub.group.utils import get_group_member_info +from seahub.group.utils import is_group_member, is_group_admin, \ + validate_group_name, check_group_name_conflict +from seahub.admin_log.signals import admin_operation +from seahub.admin_log.models import GROUP_CREATE, GROUP_DELETE, GROUP_TRANSFER +from seahub.api2.utils import api_error +from seahub.api2.throttling import UserRateThrottle +from seahub.api2.authentication import TokenAuthentication + +logger = logging.getLogger(__name__) + +def address_book_group_to_dict(group): + if isinstance(group, int): + group = ccnet_api.get_group(group) + + return { + "id": group.id, + "name": group.group_name, + "owner": group.creator_name, + "created_at": timestamp_to_isoformat_timestr(group.timestamp), + "parent_group_id": group.parent_group_id, + } + +class AdminAddressBookGroup(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + throttle_classes = (UserRateThrottle,) + permission_classes = (IsAdminUser,) + + def get(self, request, group_id): + """List child groups and members in an address book group.""" + group_id = int(group_id) + + group = ccnet_api.get_group(group_id) + if not group: + error_msg = 'Group %d not found.' % group_id + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + + try: + avatar_size = int(request.GET.get('avatar_size', + AVATAR_DEFAULT_SIZE)) + except ValueError: + avatar_size = AVATAR_DEFAULT_SIZE + + try: + return_ancestors = to_python_boolean(request.GET.get( + 'return_ancestors', 'f')) + except ValueError: + return_ancestors = False + + ret_dict = address_book_group_to_dict(group) + ret_groups = [] + ret_members = [] + + groups = ccnet_api.get_child_groups(group_id) + for group in groups: + ret_groups.append(address_book_group_to_dict(group)) + + try: + members = ccnet_api.get_group_members(group_id) + except Exception as e: + logger.error(e) + error_msg = 'Internal Server Error' + return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg) + for m in members: + member_info = get_group_member_info(request, group_id, m.user_name, + avatar_size) + ret_members.append(member_info) + + ret_dict['groups'] = ret_groups + ret_dict['members'] = ret_members + + if return_ancestors: + # get ancestor groups and remove last group which is self + ancestor_groups = ccnet_api.get_ancestor_groups(group_id)[:-1] + ret_dict['ancestor_groups'] = [address_book_group_to_dict(grp) + for grp in ancestor_groups] + else: + ret_dict['ancestor_groups'] = [] + + return Response(ret_dict) + + def delete(self, request, group_id): + """Dismiss a specific group.""" + group_id = int(group_id) + + group = ccnet_api.get_group(group_id) + if not group: + return Response({'success': True}) + + group_owner = group.creator_name + group_name = group.group_name + + try: + ret_code = ccnet_api.remove_group(group_id) + if ret_code == -1: + error_msg = 'Failed to remove: this group has child groups.' + return api_error(status.HTTP_400_BAD_REQUEST, + error_msg) + + seafile_api.remove_group_repos(group_id) + except Exception as e: + logger.error(e) + error_msg = 'Internal Server Error' + return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg) + + # send admin operation log signal + admin_op_detail = { + "id": group_id, + "name": group_name, + "owner": group_owner, + } + admin_operation.send(sender=None, admin_name=request.user.username, + operation=GROUP_DELETE, detail=admin_op_detail) + + return Response({'success': True}) diff --git a/seahub/api2/endpoints/admin/address_book/groups.py b/seahub/api2/endpoints/admin/address_book/groups.py new file mode 100644 index 0000000000..33c8b5ced3 --- /dev/null +++ b/seahub/api2/endpoints/admin/address_book/groups.py @@ -0,0 +1,110 @@ +import logging + +from rest_framework.authentication import SessionAuthentication +from rest_framework.permissions import IsAdminUser +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, ccnet_api +from pysearpc import SearpcError + +from seahub.base.accounts import User +from seahub.utils import is_valid_username +from seahub.utils.timeutils import timestamp_to_isoformat_timestr +from seahub.group.utils import is_group_member, is_group_admin, \ + validate_group_name, check_group_name_conflict +from seahub.admin_log.signals import admin_operation +from seahub.admin_log.models import GROUP_CREATE, GROUP_DELETE, GROUP_TRANSFER +from seahub.api2.utils import api_error +from seahub.api2.throttling import UserRateThrottle +from seahub.api2.authentication import TokenAuthentication + +logger = logging.getLogger(__name__) + +def address_book_group_to_dict(group): + if isinstance(group, int): + group = ccnet_api.get_group(group) + + return { + "id": group.id, + "name": group.group_name, + "owner": group.creator_name, + "created_at": timestamp_to_isoformat_timestr(group.timestamp), + "parent_group_id": group.parent_group_id, + } + +class AdminAddressBookGroups(APIView): + authentication_classes = (TokenAuthentication, SessionAuthentication) + throttle_classes = (UserRateThrottle,) + permission_classes = (IsAdminUser,) + + def get(self, request): + """List top groups in address book.""" + return_results = [] + + groups = ccnet_api.get_top_groups() + for group in groups: + return_results.append(address_book_group_to_dict(group)) + + return Response({"data": return_results}) + + def post(self, request): + """Add a group in address book. + + parent_group: -1 - no parent group; + > 0 - have parent group. + group_owner: default to system admin + group_staff: default to system admin + """ + group_name = request.data.get('group_name', '').strip() + if not group_name: + error_msg = 'group_name %s invalid.' % group_name + return api_error(status.HTTP_400_BAD_REQUEST, error_msg) + + # Check whether group name is validate. + if not validate_group_name(group_name): + error_msg = _(u'Group name can only contain letters, numbers, blank, hyphen or underscore') + return api_error(status.HTTP_400_BAD_REQUEST, error_msg) + + # Check whether group name is duplicated. + if check_group_name_conflict(request, group_name): + error_msg = _(u'There is already a group with that name.') + return api_error(status.HTTP_400_BAD_REQUEST, error_msg) + + group_owner = request.data.get('group_owner', '') + if group_owner: + try: + User.objects.get(email=group_owner) + except User.DoesNotExist: + error_msg = 'User %s not found.' % group_owner + return api_error(status.HTTP_404_NOT_FOUND, error_msg) + else: + group_owner = request.user.username + + try: + parent_group = int(request.data.get('parent_group', -1)) + except ValueError: + error_msg = 'parent_group invalid' + return api_error(status.HTTP_400_BAD_REQUEST, error_msg) + + if parent_group < 0 and parent_group != -1: + error_msg = 'parent_group invalid' + return api_error(status.HTTP_400_BAD_REQUEST, error_msg) + + # TODO: check parent group exists + + try: + group_id = ccnet_api.create_group(group_name, group_owner, + parent_group_id=parent_group) + except SearpcError as e: + logger.error(e) + error_msg = 'Internal Server Error' + return api_error(status.HTTP_500_INTERNAL_SERVER_ERROR, error_msg) + + # get info of new group + group_info = address_book_group_to_dict(group_id) + + return Response(group_info, status=status.HTTP_200_OK) diff --git a/seahub/urls.py b/seahub/urls.py index 3c1aad739c..2fba156293 100644 --- a/seahub/urls.py +++ b/seahub/urls.py @@ -91,6 +91,8 @@ from seahub.api2.endpoints.admin.invitations import InvitationsView as AdminInvi 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 +from seahub.api2.endpoints.admin.address_book.groups import AdminAddressBookGroups +from seahub.api2.endpoints.admin.address_book.group import AdminAddressBookGroup # Uncomment the next two lines to enable the admin: #from django.contrib import admin @@ -384,6 +386,10 @@ urlpatterns = patterns( url(r'^invite/', include('seahub.invitations.urls', app_name='invitations', namespace='invitations')), url(r'^terms/', include('termsandconditions.urls')), + ## admin::address book + url(r'^api/v2.1/admin/address-book/groups/$', AdminAddressBookGroups.as_view(), name='api-v2.1-admin-address-book-groups'), + url(r'^api/v2.1/admin/address-book/groups/(?P\d+)/$', AdminAddressBookGroup.as_view(), name='api-v2.1-admin-address-book-group'), + ### system admin ### url(r'^sysadmin/$', sysadmin, name='sysadmin'), url(r'^sys/settings/$', sys_settings, name='sys_settings'), diff --git a/tests/api/endpoints/admin/address_book/test_group.py b/tests/api/endpoints/admin/address_book/test_group.py new file mode 100644 index 0000000000..80ae72d675 --- /dev/null +++ b/tests/api/endpoints/admin/address_book/test_group.py @@ -0,0 +1,66 @@ +import json + +from django.core.urlresolvers import reverse +from seaserv import ccnet_api + +from seahub.test_utils import BaseTestCase +from tests.common.utils import randstring + +class GroupsTest(BaseTestCase): + + def setUp(self): + self.user_name = self.user.username + self.admin_name = self.admin.username + + group_name = 'top group xxx' + self.top_group_id = ccnet_api.create_group(group_name, self.admin_name, + parent_group_id=-1) + self.login_as(self.admin) + self.url = reverse('api-v2.1-admin-address-book-group', + args=[self.top_group_id]) + + def tearDown(self): + self.remove_group(self.top_group_id) + + def test_can_list_child_groups(self): + child_group_id = ccnet_api.create_group('child group xxx', + self.user.username, + parent_group_id=self.top_group_id) + + resp = self.client.get(self.url) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp['groups']) >= 1 + assert len(json_resp['members']) >= 1 + assert len(json_resp['ancestor_groups']) == 0 + assert json_resp['id'] == self.top_group_id + self.remove_group(child_group_id) + + def test_can_ancestor_groups(self): + child_group_id = ccnet_api.create_group('child group xxx', + self.user.username, + parent_group_id=self.top_group_id) + + url = reverse('api-v2.1-admin-address-book-group', + args=[child_group_id]) + '?return_ancestors=true' + resp = self.client.get(url) + self.assertEqual(200, resp.status_code) + json_resp = json.loads(resp.content) + assert len(json_resp['groups']) == 0 + assert len(json_resp['ancestor_groups']) >= 1 + assert json_resp['ancestor_groups'][-1]['id'] == self.top_group_id + self.remove_group(child_group_id) + + def test_can_delete_group(self): + resp = self.client.delete(self.url) + self.assertEqual(200, resp.status_code) + + def test_cannot_delete_group_with_child(self): + child_group_id = ccnet_api.create_group('child group xxx', + self.user.username, + parent_group_id=self.top_group_id) + + resp = self.client.delete(self.url) + self.assertEqual(400, resp.status_code) + + self.remove_group(child_group_id) diff --git a/tests/api/endpoints/admin/address_book/test_groups.py b/tests/api/endpoints/admin/address_book/test_groups.py new file mode 100644 index 0000000000..4fc98cad17 --- /dev/null +++ b/tests/api/endpoints/admin/address_book/test_groups.py @@ -0,0 +1,61 @@ +import json + +from django.core.urlresolvers import reverse +from seaserv import ccnet_api + +from seahub.test_utils import BaseTestCase +from tests.common.utils import randstring + + +class GroupsTest(BaseTestCase): + + def setUp(self): + self.user_name = self.user.username + self.admin_name = self.admin.username + + self.login_as(self.admin) + self.url = reverse('api-v2.1-admin-address-book-groups') + + def test_can_list_top_groups(self): + top_group_id = ccnet_api.create_group('top group xxx', self.user.username, + parent_group_id=-1) + + resp = self.client.get(self.url) + self.assertEqual(200, resp.status_code) + + json_resp = json.loads(resp.content) + assert len(json_resp['data']) >= 1 + + self.remove_group(top_group_id) + + def test_can_create_top_group(self): + resp = self.client.post(self.url, { + 'group_name': randstring(10), + 'parent_group': -1, + 'group_owner': self.user.username + }) + self.assertEqual(200, resp.status_code) + + json_resp = json.loads(resp.content) + assert len(json_resp['name']) == 10 + assert json_resp['parent_group_id'] == -1 + + self.remove_group(json_resp['id']) + + def test_can_create_child_group(self): + top_group_id = ccnet_api.create_group('top group xxx', self.user.username, + parent_group_id=-1) + + resp = self.client.post(self.url, { + 'group_name': randstring(10), + 'parent_group': top_group_id, + 'group_owner': self.user.username + }) + self.assertEqual(200, resp.status_code) + + json_resp = json.loads(resp.content) + assert len(json_resp['name']) == 10 + assert json_resp['parent_group_id'] == top_group_id + + self.remove_group(json_resp['id']) + self.remove_group(top_group_id)