diff --git a/ckan/config/routing.py b/ckan/config/routing.py
index 0fbb02bb886..8f78a6c099d 100644
--- a/ckan/config/routing.py
+++ b/ckan/config/routing.py
@@ -359,6 +359,7 @@ def make_map():
action='followers', ckan_icon='group')
m.connect('user_edit', '/user/edit/{id:.*}', action='edit',
ckan_icon='cog')
+ m.connect('user_delete', '/user/delete/{id}', action='delete')
m.connect('/user/reset/{id:.*}', action='perform_reset')
m.connect('register', '/user/register', action='register')
m.connect('login', '/user/login', action='login')
diff --git a/ckan/controllers/user.py b/ckan/controllers/user.py
index ab162ef0659..e0b1ba6c762 100644
--- a/ckan/controllers/user.py
+++ b/ckan/controllers/user.py
@@ -178,6 +178,21 @@ def new(self, data=None, errors=None, error_summary=None):
c.form = render(self.new_user_form, extra_vars=vars)
return render('user/new.html')
+ def delete(self, id):
+ '''Delete user with id passed as parameter'''
+ context = {'model': model,
+ 'session': model.Session,
+ 'user': c.user}
+ data_dict = {'id': id}
+
+ try:
+ get_action('user_delete')(context, data_dict)
+ user_index = h.url_for(controller='user', action='index')
+ h.redirect_to(user_index)
+ except NotAuthorized:
+ msg = _('Unauthorized to delete user with id "{user_id}".')
+ abort(401, msg.format(user_id=id))
+
def _save_new(self, context):
try:
data_dict = logic.clean_dict(unflatten(
diff --git a/ckan/lib/authenticator.py b/ckan/lib/authenticator.py
index 6f061caad91..f64db4f5b8d 100644
--- a/ckan/lib/authenticator.py
+++ b/ckan/lib/authenticator.py
@@ -12,9 +12,9 @@ class OpenIDAuthenticator(object):
def authenticate(self, environ, identity):
if 'repoze.who.plugins.openid.userid' in identity:
- openid = identity.get('repoze.who.plugins.openid.userid')
+ openid = identity['repoze.who.plugins.openid.userid']
user = User.by_openid(openid)
- if user is None:
+ if user is None or user.is_deleted():
return None
else:
return user.name
@@ -25,14 +25,20 @@ class UsernamePasswordAuthenticator(object):
implements(IAuthenticator)
def authenticate(self, environ, identity):
- if not 'login' in identity or not 'password' in identity:
+ if not ('login' in identity and 'password' in identity):
return None
- user = User.by_name(identity.get('login'))
+
+ login = identity['login']
+ user = User.by_name(login)
+
if user is None:
- log.debug('Login failed - username %r not found', identity.get('login'))
- return None
- if user.validate_password(identity.get('password')):
+ log.debug('Login failed - username %r not found', login)
+ elif user.is_deleted():
+ log.debug('Login as %r failed - user is deleted', login)
+ elif not user.validate_password(identity['password']):
+ log.debug('Login as %r failed - password not valid', login)
+ else:
return user.name
- log.debug('Login as %r failed - password not valid', identity.get('login'))
+
return None
diff --git a/ckan/lib/base.py b/ckan/lib/base.py
index 1298e0665c7..bfffcaf2a5d 100644
--- a/ckan/lib/base.py
+++ b/ckan/lib/base.py
@@ -290,8 +290,9 @@ def _identify_user_default(self):
if c.user:
c.user = c.user.decode('utf8')
c.userobj = model.User.by_name(c.user)
- if c.userobj is None:
- # This occurs when you are logged in, clean db
+ if c.userobj is None or c.userobj.is_deleted():
+ # This occurs when a user that was still logged in is deleted,
+ # or when you are logged in, clean db
# and then restart (or when you change your username)
# There is no user object, so even though repoze thinks you
# are logged in and your cookie has ckan_display_name, we
diff --git a/ckan/lib/create_test_data.py b/ckan/lib/create_test_data.py
index 3edf865c7c7..6c7a275c36f 100644
--- a/ckan/lib/create_test_data.py
+++ b/ckan/lib/create_test_data.py
@@ -519,8 +519,9 @@ def _create_user_without_commit(cls, name='', **user_dict):
@classmethod
def create_user(cls, name='', **kwargs):
- cls._create_user_without_commit(name, **kwargs)
+ user = cls._create_user_without_commit(name, **kwargs)
model.Session.commit()
+ return user
@classmethod
def flag_for_deletion(cls, pkg_names=[], tag_names=[], group_names=[],
diff --git a/ckan/logic/action/delete.py b/ckan/logic/action/delete.py
index 933e7624b43..9191151aa01 100644
--- a/ckan/logic/action/delete.py
+++ b/ckan/logic/action/delete.py
@@ -18,6 +18,28 @@
_get_or_bust = ckan.logic.get_or_bust
_get_action = ckan.logic.get_action
+def user_delete(context, data_dict):
+ '''Delete a user.
+
+ Only sysadmins can delete users.
+
+ :param id: the id or usernamename of the user to delete
+ :type id: string
+ '''
+
+ _check_access('user_delete', context, data_dict)
+
+ model = context['model']
+ user_id = _get_or_bust(data_dict, 'id')
+ user = model.User.get(user_id)
+
+ if user is None:
+ raise NotFound('User "{id}" was not found.'.format(id=user_id))
+
+ user.delete()
+ model.repo.commit()
+
+
def package_delete(context, data_dict):
'''Delete a dataset (package).
diff --git a/ckan/logic/action/get.py b/ckan/logic/action/get.py
index 26e7451a1a7..af5099ffab2 100644
--- a/ckan/logic/action/get.py
+++ b/ckan/logic/action/get.py
@@ -7,6 +7,7 @@
from pylons import config
import sqlalchemy
+import vdm.sqlalchemy
import ckan.lib.dictization
import ckan.logic as logic
@@ -672,6 +673,9 @@ def user_list(context, data_dict):
else_=model.User.fullname)
)
+ # Filter deleted users
+ query = query.filter(model.User.state != vdm.sqlalchemy.State.DELETED)
+
## hack for pagination
if context.get('return_query'):
return query
@@ -1181,7 +1185,9 @@ def user_autocomplete(context, data_dict):
q = data_dict['q']
limit = data_dict.get('limit', 20)
- query = model.User.search(q).limit(limit)
+ query = model.User.search(q)
+ query = query.filter(model.User.state != vdm.sqlalchemy.State.DELETED)
+ query = query.limit(limit)
user_list = []
for user in query.all():
diff --git a/ckan/logic/auth/delete.py b/ckan/logic/auth/delete.py
index 61b44697407..a2b121c11ad 100644
--- a/ckan/logic/auth/delete.py
+++ b/ckan/logic/auth/delete.py
@@ -4,6 +4,10 @@
from ckan.logic.auth import get_resource_object
from ckan.lib.base import _
+def user_delete(context, data_dict):
+ # Only sysadmins are authorized to purge organizations.
+ return {'success': False}
+
def package_delete(context, data_dict):
user = context['user']
package = get_package_object(context, data_dict)
diff --git a/ckan/migration/versions/070_add_state_column_to_user_table.py b/ckan/migration/versions/070_add_state_column_to_user_table.py
new file mode 100644
index 00000000000..1dfadbe7d32
--- /dev/null
+++ b/ckan/migration/versions/070_add_state_column_to_user_table.py
@@ -0,0 +1,17 @@
+import vdm.sqlalchemy
+
+
+def upgrade(migrate_engine):
+ migrate_engine.execute(
+ '''
+ ALTER TABLE "user" ADD COLUMN "state" text NOT NULL DEFAULT '%s'
+ ''' % vdm.sqlalchemy.State.ACTIVE
+ )
+
+
+def downgrade(migrate_engine):
+ migrate_engine.exeecute(
+ '''
+ ALTER TABLE "user" DROP COLUMN "state"
+ '''
+ )
diff --git a/ckan/model/follower.py b/ckan/model/follower.py
index 6e6096ed5cc..d14ffe667ae 100644
--- a/ckan/model/follower.py
+++ b/ckan/model/follower.py
@@ -1,30 +1,28 @@
-import sqlalchemy
import meta
import datetime
-import domain_object
+import sqlalchemy
+import vdm.sqlalchemy
-class UserFollowingUser(domain_object.DomainObject):
- '''A many-many relationship between users.
+import ckan.model
+import domain_object
- A relationship between one user (the follower) and another (the object),
- that means that the follower is currently following the object.
- '''
+class ModelFollowingModel(domain_object.DomainObject):
def __init__(self, follower_id, object_id):
self.follower_id = follower_id
self.object_id = object_id
self.datetime = datetime.datetime.now()
@classmethod
- def get(self, follower_id, object_id):
- '''Return a UserFollowingUser object for the given follower_id and
+ def get(cls, follower_id, object_id):
+ '''Return a ModelFollowingModel object for the given follower_id and
object_id, or None if no such follower exists.
'''
- query = meta.Session.query(UserFollowingUser)
- query = query.filter(UserFollowingUser.follower_id==follower_id)
- query = query.filter(UserFollowingUser.object_id==object_id)
- return query.first()
+ query = cls._get(follower_id, object_id)
+ following = cls._filter_following_objects(query)
+ if len(following) == 1:
+ return following[0]
@classmethod
def is_following(cls, follower_id, object_id):
@@ -32,33 +30,76 @@ def is_following(cls, follower_id, object_id):
otherwise.
'''
- return UserFollowingUser.get(follower_id, object_id) is not None
-
+ return cls.get(follower_id, object_id) is not None
@classmethod
def followee_count(cls, follower_id):
- '''Return the number of users followed by a user.'''
- return meta.Session.query(UserFollowingUser).filter(
- UserFollowingUser.follower_id == follower_id).count()
+ '''Return the number of objects followed by the follower.'''
+ return cls._get_followees(follower_id).count()
@classmethod
def followee_list(cls, follower_id):
- '''Return a list of users followed by a user.'''
- return meta.Session.query(UserFollowingUser).filter(
- UserFollowingUser.follower_id == follower_id).all()
+ '''Return a list of objects followed by the follower.'''
+ query = cls._get_followees(follower_id).all()
+ followees = cls._filter_following_objects(query)
+ return followees
+
+ @classmethod
+ def follower_count(cls, object_id):
+ '''Return the number of followers of the object.'''
+ return cls._get_followers(object_id).count()
+
+ @classmethod
+ def follower_list(cls, object_id):
+ '''Return a list of followers of the object.'''
+ query = cls._get_followers(object_id).all()
+ followers = cls._filter_following_objects(query)
+ return followers
+
+ @classmethod
+ def _filter_following_objects(cls, query):
+ return [q[0] for q in query]
+
+ @classmethod
+ def _get_followees(cls, follower_id):
+ return cls._get(follower_id)
+
+ @classmethod
+ def _get_followers(cls, object_id):
+ return cls._get(None, object_id)
+
+ @classmethod
+ def _get(cls, follower_id=None, object_id=None):
+ follower_alias = sqlalchemy.orm.aliased(cls._follower_class())
+ object_alias = sqlalchemy.orm.aliased(cls._object_class())
+
+ follower_id = follower_id or cls.follower_id
+ object_id = object_id or cls.object_id
+
+ query = meta.Session.query(cls, follower_alias, object_alias)\
+ .filter(sqlalchemy.and_(follower_alias.id == follower_id,\
+ cls.follower_id == follower_alias.id,\
+ cls.object_id == object_alias.id,\
+ follower_alias.state != vdm.sqlalchemy.State.DELETED,\
+ object_alias.state != vdm.sqlalchemy.State.DELETED,\
+ object_alias.id == object_id))
+ return query
+
+class UserFollowingUser(ModelFollowingModel):
+ '''A many-many relationship between users.
+
+ A relationship between one user (the follower) and another (the object),
+ that means that the follower is currently following the object.
+ '''
@classmethod
- def follower_count(cls, user_id):
- '''Return the number of followers of a user.'''
- return meta.Session.query(UserFollowingUser).filter(
- UserFollowingUser.object_id == user_id).count()
+ def _follower_class(cls):
+ return ckan.model.User
@classmethod
- def follower_list(cls, user_id):
- '''Return a list of followers of a user.'''
- return meta.Session.query(UserFollowingUser).filter(
- UserFollowingUser.object_id == user_id).all()
+ def _object_class(cls):
+ return ckan.model.User
user_following_user_table = sqlalchemy.Table('user_following_user',
@@ -76,62 +117,20 @@ def follower_list(cls, user_id):
meta.mapper(UserFollowingUser, user_following_user_table)
-class UserFollowingDataset(domain_object.DomainObject):
+class UserFollowingDataset(ModelFollowingModel):
'''A many-many relationship between users and datasets (packages).
A relationship between a user (the follower) and a dataset (the object),
that means that the user is currently following the dataset.
'''
- def __init__(self, follower_id, object_id):
- self.follower_id = follower_id
- self.object_id = object_id
- self.datetime = datetime.datetime.now()
-
- @classmethod
- def get(self, follower_id, object_id):
- '''Return a UserFollowingDataset object for the given follower_id and
- object_id, or None if no such follower exists.
-
- '''
- query = meta.Session.query(UserFollowingDataset)
- query = query.filter(UserFollowingDataset.follower_id==follower_id)
- query = query.filter(UserFollowingDataset.object_id==object_id)
- return query.first()
-
- @classmethod
- def is_following(cls, follower_id, object_id):
- '''Return True if follower_id is currently following object_id, False
- otherwise.
-
- '''
- return UserFollowingDataset.get(follower_id, object_id) is not None
-
-
- @classmethod
- def followee_count(cls, follower_id):
- '''Return the number of datasets followed by a user.'''
- return meta.Session.query(UserFollowingDataset).filter(
- UserFollowingDataset.follower_id == follower_id).count()
-
- @classmethod
- def followee_list(cls, follower_id):
- '''Return a list of datasets followed by a user.'''
- return meta.Session.query(UserFollowingDataset).filter(
- UserFollowingDataset.follower_id == follower_id).all()
-
-
@classmethod
- def follower_count(cls, dataset_id):
- '''Return the number of followers of a dataset.'''
- return meta.Session.query(UserFollowingDataset).filter(
- UserFollowingDataset.object_id == dataset_id).count()
+ def _follower_class(cls):
+ return ckan.model.User
@classmethod
- def follower_list(cls, dataset_id):
- '''Return a list of followers of a dataset.'''
- return meta.Session.query(UserFollowingDataset).filter(
- UserFollowingDataset.object_id == dataset_id).all()
+ def _object_class(cls):
+ return ckan.model.Package
user_following_dataset_table = sqlalchemy.Table('user_following_dataset',
@@ -150,60 +149,20 @@ def follower_list(cls, dataset_id):
meta.mapper(UserFollowingDataset, user_following_dataset_table)
-class UserFollowingGroup(domain_object.DomainObject):
+class UserFollowingGroup(ModelFollowingModel):
'''A many-many relationship between users and groups.
A relationship between a user (the follower) and a group (the object),
that means that the user is currently following the group.
'''
- def __init__(self, follower_id, object_id):
- self.follower_id = follower_id
- self.object_id = object_id
- self.datetime = datetime.datetime.now()
-
- @classmethod
- def get(self, follower_id, object_id):
- '''Return a UserFollowingGroup object for the given follower_id and
- object_id, or None if no such relationship exists.
-
- '''
- query = meta.Session.query(UserFollowingGroup)
- query = query.filter(UserFollowingGroup.follower_id == follower_id)
- query = query.filter(UserFollowingGroup.object_id == object_id)
- return query.first()
-
- @classmethod
- def is_following(cls, follower_id, object_id):
- '''Return True if follower_id is currently following object_id, False
- otherwise.
-
- '''
- return UserFollowingGroup.get(follower_id, object_id) is not None
-
- @classmethod
- def followee_count(cls, follower_id):
- '''Return the number of groups followed by a user.'''
- return meta.Session.query(UserFollowingGroup).filter(
- UserFollowingGroup.follower_id == follower_id).count()
-
@classmethod
- def followee_list(cls, follower_id):
- '''Return a list of groups followed by a user.'''
- return meta.Session.query(UserFollowingGroup).filter(
- UserFollowingGroup.follower_id == follower_id).all()
+ def _follower_class(cls):
+ return ckan.model.User
@classmethod
- def follower_count(cls, object_id):
- '''Return the number of users following a group.'''
- return meta.Session.query(UserFollowingGroup).filter(
- UserFollowingGroup.object_id == object_id).count()
-
- @classmethod
- def follower_list(cls, object_id):
- '''Return a list of the users following a group.'''
- return meta.Session.query(UserFollowingGroup).filter(
- UserFollowingGroup.object_id == object_id).all()
+ def _object_class(cls):
+ return ckan.model.Group
user_following_group_table = sqlalchemy.Table('user_following_group',
meta.metadata,
diff --git a/ckan/model/user.py b/ckan/model/user.py
index d3eb3d6f1c1..d36dd5343cc 100644
--- a/ckan/model/user.py
+++ b/ckan/model/user.py
@@ -6,6 +6,7 @@
from sqlalchemy.sql.expression import or_
from sqlalchemy.orm import synonym
from sqlalchemy import types, Column, Table
+import vdm.sqlalchemy
import meta
import types as _types
@@ -28,8 +29,11 @@
Column('sysadmin', types.Boolean, default=False),
)
+vdm.sqlalchemy.make_table_stateful(user_table)
-class User(domain_object.DomainObject):
+
+class User(vdm.sqlalchemy.StatefulObjectMixin,
+ domain_object.DomainObject):
VALID_NAME = re.compile(r"^[a-zA-Z0-9_\-]{3,255}$")
DOUBLE_SLASH = re.compile(':\/([^/])')
@@ -162,6 +166,9 @@ def number_administered_packages(self):
q = q.filter_by(user=self, role=model.Role.ADMIN)
return q.count()
+ def is_deleted(self):
+ return self.state == vdm.sqlalchemy.State.DELETED
+
def is_in_group(self, group):
return group in self.get_group_ids()
diff --git a/ckan/new_authz.py b/ckan/new_authz.py
index 01fb07927ac..bf1d0830d47 100644
--- a/ckan/new_authz.py
+++ b/ckan/new_authz.py
@@ -101,23 +101,30 @@ def clean_action_name(action_name):
def is_sysadmin(username):
- ''' returns True is username is a sysadmin '''
+ ''' Returns True is username is a sysadmin '''
+ user = _get_user(username)
+ return user and user.sysadmin
+
+
+def is_deleted(username):
+ ''' Returns True if username is deleted '''
+ user = _get_user(username)
+ return user and user.is_deleted()
+
+
+def _get_user(username):
+ ''' Try to get the user from c, if possible, and fallback to using the DB '''
if not username:
- return False
- # see if we can authorise without touching the database
+ return None
+ # See if we can get the user without touching the DB
try:
if c.userobj and c.userobj.name == username:
- if c.userobj.sysadmin:
- return True
- return False
+ return c.userobj
except TypeError:
# c is not available
pass
- # get user from the database
- user = model.User.get(username)
- if user and user.sysadmin:
- return True
- return False
+ # Get user from the DB
+ return model.User.get(username)
def get_group_or_org_admin_ids(group_id):
@@ -146,10 +153,14 @@ def is_authorized(action, context, data_dict=None):
action = clean_action_name(action)
auth_function = _AuthFunctions.get(action)
if auth_function:
+ username = context.get('user')
+ # deleted users are always unauthorized
+ if is_deleted(username):
+ return {'success': False}
# sysadmins can do anything unless the auth_sysadmins_check
# decorator was used in which case they are treated like all other
# users.
- if is_sysadmin(context.get('user')):
+ elif is_sysadmin(username):
if not getattr(auth_function, 'auth_sysadmins_check', False):
return {'success': True}
diff --git a/ckan/templates/user/edit_user_form.html b/ckan/templates/user/edit_user_form.html
index 19b8755b2bd..ce31e34b583 100644
--- a/ckan/templates/user/edit_user_form.html
+++ b/ckan/templates/user/edit_user_form.html
@@ -33,6 +33,12 @@
diff --git a/ckan/templates/user/read_base.html b/ckan/templates/user/read_base.html
index aa27a2b4119..f2b94e20782 100644
--- a/ckan/templates/user/read_base.html
+++ b/ckan/templates/user/read_base.html
@@ -76,6 +76,10 @@ {{ user.display_name }}
{{ _('Member Since') }}
{{ h.render_datetime(user.created) }}
+
+ - {{ _('State') }}
+ - {{ user.state }}
+
{% if c.is_myself %}
- {{ _('API Key') }} {{ _('Private') }}
diff --git a/ckan/tests/functional/test_user.py b/ckan/tests/functional/test_user.py
index 0ef528fd9ad..16795561f46 100644
--- a/ckan/tests/functional/test_user.py
+++ b/ckan/tests/functional/test_user.py
@@ -64,6 +64,25 @@ def test_user_read(self):
'rel="nofollow"')
assert 'Edit Profile' not in main_res, main_res
+ def test_user_delete_redirects_to_user_index(self):
+ user = CreateTestData.create_user('a_user')
+ url = url_for(controller='user', action='delete', id=user.id)
+ extra_environ = {'REMOTE_USER': 'testsysadmin'}
+
+ redirect_url = url_for(controller='user', action='index',
+ qualified=True)
+ res = self.app.get(url, status=302, extra_environ=extra_environ)
+
+ assert user.is_deleted(), user
+ assert res.header('Location').startswith(redirect_url), res.header('Location')
+
+ def test_user_delete_by_unauthorized_user(self):
+ user = model.User.by_name(u'annafan')
+ url = url_for(controller='user', action='delete', id=user.id)
+ extra_environ = {'REMOTE_USER': 'an_unauthorized_user'}
+
+ self.app.get(url, status=401, extra_environ=extra_environ)
+
def test_user_read_without_id(self):
offset = '/user/'
res = self.app.get(offset, status=302)
diff --git a/ckan/tests/lib/test_authenticator.py b/ckan/tests/lib/test_authenticator.py
new file mode 100644
index 00000000000..c7c026112d2
--- /dev/null
+++ b/ckan/tests/lib/test_authenticator.py
@@ -0,0 +1,106 @@
+import ckan
+
+import ckan.lib.create_test_data as ctd
+import ckan.lib.authenticator as authenticator
+
+CreateTestData = ctd.CreateTestData
+
+
+class TestUsernamePasswordAuthenticator(object):
+ @classmethod
+ def setup_class(cls):
+ auth = authenticator.UsernamePasswordAuthenticator()
+ cls.authenticate = auth.authenticate
+
+ @classmethod
+ def teardown(cls):
+ ckan.model.repo.rebuild_db()
+
+ def test_authenticate_succeeds_if_login_and_password_are_correct(self):
+ environ = {}
+ password = 'somepass'
+ user = CreateTestData.create_user('a_user', **{'password': password})
+ identity = {'login': user.name, 'password': password}
+
+ username = self.authenticate(environ, identity)
+ assert username == user.name, username
+
+ def test_authenticate_fails_if_user_is_deleted(self):
+ environ = {}
+ password = 'somepass'
+ user = CreateTestData.create_user('a_user', **{'password': password})
+ identity = {'login': user.name, 'password': password}
+ user.delete()
+
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_password_is_wrong(self):
+ environ = {}
+ user = CreateTestData.create_user('a_user')
+ identity = {'login': user.name, 'password': 'wrong-password'}
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_received_no_login_or_pass(self):
+ environ = {}
+ identity = {}
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_received_just_login(self):
+ environ = {}
+ identity = {'login': 'some-user'}
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_received_just_password(self):
+ environ = {}
+ identity = {'password': 'some-password'}
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_user_doesnt_exist(self):
+ environ = {}
+ identity = {'login': 'inexistent-user'}
+ assert self.authenticate(environ, identity) is None
+
+
+class TestOpenIDAuthenticator(object):
+ @classmethod
+ def setup_class(cls):
+ auth = authenticator.OpenIDAuthenticator()
+ cls.authenticate = auth.authenticate
+
+ @classmethod
+ def teardown(cls):
+ ckan.model.repo.rebuild_db()
+
+ def test_authenticate_succeeds_if_openid_is_correct(self):
+ environ = {}
+ openid = 'some-openid-key'
+ user = CreateTestData.create_user('a_user', **{'openid': openid})
+ identity = {'login': user.name,
+ 'repoze.who.plugins.openid.userid': openid}
+
+ username = self.authenticate(environ, identity)
+ assert username == user.name, username
+
+ def test_authenticate_fails_if_openid_is_incorrect(self):
+ environ = {}
+ openid = 'wrong-openid-key'
+ user = CreateTestData.create_user('a_user')
+ identity = {'login': user.name,
+ 'repoze.who.plugins.openid.userid': openid}
+
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_user_is_deleted(self):
+ environ = {}
+ openid = 'some-openid-key'
+ user = CreateTestData.create_user('a_user', **{'openid': openid})
+ user.delete()
+ identity = {'login': user.name,
+ 'repoze.who.plugins.openid.userid': openid}
+
+ assert self.authenticate(environ, identity) is None
+
+ def test_authenticate_fails_if_user_have_no_openid(self):
+ environ = {}
+ identity = {}
+ assert self.authenticate(environ, identity) is None
diff --git a/ckan/tests/lib/test_dictization.py b/ckan/tests/lib/test_dictization.py
index fb8bb49e628..72ca6fa31e5 100644
--- a/ckan/tests/lib/test_dictization.py
+++ b/ckan/tests/lib/test_dictization.py
@@ -939,6 +939,7 @@ def test_16_group_dictized(self):
'users': [{'about': u'I love reading Annakarenina. My site: http://anna.com',
'display_name': u'annafan',
'capacity' : 'public',
+ 'state': 'active',
'sysadmin': False,
'email_hash': 'd41d8cd98f00b204e9800998ecf8427e',
'fullname': None,
diff --git a/ckan/tests/logic/test_action.py b/ckan/tests/logic/test_action.py
index 73a3b838f0c..fecbdf96b53 100644
--- a/ckan/tests/logic/test_action.py
+++ b/ckan/tests/logic/test_action.py
@@ -7,6 +7,7 @@
from pylons import config
import datetime
+import vdm.sqlalchemy
import ckan
from ckan.lib.create_test_data import CreateTestData
from ckan.lib.dictization.model_dictize import resource_dictize
@@ -334,6 +335,11 @@ def test_42_create_resource_with_error(self):
def test_04_user_list(self):
+ # Create deleted user to make sure he won't appear in the user_list
+ deleted_user = CreateTestData.create_user('deleted_user')
+ deleted_user.delete()
+ model.repo.commit()
+
postparams = '%s=1' % json.dumps({})
res = self.app.post('/api/action/user_list', params=postparams)
res_obj = json.loads(res.body)
@@ -555,6 +561,33 @@ def test_12_user_update_errors(self):
for expected_message in test_call['messages']:
assert expected_message[1] in ''.join(res_obj['error'][expected_message[0]])
+ def test_user_delete(self):
+ name = 'normal_user'
+ CreateTestData.create_user(name)
+ user = model.User.get(name)
+ user_dict = {'id': user.id}
+ postparams = '%s=1' % json.dumps(user_dict)
+
+ res = self.app.post('/api/action/user_delete', params=postparams,
+ extra_environ={'Authorization': str(self.sysadmin_user.apikey)})
+
+ res_obj = json.loads(res.body)
+ deleted_user = model.User.get(name)
+ assert res_obj['success'] is True
+ assert deleted_user.is_deleted(), deleted_user
+
+ def test_user_delete_requires_data_dict_with_key_id(self):
+ user_dict = {'name': 'normal_user'}
+ postparams = '%s=1' % json.dumps(user_dict)
+
+ res = self.app.post('/api/action/user_delete', params=postparams,
+ extra_environ={'Authorization': str(self.sysadmin_user.apikey)},
+ status=StatusCodes.STATUS_409_CONFLICT)
+
+ res_obj = json.loads(res.body)
+ assert res_obj['success'] is False
+ assert res_obj['error']['id'] == ['Missing value']
+
def test_13_group_list(self):
postparams = '%s=1' % json.dumps({})
res = self.app.post('/api/action/group_list', params=postparams)
@@ -633,6 +666,11 @@ def test_14_group_show(self):
assert res_obj['success'] is False
def test_16_user_autocomplete(self):
+ # Create deleted user to make sure he won't appear in the user_list
+ deleted_user = CreateTestData.create_user('joe')
+ deleted_user.delete()
+ model.repo.commit()
+
#Empty query
postparams = '%s=1' % json.dumps({})
res = self.app.post(
diff --git a/ckan/tests/logic/test_auth.py b/ckan/tests/logic/test_auth.py
index 3ea9f2db39f..0f0a63372b2 100644
--- a/ckan/tests/logic/test_auth.py
+++ b/ckan/tests/logic/test_auth.py
@@ -48,8 +48,33 @@ def create_user(self, name):
self.apikeys[name] = str(json.loads(res.body)['result']['apikey'])
-class TestAuthOrgs(TestAuth):
+class TestAuthUsers(TestAuth):
+ def test_only_sysadmins_can_delete_users(self):
+ username = 'username'
+ user = {'id': username}
+ self.create_user(username)
+
+ self._call_api('user_delete', user, username, 403)
+ self._call_api('user_delete', user, 'sysadmin', 200)
+
+ def test_auth_deleted_users_are_always_unauthorized(self):
+ always_success = lambda x,y: {'success': True}
+ new_authz._AuthFunctions._build()
+ new_authz._AuthFunctions._functions['always_success'] = always_success
+ # We can't reuse the username with the other tests because we can't
+ # rebuild_db(), because in the setup_class we get the sysadmin. If we
+ # rebuild the DB, we would delete the sysadmin as well.
+ username = 'other_username'
+ self.create_user(username)
+ user = model.User.get(username)
+ user.delete()
+
+ assert not new_authz.is_authorized_boolean('always_success', {'user': username})
+ del new_authz._AuthFunctions._functions['always_success']
+
+
+class TestAuthOrgs(TestAuth):
def test_01_create_users(self):
# actual roles assigned later
self.create_user('org_admin')
diff --git a/ckan/tests/models/test_follower.py b/ckan/tests/models/test_follower.py
new file mode 100644
index 00000000000..9f4eb51ea3f
--- /dev/null
+++ b/ckan/tests/models/test_follower.py
@@ -0,0 +1,124 @@
+import ckan.model as model
+import ckan.lib.create_test_data as ctd
+
+CreateTestData = ctd.CreateTestData
+
+
+class FollowerClassesTests(object):
+ @classmethod
+ def teardown_class(cls):
+ model.repo.rebuild_db()
+
+ def test_get(self):
+ following = self.FOLLOWER_CLASS.get(self.follower.id, self.followee.id)
+ assert following.follower_id == self.follower.id, following
+ assert following.object_id == self.followee.id, following
+
+ def test_get_returns_none_if_couldnt_find_users(self):
+ following = self.FOLLOWER_CLASS.get('some-id', 'other-id')
+ assert following is None, following
+
+ def test_is_following(self):
+ assert self.FOLLOWER_CLASS.is_following(self.follower.id,
+ self.followee.id)
+
+ def test_is_following_returns_false_if_user_isnt_following(self):
+ assert not self.FOLLOWER_CLASS.is_following(self.followee.id,
+ self.follower.id)
+
+ def test_followee_count(self):
+ count = self.FOLLOWER_CLASS.followee_count(self.follower.id)
+ assert count == 1, count
+
+ def test_followee_list(self):
+ followees = self.FOLLOWER_CLASS.followee_list(self.follower.id)
+ object_ids = [f.object_id for f in followees]
+ assert object_ids == [self.followee.id], object_ids
+
+ def test_follower_count(self):
+ count = self.FOLLOWER_CLASS.follower_count(self.followee.id)
+ assert count == 1, count
+
+ def test_follower_list(self):
+ followers = self.FOLLOWER_CLASS.follower_list(self.followee.id)
+ follower_ids = [f.follower_id for f in followers]
+ assert follower_ids == [self.follower.id], follower_ids
+
+
+class TestUserFollowingUser(FollowerClassesTests):
+ FOLLOWER_CLASS = model.UserFollowingUser
+
+ @classmethod
+ def setup_class(cls):
+ model.repo.rebuild_db()
+ cls.follower = CreateTestData.create_user('follower')
+ cls.followee = CreateTestData.create_user('followee')
+ cls.FOLLOWER_CLASS(cls.follower.id, cls.followee.id).save()
+ cls._create_deleted_models()
+
+ @classmethod
+ def _create_deleted_models(cls):
+ deleted_user = CreateTestData.create_user('deleted_user')
+ cls.FOLLOWER_CLASS(deleted_user.id, cls.followee.id).save()
+ cls.FOLLOWER_CLASS(cls.follower.id, deleted_user.id).save()
+ deleted_user.delete()
+ deleted_user.save()
+
+
+class TestUserFollowingDataset(FollowerClassesTests):
+ FOLLOWER_CLASS = model.UserFollowingDataset
+
+ @classmethod
+ def setup_class(cls):
+ model.repo.rebuild_db()
+ cls.follower = CreateTestData.create_user('follower')
+ cls.followee = cls._create_dataset('followee')
+ cls.FOLLOWER_CLASS(cls.follower.id, cls.followee.id).save()
+ cls._create_deleted_models()
+
+ @classmethod
+ def _create_deleted_models(cls):
+ deleted_user = CreateTestData.create_user('deleted_user')
+ cls.FOLLOWER_CLASS(deleted_user.id, cls.followee.id).save()
+ deleted_user.delete()
+ deleted_user.save()
+ deleted_dataset = cls._create_dataset('deleted_dataset')
+ cls.FOLLOWER_CLASS(cls.follower.id, deleted_dataset.id).save()
+ deleted_dataset.delete()
+ deleted_dataset.save()
+
+ @classmethod
+ def _create_dataset(self, name):
+ CreateTestData.create_arbitrary({'name': name})
+ return model.Package.get(name)
+
+
+class TestUserFollowingGroup(FollowerClassesTests):
+ FOLLOWER_CLASS = model.UserFollowingGroup
+
+ @classmethod
+ def setup_class(cls):
+ model.repo.rebuild_db()
+ model.repo.new_revision()
+ cls.follower = CreateTestData.create_user('follower')
+ cls.followee = cls._create_group('followee')
+ cls.FOLLOWER_CLASS(cls.follower.id, cls.followee.id).save()
+ cls._create_deleted_models()
+ model.repo.commit_and_remove()
+
+ @classmethod
+ def _create_deleted_models(cls):
+ deleted_user = CreateTestData.create_user('deleted_user')
+ cls.FOLLOWER_CLASS(deleted_user.id, cls.followee.id).save()
+ deleted_user.delete()
+ deleted_user.save()
+ deleted_group = cls._create_group('deleted_group')
+ cls.FOLLOWER_CLASS(cls.follower.id, deleted_group.id).save()
+ deleted_group.delete()
+ deleted_group.save()
+
+ @classmethod
+ def _create_group(self, name):
+ group = model.Group(name)
+ group.save()
+ return group
diff --git a/ckan/tests/models/test_user.py b/ckan/tests/models/test_user.py
index b9ef7f0624f..c4b356ae854 100644
--- a/ckan/tests/models/test_user.py
+++ b/ckan/tests/models/test_user.py
@@ -56,6 +56,13 @@ def test_4_get_openid_missing_slash(self):
assert out
assert out.fullname == u'Sandra'
+ def test_is_deleted(self):
+ user = CreateTestData._create_user_without_commit('a_user')
+ assert not user.is_deleted(), user
+ user.delete()
+ assert user.is_deleted(), user
+
+
def to_names(domain_obj_list):
'''Takes a list of domain objects and returns a corresponding list
of their names.'''