Create an action function to the heavy lifting

ckan/logic/action/update.py

@@ -664,6 +664,40 @@ def user_update(context, data_dict):
         model.repo.commit()
     return model_dictize.user_dictize(user, context)
 
+
+def user_cycle_apikey(context, data_dict):
+    '''Cycle a user's API key
+
+    :param id: the name or id of the user whose key needs to be updated
+    :type id: string
+
+    :returns: the updated user
+    :rtype: dictionary
+    '''
+    model = context['model']
+    user = context['user']
+    session = context['session']
+    schema = context.get('schema') or schema_.default_cycle_apikey_user_schema()
+    context['schema'] = schema
+    # check if user id in data_dict
+    id = _get_or_bust(data_dict, 'id')
+
+    # check if user exists
+    user_obj = model.User.get(id)
+    context['user_obj'] = user_obj
+    if user_obj is None:
+        raise NotFound('User was not found.')
+
+    # check permission
+    _check_access('user_cycle_apikey', context, data_dict)
+
+    # change key
+    old_data = _get_action('user_show')(context, data_dict)
+    old_data['apikey'] = model.types.make_uuid()
+    data_dict = old_data
+    _get_action('user_update')(context, data_dict)
+
+
 def task_status_update(context, data_dict):
     '''Update a task status.
 

ckan/logic/auth/update.py

@@ -212,6 +212,16 @@ def user_update(context, data_dict):
                         (user, user_obj.id)}
 
 
+def user_cycle_apikey(context, data_dict):
+    user = context['user']
+    user_obj = logic_auth.get_user_object(context, data_dict)
+    if user == user_obj.name:
+        # Allow users to update only their own user accounts.
+        return {'success': True}
+    return {'success': False, 'msg': _('User {0} not authorized to update user'
+            ' {1}'.format(user, user_obj.id))}
+
+
 def revision_change_state(context, data_dict):
     # FIXME currently only sysadmins can change state
     user = context['user']

ckan/new_tests/logic/action/test_update.py

@@ -79,6 +79,17 @@ def test_user_update_name(self):
 
     ## END-BEFORE
 
+    def test_user_cycle_apikey(self):
+        '''Test that updating a user's name works successfully.'''
+
+        user = factories.User()
+        # Required because you can only cycle your own API key
+        context = {'user': user['name']}
+        helpers.call_action('user_cycle_apikey', context=context, id=user['name'])
+        updated_user = helpers.call_action('user_show', context=context, id=user['id'])
+
+        assert updated_user['apikey'] != user['apikey']
+
     def test_user_update_with_id_that_does_not_exist(self):
         user_dict = factories.User.attributes()
         user_dict['id'] = "there's no user with this id"