[address book] Add admin api

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})

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)

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<group_id>\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'),

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)

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)