[#2838] fix conflicts in tests

ckan/cli/db.py

@@ -4,9 +4,10 @@
 import logging
 import ckan.migration as migration_repo
 import click
+from itertools import groupby
 
 from ckan.cli import error_shout
-
+import ckan.model as model
 log = logging.getLogger(__name__)
 
 
@@ -86,6 +87,34 @@ def version(hash):
     )
 
 
+@db.command(u'duplicate_emails', short_help=u'Check users email for duplicate')
+def duplicate_emails():
+    u'''Check users email for duplicate'''
+    log.info(u"Searching for accounts with duplicate emails.")
+
+    q = model.Session.query(model.User.email,
+                            model.User.name) \
+        .filter(model.User.state == u"active") \
+        .filter(model.User.email != u"") \
+        .order_by(model.User.email)
+
+    if q.all():
+        try:
+            for k, grp in groupby(q.all(), lambda x: x[0]):
+                users = [user[1] for user in grp]
+                if len(users) > 1:
+                    s = u'{} appears {} time(s). Users: {}'
+                    click.secho(s.format(k, len(users), u', '.join(users)))
+        except Exception as e:
+            error_shout(e)
+        else:
+            click.secho(u'Duplicate email search: SUCCESS',
+                        fg=u'green', bold=True)
+    else:
+        click.secho(u'Duplicate email search: NOT FOUND',
+                    fg=u'green', bold=True)
+
+
 def _version_hash_to_ordinal(version):
     if u'base' == version:
         return 0

ckan/logic/schema.py

@@ -410,9 +410,11 @@ def default_user_schema(
 @validator_args
 def user_new_form_schema(
         unicode_safe, user_both_passwords_entered,
-        user_password_validator, user_passwords_match):
+        user_password_validator, user_passwords_match,
+        email_is_unique):
     schema = default_user_schema()
 
+    schema['email'] = [email_is_unique]
     schema['password1'] = [text_type, user_both_passwords_entered,
                            user_password_validator, user_passwords_match]
     schema['password2'] = [text_type]
@@ -423,9 +425,10 @@ def user_new_form_schema(
 @validator_args
 def user_edit_form_schema(
         ignore_missing, unicode_safe, user_both_passwords_entered,
-        user_password_validator, user_passwords_match):
+        user_password_validator, user_passwords_match, email_is_unique):
     schema = default_user_schema()
 
+    schema['email'] = [email_is_unique]
     schema['password'] = [ignore_missing]
     schema['password1'] = [ignore_missing, unicode_safe,
                            user_password_validator, user_passwords_match]

ckan/logic/validators.py

@@ -861,6 +861,18 @@ def email_validator(value, context):
             raise Invalid(_('Email {email} is not a valid format').format(email=value))
     return value
 
+def email_is_unique(key, data, errors, context):
+    '''Validate email is unique'''
+    model = context['model']
+    session = context['session']
+    user = session.query(model.User).filter_by(email=data[key]).first()
+
+    # check if email belongs to updated user
+    if user and user.name != data[('name',)]:
+        raise Invalid(
+            _('The email address \'{email}\' belongs to a registered user.').
+            format(email=data[key]))
+    return
 
 def one_of(list_of_value):
     ''' Checks if the provided value is present in a list '''

ckan/tests/controllers/test_user.py

@@ -301,6 +301,22 @@ def test_email_change_with_password(self, app):
         response = submit_and_follow(app, form, env, "save")
         assert "Profile updated" in response
 
+    def test_email_change_on_existed_email(self):
+        app = self._get_test_app()
+        env, response, user = _get_user_edit_page(app)
+        factories.User(email='existed@email.com')
+
+        form = response.forms['user-edit-form']
+
+        # new values
+        form['email'] = 'existed@email.com'
+
+        # factory returns user with password 'pass'
+        form.fields['old_password'][0].value = 'RandomPassword123'
+
+        response = webtest_submit(form, 'save', status=200, extra_environ=env)
+        assert_true('belongs to a registered user' in response)
+
     def test_edit_user_logged_in_username_change(self, app):
 
         user_pass = "TestPassword1"

ckan/tests/logic/test_validators.py

@@ -10,6 +10,8 @@
 import mock
 import pytest
 
+from collections import namedtuple
+
 import ckan.lib.navl.dictization_functions as df
 import ckan.logic.validators as validators
 import ckan.model as model
@@ -160,6 +162,97 @@ def call_and_assert(key, data, errors, context):
     return decorator
 
 
+def test_email_is_unique_validator_with_existed_value(self):
+    user = namedtuple('User', 'name email')
+    user.name = 'test_user_888'
+    user.email = 'existed@email.com'
+    model = mock.MagicMock()
+    session = model.Session
+    context = {'model': model, 'session': session}
+
+    # mock user object with email
+    session.query(
+        model.User).filter_by.return_value.first.return_value = user
+
+    data = factories.validator_data_dict()
+    key = ('email',)
+    data[key] = 'existed@email.com'
+    data[('name',)] = 'test_user_999'
+    errors = factories.validator_errors_dict()
+    errors[key] = []
+
+    @raises_Invalid
+    def call_validator(*args, **kwargs):
+        return validators.email_is_unique(*args, **kwargs)
+    call_validator(key, data, errors, context)
+
+
+def test_email_is_unique_validator_with_unique_value(self):
+    model = mock.MagicMock()
+    session = model.Session
+    context = {'model': model, 'session': session}
+
+    session.query(model.User).filter_by.return_value.first.return_value = None
+
+    data = factories.validator_data_dict()
+    key = ('email',)
+    data[key] = 'unique@email.com'
+    data[('name',)] = 'test_user_888'
+    errors = factories.validator_errors_dict()
+    errors[key] = []
+
+    @t.returns_None
+    def call_validator(*args, **kwargs):
+        return validators.email_is_unique(*args, **kwargs)
+    call_validator(key, data, errors, context)
+
+
+def test_email_is_unique_validator_user_update_email_unchanged(self):
+    user = namedtuple('User', 'name email')
+    user.name = 'test_user_888'
+    user.email = 'existed@email.com'
+    model = mock.MagicMock()
+    session = model.Session
+    context = {'model': model, 'session': session}
+
+    session.query(model.User).filter_by.return_value.first.return_value = user
+
+    data = factories.validator_data_dict()
+    key = ('email',)
+    data[key] = 'exited@esmail.com'
+    data[('name',)] = 'test_user_888'
+    errors = factories.validator_errors_dict()
+    errors[key] = []
+
+    @t.returns_None
+    def call_validator(*args, **kwargs):
+        return validators.email_is_unique(*args, **kwargs)
+    call_validator(key, data, errors, context)
+
+
+def test_email_is_unique_validator_user_update_email_new(self):
+    user = namedtuple('User', 'name email')
+    user.name = 'test_user_888'
+    user.email = 'existed@email.com'
+    model = mock.MagicMock()
+    session = model.Session
+    context = {'model': model, 'session': session}
+
+    session.query(model.User).filter_by.return_value.first.return_value = user
+
+    data = factories.validator_data_dict()
+    key = ('email',)
+    data[key] = 'new@email.com'
+    data[('name',)] = 'test_user_888'
+    errors = factories.validator_errors_dict()
+    errors[key] = []
+
+    @t.returns_None
+    def call_validator(*args, **kwargs):
+        return validators.email_is_unique(*args, **kwargs)
+    call_validator(key, data, errors, context)
+
+
 def test_name_validator_with_invalid_value():
     """If given an invalid value name_validator() should do raise Invalid.