Implement group management for cognito-idp

moto/cognitoidp/exceptions.py

@@ -24,6 +24,16 @@ class UserNotFoundError(BadRequest):
         })
 
 
+class GroupExistsException(BadRequest):
+
+    def __init__(self, message):
+        super(GroupExistsException, self).__init__()
+        self.description = json.dumps({
+            "message": message,
+            '__type': 'GroupExistsException',
+        })
+
+
 class NotAuthorizedError(BadRequest):
 
     def __init__(self, message):

moto/cognitoidp/models.py

@@ -11,8 +11,7 @@ from jose import jws
 
 from moto.compat import OrderedDict
 from moto.core import BaseBackend, BaseModel
-from .exceptions import NotAuthorizedError, ResourceNotFoundError, UserNotFoundError
-
+from .exceptions import GroupExistsException, NotAuthorizedError, ResourceNotFoundError, UserNotFoundError
 
 UserStatus = {
     "FORCE_CHANGE_PASSWORD": "FORCE_CHANGE_PASSWORD",
@@ -33,6 +32,7 @@ class CognitoIdpUserPool(BaseModel):
 
         self.clients = OrderedDict()
         self.identity_providers = OrderedDict()
+        self.groups = OrderedDict()
         self.users = OrderedDict()
         self.refresh_tokens = {}
         self.access_tokens = {}
@@ -185,6 +185,29 @@ class CognitoIdpIdentityProvider(BaseModel):
         return identity_provider_json
 
 
+class CognitoIdpGroup(BaseModel):
+
+    def __init__(self, user_pool_id, group_name, description, role_arn, precedence):
+        self.user_pool_id = user_pool_id
+        self.group_name = group_name
+        self.description = description or ""
+        self.role_arn = role_arn
+        self.precedence = precedence
+        self.last_modified_date = datetime.datetime.now()
+        self.creation_date = self.last_modified_date
+
+    def to_json(self):
+        return {
+            "GroupName": self.group_name,
+            "UserPoolId": self.user_pool_id,
+            "Description": self.description,
+            "RoleArn": self.role_arn,
+            "Precedence": self.precedence,
+            "LastModifiedDate": time.mktime(self.last_modified_date.timetuple()),
+            "CreationDate": time.mktime(self.creation_date.timetuple()),
+        }
+
+
 class CognitoIdpUser(BaseModel):
 
     def __init__(self, user_pool_id, username, password, status, attributes):
@@ -367,6 +390,46 @@ class CognitoIdpBackend(BaseBackend):
 
         del user_pool.identity_providers[name]
 
+    # Group
+    def create_group(self, user_pool_id, group_name, description, role_arn, precedence):
+        user_pool = self.user_pools.get(user_pool_id)
+        if not user_pool:
+            raise ResourceNotFoundError(user_pool_id)
+
+        group = CognitoIdpGroup(user_pool_id, group_name, description, role_arn, precedence)
+        if group.group_name in user_pool.groups:
+            raise GroupExistsException("A group with the name already exists")
+        user_pool.groups[group.group_name] = group
+
+        return group
+
+    def get_group(self, user_pool_id, group_name):
+        user_pool = self.user_pools.get(user_pool_id)
+        if not user_pool:
+            raise ResourceNotFoundError(user_pool_id)
+
+        if group_name not in user_pool.groups:
+            raise ResourceNotFoundError(group_name)
+
+        return user_pool.groups[group_name]
+
+    def list_groups(self, user_pool_id):
+        user_pool = self.user_pools.get(user_pool_id)
+        if not user_pool:
+            raise ResourceNotFoundError(user_pool_id)
+
+        return user_pool.groups.values()
+
+    def delete_group(self, user_pool_id, group_name):
+        user_pool = self.user_pools.get(user_pool_id)
+        if not user_pool:
+            raise ResourceNotFoundError(user_pool_id)
+
+        if group_name not in user_pool.groups:
+            raise ResourceNotFoundError(group_name)
+
+        del user_pool.groups[group_name]
+
     # User
     def admin_create_user(self, user_pool_id, username, temporary_password, attributes):
         user_pool = self.user_pools.get(user_pool_id)

moto/cognitoidp/responses.py

@@ -129,6 +129,47 @@ class CognitoIdpResponse(BaseResponse):
         cognitoidp_backends[self.region].delete_identity_provider(user_pool_id, name)
         return ""
 
+    # Group
+    def create_group(self):
+        group_name = self._get_param("GroupName")
+        user_pool_id = self._get_param("UserPoolId")
+        description = self._get_param("Description")
+        role_arn = self._get_param("RoleArn")
+        precedence = self._get_param("Precedence")
+
+        group = cognitoidp_backends[self.region].create_group(
+            user_pool_id,
+            group_name,
+            description,
+            role_arn,
+            precedence,
+        )
+
+        return json.dumps({
+            "Group": group.to_json(),
+        })
+
+    def get_group(self):
+        group_name = self._get_param("GroupName")
+        user_pool_id = self._get_param("UserPoolId")
+        group = cognitoidp_backends[self.region].get_group(user_pool_id, group_name)
+        return json.dumps({
+            "Group": group.to_json(),
+        })
+
+    def list_groups(self):
+        user_pool_id = self._get_param("UserPoolId")
+        groups = cognitoidp_backends[self.region].list_groups(user_pool_id)
+        return json.dumps({
+            "Groups": [group.to_json() for group in groups],
+        })
+
+    def delete_group(self):
+        group_name = self._get_param("GroupName")
+        user_pool_id = self._get_param("UserPoolId")
+        cognitoidp_backends[self.region].delete_group(user_pool_id, group_name)
+        return ""
+
     # User
     def admin_create_user(self):
         user_pool_id = self._get_param("UserPoolId")