adding users to tgroups

luncho/blueprints/groups.py

@@ -68,6 +68,34 @@ class UserIsNotAdminException(LunchoException):
         self.message = 'User is not admin'
 
 
+class SomeUsersNotFoundException(LunchoException):
+    """Some users in the add list do not exist.
+
+    .. sourcecode:: http
+
+       HTTP/1.1 404 Not Found
+       Content-Type: text/json
+
+       { "status": "ERROR",
+         "message", "Some users in the add list do not exist",
+         "users": ["<username>", "<username>", ...]}
+    """
+    def __init__(self, users=None):
+        super(SomeUsersNotFoundException, self).__init__()
+        self.status = 404
+        self.message = 'Some users in the add list do not exist'
+        self.users = users
+
+    def response(self):
+        json = {'status': 'ERROR',
+                'message': self.message}
+        if self.users:
+            json['users'] = self.users
+        response = jsonify(json)
+        response.status_code = self.status
+        return response
+
+
 groups = Blueprint('groups', __name__)
 
 LOG = logging.getLogger('luncho.blueprints.groups')
@@ -256,3 +284,64 @@ def delete_group(groupId):
     db.session.commit()
 
     return jsonify(status='OK')
+
+
+@groups.route('<groupId>/users/', methods=['PUT'])
+@ForceJSON(required=['usernames'])
+@auth
+def add_users_to_group(groupId):
+    """*Authenticated request* Add users to the group. Only the group
+    administrator can add users to their groups.
+
+    **Example request**:
+
+    .. sourcecode:: http
+
+        { "usernames": ["<username>", "<username>", ...] }
+
+    **Success (200)**:
+
+    .. sourcecode:: http
+
+       HTTP/1.1 200 OK
+       Content-Type: text/json
+
+       { "status": "OK" }
+
+    **Request not in JSON format (400)**:
+        :py:class:`RequestMustBeJSONException`
+
+    **User is not administrator of the group (403)**:
+        :py:class:`UserIsNotAdminException`
+
+    **User not found (via token) (404)**:
+        :py:class:`UserNotFoundException`
+
+    **Incomplete request, some users not found (404)**:
+        :py:class:`SomeUsersNotFoundException`
+
+    **Authorization required (412)**:
+        :py:class:`AuthorizationRequiredException`
+    """
+    user = request.user
+    group = Group.query.get(groupId)
+    if not group:
+        raise ElementNotFoundException('Group')
+
+    if not group.owner == user.username:
+        raise UserIsNotAdminException()
+
+    json = request.get_json(force=True)
+    unknown = []
+    for user in json['usernames']:
+        user_obj = User.query.get(user)
+        if not user_obj:
+            unknown.append(user)
+            continue
+
+        user_obj.groups.append(group)
+
+    if unknown:
+        raise SomeUsersNotFoundException(unknown)
+
+    return jsonify(status='OK')

tests/group_tests.py

@@ -215,5 +215,67 @@ class TestExistingGroups(LunchoTests):
                          token='invalid')
         self.assertJsonError(rv, 404, 'User not found (via token)')
 
+
+class TestUsersInGroup(LunchoTests):
+    """Tests for managing users in the group."""
+    def setUp(self):
+        super(TestUsersInGroup, self).setUp()
+        # create a user to have a token
+        self.user = User(username='test',
+                         fullname='Test User',
+                         passhash='hash')
+        self.user.verified = True
+        server.db.session.add(self.user)
+
+        # create a group for the user
+        self.group = Group(name='Test group',
+                           owner=self.user.username)
+        server.db.session.add(self.group)
+        server.db.session.commit()
+        self.user.get_token()
+
+    def tearDown(self):
+        super(TestUsersInGroup, self).tearDown()
+
+    def test_add_user(self):
+        """Try to add another user in the group."""
+        new_user = User(username='another_user',
+                        fullname='Another user',
+                        passhash='hash')
+        server.db.session.add(new_user)
+        server.db.session.commit()
+
+        request = {'usernames': [new_user.username]}
+
+        rv = self.put('/group/{groupId}/users/'.format(groupId=self.group.id),
+                      request,
+                      token=self.user.token)
+        self.assertJsonOk(rv)
+
+    def test_add_no_owner(self):
+        """Try to add users without being the group admin."""
+        new_user = User(username='another_user',
+                        fullname='Another user',
+                        passhash='hash')
+        server.db.session.add(new_user)
+        server.db.session.commit()
+        new_user.get_token()
+
+        request = {'usernames': [new_user.username]}
+
+        rv = self.put('/group/{groupId}/users/'.format(groupId=self.group.id),
+                      request,
+                      token=new_user.token)
+        self.assertJsonError(rv, 403, 'User is not admin')
+
+    def test_add_no_such_user(self):
+        """Try to add an unknown user to the group."""
+        request = {'usernames': ['unknown']}
+        rv = self.put('/group/{groupId}/users/'.format(groupId=self.group.id),
+                      request,
+                      token=self.user.token)
+        self.assertJsonError(rv, 404,
+                             'Some users in the add list do not exist')
+
 if __name__ == '__main__':
     unittest.main()