Add API function add role to user

database/schema.py

@@ -53,6 +53,22 @@ class CreateRole(graphene.Mutation):
         return CreateRole(role=role)
 
 
+class AddRoleToUser(graphene.Mutation):
+    class Arguments:
+        username = graphene.String(required=True)
+        role = graphene.String(required=True)
+
+    user = graphene.Field(User)
+    role = graphene.Field(Role)
+
+    def mutate(self, info, username, role):
+        user = user_datastore.db.session().query(UserModel).filter_by(username=username).first()
+        role = user_datastore.find_role(role)
+        user_datastore.add_role_to_user(user, role)
+        user_datastore.commit()
+        return AddRoleToUser(user=user,role=role)
+
+
 class Query(graphene.ObjectType):
     node = relay.Node.Field()
     all_users = SQLAlchemyConnectionField(User)
@@ -65,6 +81,7 @@ class Query(graphene.ObjectType):
 class Mutation(graphene.ObjectType):
     create_user = UserMutation.Field()
     create_role = CreateRole.Field()
+    add_role_to_user = AddRoleToUser.Field()
 
 
 schema = graphene.Schema(query=Query, types=[User, Role, Application], mutation=Mutation)

test/test_graphql.py

@@ -72,7 +72,7 @@ class GraphQlTests(LiveServerTestCase):
         '''.format(username, password, email).strip()
         return loads(self.client.execute(querystring))
 
-    def test_create_role(self):
+    def test_create_and_assign_role(self):
         role_name = "testusers"
         role_description = "A role assigned to users during unittests"
         result = self._create_role(role_name, role_description)
@@ -82,6 +82,52 @@ class GraphQlTests(LiveServerTestCase):
         role = result["data"]["createRole"]["role"]
         self.assertEqual(role["name"], role_name)
         self.assertEqual(role["description"], role_description)
+        username = "graphqluser2"
+        password = "randompassword"
+        email = "user2@example.com"
+        self._create_user(username, password, email)
+        querystring = '''
+        mutation{{
+          addRoleToUser(
+            username: "{0}",
+            role: "{1}"){{
+              user{{
+                username,
+                roles{{
+                  edges{{
+                    node{{
+                      name
+                    }}
+                  }}
+                }}
+              }}
+              role{{
+                name,
+                description
+                users{{
+                  edges{{
+                    node{{
+                      username
+                    }}
+                  }}
+                }}
+              }}
+           }}
+        }}
+        '''.format(username, role_name).strip()
+        result = loads(self.client.execute(querystring))
+        self.assertIn("data", result)
+        self.assertIn("addRoleToUser", result["data"])
+        user = result["data"]["addRoleToUser"]["user"]
+        self.assertEqual(username, user["username"])
+        self.assertIn("edges", user["roles"])
+        roles = user["roles"]["edges"]
+        self.assertEqual(role_name, roles[0]["node"]["name"])
+        role = result["data"]["addRoleToUser"]["role"]
+        self.assertEqual(role_name, role["name"])
+        self.assertIn("edges", role["users"])
+        users = role["users"]["edges"]
+        self.assertEqual(username, users[0]["node"]["username"])
 
     def _create_role(self, name, description):
         querystring = '''