fixed ldap tests when ldap lib is installed

rhodecode/model/forms.py

@@ -1,387 +1,384 b''
 """ this is forms validation classes
 http://formencode.org/module-formencode.validators.html
 for list off all availible validators
 
 we can create our own validators
 
 The table below outlines the options which can be used in a schema in addition to the validators themselves
 pre_validators          []     These validators will be applied before the schema
 chained_validators      []     These validators will be applied after the schema
 allow_extra_fields      False     If True, then it is not an error when keys that aren't associated with a validator are present
 filter_extra_fields     False     If True, then keys that aren't associated with a validator are removed
 if_key_missing          NoDefault If this is given, then any keys that aren't available but are expected will be replaced with this value (and then validated). This does not override a present .if_missing attribute on validators. NoDefault is a special FormEncode class to mean that no default values has been specified and therefore missing keys shouldn't take a default value.
 ignore_key_missing      False     If True, then missing keys will be missing in the result, if the validator doesn't have .if_missing on it already
 
 
 <name> = formencode.validators.<name of validator>
 <name> must equal form name
 list=[1,2,3,4,5]
 for SELECT use formencode.All(OneOf(list), Int())
 
 """
 import logging
 
 import formencode
 from formencode import All
 
 from pylons.i18n.translation import _
 
 from rhodecode.model import validators as v
 from rhodecode import BACKENDS
 
 log = logging.getLogger(__name__)
 
 
 class LoginForm(formencode.Schema):
     allow_extra_fields = True
     filter_extra_fields = True
     username = v.UnicodeString(
         strip=True,
         min=1,
         not_empty=True,
         messages={
            'empty': _(u'Please enter a login'),
            'tooShort': _(u'Enter a value %(min)i characters long or more')}
     )
 
     password = v.UnicodeString(
         strip=False,
         min=3,
         not_empty=True,
         messages={
             'empty': _(u'Please enter a password'),
             'tooShort': _(u'Enter %(min)i characters or more')}
     )
 
     remember = v.StringBoolean(if_missing=False)
 
     chained_validators = [v.ValidAuth()]
 
 
 def UserForm(edit=False, old_data={}):
     class _UserForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                        v.ValidUsername(edit, old_data))
         if edit:
             new_password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False),
             )
             admin = v.StringBoolean(if_missing=False)
         else:
             password = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=True)
             )
             password_confirmation = All(
                 v.ValidPassword(),
                 v.UnicodeString(strip=False, min=6, not_empty=False)
             )
 
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
 
         chained_validators = [v.ValidPasswordsMatch()]
 
     return _UserForm
 
 
 def UserGroupForm(edit=False, old_data={}, available_members=[]):
     class _UserGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         users_group_name = All(
             v.UnicodeString(strip=True, min=1, not_empty=True),
             v.ValidUserGroup(edit, old_data)
         )
 
         users_group_active = v.StringBoolean(if_missing=False)
 
         if edit:
             users_group_members = v.OneOf(
                 available_members, hideList=False, testValueList=True,
                 if_missing=None, not_empty=False
             )
 
     return _UserGroupForm
 
 
 def ReposGroupForm(edit=False, old_data={}, available_groups=[],
                    can_create_in_root=False):
     class _ReposGroupForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
 
         group_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                                v.SlugifyName())
         group_description = v.UnicodeString(strip=True, min=1,
                                                 not_empty=False)
         if edit:
             #FIXME: do a special check that we cannot move a group to one of
             #it's children
             pass
         group_parent_id = All(v.CanCreateGroup(can_create_in_root),
                               v.OneOf(available_groups, hideList=False,
                                       testValueList=True,
                                       if_missing=None, not_empty=True))
         enable_locking = v.StringBoolean(if_missing=False)
         recursive = v.StringBoolean(if_missing=False)
         chained_validators = [v.ValidReposGroup(edit, old_data),
                               v.ValidPerms('group')]
 
     return _ReposGroupForm
 
 
 def RegisterForm(edit=False, old_data={}):
     class _RegisterForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         username = All(
             v.ValidUsername(edit, old_data),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         password = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         password_confirmation = All(
             v.ValidPassword(),
             v.UnicodeString(strip=False, min=6, not_empty=True)
         )
         active = v.StringBoolean(if_missing=False)
         firstname = v.UnicodeString(strip=True, min=1, not_empty=False)
         lastname = v.UnicodeString(strip=True, min=1, not_empty=False)
         email = All(v.Email(not_empty=True), v.UniqSystemEmail(old_data))
 
         chained_validators = [v.ValidPasswordsMatch()]
 
     return _RegisterForm
 
 
 def PasswordResetForm():
     class _PasswordResetForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         email = All(v.ValidSystemEmail(), v.Email(not_empty=True))
     return _PasswordResetForm
 
 
 def RepoForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
              repo_groups=[], landing_revs=[]):
     class _RepoForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(old_data),
                          v.OneOf(repo_groups, hideList=True))
         repo_type = v.OneOf(supported_backends)
         repo_description = v.UnicodeString(strip=True, min=1, not_empty=False)
         repo_private = v.StringBoolean(if_missing=False)
         repo_landing_rev = v.OneOf(landing_revs, hideList=True)
         clone_uri = All(v.UnicodeString(strip=True, min=1, not_empty=False))
 
         repo_enable_statistics = v.StringBoolean(if_missing=False)
         repo_enable_downloads = v.StringBoolean(if_missing=False)
         repo_enable_locking = v.StringBoolean(if_missing=False)
 
         if edit:
             #this is repo owner
             user = All(v.UnicodeString(not_empty=True), v.ValidRepoUser())
 
         chained_validators = [v.ValidCloneUri(),
                               v.ValidRepoName(edit, old_data)]
     return _RepoForm
 
 
 def RepoPermsForm():
     class _RepoPermsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         chained_validators = [v.ValidPerms()]
     return _RepoPermsForm
 
 
 def RepoFieldForm():
     class _RepoFieldForm(formencode.Schema):
         filter_extra_fields = True
         allow_extra_fields = True
 
         new_field_key = All(v.FieldKey(),
                             v.UnicodeString(strip=True, min=3, not_empty=True))
         new_field_value = v.UnicodeString(not_empty=False, if_missing='')
         new_field_type = v.OneOf(['str', 'unicode', 'list', 'tuple'],
                                  if_missing='str')
         new_field_label = v.UnicodeString(not_empty=False)
         new_field_desc = v.UnicodeString(not_empty=False)
 
     return _RepoFieldForm
 
 
 def RepoForkForm(edit=False, old_data={}, supported_backends=BACKENDS.keys(),
                  repo_groups=[], landing_revs=[]):
     class _RepoForkForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         repo_name = All(v.UnicodeString(strip=True, min=1, not_empty=True),
                         v.SlugifyName())
         repo_group = All(v.CanWriteGroup(),
                          v.OneOf(repo_groups, hideList=True))
         repo_type = All(v.ValidForkType(old_data), v.OneOf(supported_backends))
         description = v.UnicodeString(strip=True, min=1, not_empty=True)
         private = v.StringBoolean(if_missing=False)
         copy_permissions = v.StringBoolean(if_missing=False)
         update_after_clone = v.StringBoolean(if_missing=False)
         fork_parent_id = v.UnicodeString()
         chained_validators = [v.ValidForkName(edit, old_data)]
         landing_rev = v.OneOf(landing_revs, hideList=True)
 
     return _RepoForkForm
 
 
 def ApplicationSettingsForm():
     class _ApplicationSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         rhodecode_title = v.UnicodeString(strip=True, min=1, not_empty=True)
         rhodecode_realm = v.UnicodeString(strip=True, min=1, not_empty=True)
         rhodecode_ga_code = v.UnicodeString(strip=True, min=1, not_empty=False)
 
     return _ApplicationSettingsForm
 
 
 def ApplicationVisualisationForm():
     class _ApplicationVisualisationForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         rhodecode_show_public_icon = v.StringBoolean(if_missing=False)
         rhodecode_show_private_icon = v.StringBoolean(if_missing=False)
         rhodecode_stylify_metatags = v.StringBoolean(if_missing=False)
 
         rhodecode_lightweight_dashboard = v.StringBoolean(if_missing=False)
         rhodecode_repository_fields = v.StringBoolean(if_missing=False)
         rhodecode_lightweight_journal = v.StringBoolean(if_missing=False)
 
     return _ApplicationVisualisationForm
 
 
 def ApplicationUiSettingsForm():
     class _ApplicationUiSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = False
         web_push_ssl = v.StringBoolean(if_missing=False)
         paths_root_path = All(
             v.ValidPath(),
             v.UnicodeString(strip=True, min=1, not_empty=True)
         )
         hooks_changegroup_update = v.StringBoolean(if_missing=False)
         hooks_changegroup_repo_size = v.StringBoolean(if_missing=False)
         hooks_changegroup_push_logger = v.StringBoolean(if_missing=False)
         hooks_outgoing_pull_logger = v.StringBoolean(if_missing=False)
 
         extensions_largefiles = v.StringBoolean(if_missing=False)
         extensions_hgsubversion = v.StringBoolean(if_missing=False)
         extensions_hggit = v.StringBoolean(if_missing=False)
 
     return _ApplicationUiSettingsForm
 
 
 def DefaultPermissionsForm(repo_perms_choices, group_perms_choices,
                            register_choices, create_choices, fork_choices):
     class _DefaultPermissionsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         overwrite_default_repo = v.StringBoolean(if_missing=False)
         overwrite_default_group = v.StringBoolean(if_missing=False)
         anonymous = v.StringBoolean(if_missing=False)
         default_repo_perm = v.OneOf(repo_perms_choices)
         default_group_perm = v.OneOf(group_perms_choices)
         default_register = v.OneOf(register_choices)
         default_create = v.OneOf(create_choices)
         default_fork = v.OneOf(fork_choices)
 
     return _DefaultPermissionsForm
 
 
 def DefaultsForm(edit=False, old_data={}, supported_backends=BACKENDS.keys()):
     class _DefaultsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         default_repo_type = v.OneOf(supported_backends)
         default_repo_private = v.StringBoolean(if_missing=False)
         default_repo_enable_statistics = v.StringBoolean(if_missing=False)
         default_repo_enable_downloads = v.StringBoolean(if_missing=False)
         default_repo_enable_locking = v.StringBoolean(if_missing=False)
 
     return _DefaultsForm
 
 
 def LdapSettingsForm(tls_reqcert_choices, search_scope_choices,
                      tls_kind_choices):
     class _LdapSettingsForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
         #pre_validators = [LdapLibValidator]
         ldap_active = v.StringBoolean(if_missing=False)
         ldap_host = v.UnicodeString(strip=True,)
         ldap_port = v.Number(strip=True,)
         ldap_tls_kind = v.OneOf(tls_kind_choices)
         ldap_tls_reqcert = v.OneOf(tls_reqcert_choices)
         ldap_dn_user = v.UnicodeString(strip=True,)
         ldap_dn_pass = v.UnicodeString(strip=True,)
         ldap_base_dn = v.UnicodeString(strip=True,)
         ldap_filter = v.UnicodeString(strip=True,)
         ldap_search_scope = v.OneOf(search_scope_choices)
-        ldap_attr_login = All(
-            v.AttrLoginValidator(),
-            v.UnicodeString(strip=True,)
-        )
+        ldap_attr_login = v.AttrLoginValidator()(not_empty=True)
         ldap_attr_firstname = v.UnicodeString(strip=True,)
         ldap_attr_lastname = v.UnicodeString(strip=True,)
         ldap_attr_email = v.UnicodeString(strip=True,)
 
     return _LdapSettingsForm
 
 
 def UserExtraEmailForm():
     class _UserExtraEmailForm(formencode.Schema):
         email = All(v.UniqSystemEmail(), v.Email(not_empty=True))
     return _UserExtraEmailForm
 
 
 def UserExtraIpForm():
     class _UserExtraIpForm(formencode.Schema):
         ip = v.ValidIp()(not_empty=True)
     return _UserExtraIpForm
 
 
 def PullRequestForm(repo_id):
     class _PullRequestForm(formencode.Schema):
         allow_extra_fields = True
         filter_extra_fields = True
 
         user = v.UnicodeString(strip=True, required=True)
         org_repo = v.UnicodeString(strip=True, required=True)
         org_ref = v.UnicodeString(strip=True, required=True)
         other_repo = v.UnicodeString(strip=True, required=True)
         other_ref = v.UnicodeString(strip=True, required=True)
         revisions = All(#v.NotReviewedRevisions(repo_id)(),
                         v.UniqueList(not_empty=True))
         review_members = v.UniqueList(not_empty=True)
 
         pullrequest_title = v.UnicodeString(strip=True, required=True, min=3)
         pullrequest_desc = v.UnicodeString(strip=True, required=False)
 
         ancestor_rev = v.UnicodeString(strip=True, required=True)
         merge_rev = v.UnicodeString(strip=True, required=True)
 
     return _PullRequestForm

rhodecode/model/validators.py

@@ -1,813 +1,807 b''
 """
 Set of generic validators
 """
 import os
 import re
 import formencode
 import logging
 from collections import defaultdict
 from pylons.i18n.translation import _
 from webhelpers.pylonslib.secure_form import authentication_token
 
 from formencode.validators import (
     UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,
     NotEmpty, IPAddress, CIDR
 )
 from rhodecode.lib.compat import OrderedSet
 from rhodecode.lib import ipaddr
 from rhodecode.lib.utils import repo_name_slug
 from rhodecode.lib.utils2 import safe_int, str2bool
 from rhodecode.model.db import RepoGroup, Repository, UserGroup, User,\
     ChangesetStatus
 from rhodecode.lib.exceptions import LdapImportError
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.lib.auth import HasReposGroupPermissionAny, HasPermissionAny
 
 # silence warnings and pylint
 UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \
     NotEmpty, IPAddress, CIDR
 
 log = logging.getLogger(__name__)
 
 
 class UniqueList(formencode.FancyValidator):
     """
     Unique List !
     """
     messages = dict(
         empty=_('Value cannot be an empty list'),
         missing_value=_('Value cannot be an empty list'),
     )
 
     def _to_python(self, value, state):
         if isinstance(value, list):
             return value
         elif isinstance(value, set):
             return list(value)
         elif isinstance(value, tuple):
             return list(value)
         elif value is None:
             return []
         else:
             return [value]
 
     def empty_value(self, value):
         return []
 
 
 class StateObj(object):
     """
     this is needed to translate the messages using _() in validators
     """
     _ = staticmethod(_)
 
 
 def M(self, key, state=None, **kwargs):
     """
     returns string from self.message based on given key,
     passed kw params are used to substitute %(named)s params inside
     translated strings
 
     :param msg:
     :param state:
     """
     if state is None:
         state = StateObj()
     else:
         state._ = staticmethod(_)
     #inject validator into state object
     return self.message(key, state, **kwargs)
 
 
 def ValidUsername(edit=False, old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'username_exists': _(u'Username "%(username)s" already exists'),
             'system_invalid_username':
                 _(u'Username "%(username)s" is forbidden'),
             'invalid_username':
                 _(u'Username may only contain alphanumeric characters '
                   'underscores, periods or dashes and must begin with '
                   'alphanumeric character')
         }
 
         def validate_python(self, value, state):
             if value in ['default', 'new_user']:
                 msg = M(self, 'system_invalid_username', state, username=value)
                 raise formencode.Invalid(msg, value, state)
             #check if user is unique
             old_un = None
             if edit:
                 old_un = User.get(old_data.get('user_id')).username
 
             if old_un != value or not edit:
                 if User.get_by_username(value, case_insensitive=True):
                     msg = M(self, 'username_exists', state, username=value)
                     raise formencode.Invalid(msg, value, state)
 
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]*$', value) is None:
                 msg = M(self, 'invalid_username', state)
                 raise formencode.Invalid(msg, value, state)
     return _validator
 
 
 def ValidRepoUser():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_username': _(u'Username %(username)s is not valid')
         }
 
         def validate_python(self, value, state):
             try:
                 User.query().filter(User.active == True)\
                     .filter(User.username == value).one()
             except Exception:
                 msg = M(self, 'invalid_username', state, username=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(username=msg)
                 )
 
     return _validator
 
 
 def ValidUserGroup(edit=False, old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_group': _(u'Invalid user group name'),
             'group_exist': _(u'User group "%(usergroup)s" already exists'),
             'invalid_usergroup_name':
                 _(u'user group name may only contain alphanumeric '
                   'characters underscores, periods or dashes and must begin '
                   'with alphanumeric character')
         }
 
         def validate_python(self, value, state):
             if value in ['default']:
                 msg = M(self, 'invalid_group', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(users_group_name=msg)
                 )
             #check if group is unique
             old_ugname = None
             if edit:
                 old_id = old_data.get('users_group_id')
                 old_ugname = UserGroup.get(old_id).users_group_name
 
             if old_ugname != value or not edit:
                 is_existing_group = UserGroup.get_by_group_name(value,
                                                         case_insensitive=True)
                 if is_existing_group:
                     msg = M(self, 'group_exist', state, usergroup=value)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(users_group_name=msg)
                     )
 
             if re.match(r'^[a-zA-Z0-9]{1}[a-zA-Z0-9\-\_\.]+$', value) is None:
                 msg = M(self, 'invalid_usergroup_name', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(users_group_name=msg)
                 )
 
     return _validator
 
 
 def ValidReposGroup(edit=False, old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'group_parent_id': _(u'Cannot assign this group as parent'),
             'group_exists': _(u'Group "%(group_name)s" already exists'),
             'repo_exists':
                 _(u'Repository with name "%(group_name)s" already exists')
         }
 
         def validate_python(self, value, state):
             # TODO WRITE VALIDATIONS
             group_name = value.get('group_name')
             group_parent_id = value.get('group_parent_id')
 
             # slugify repo group just in case :)
             slug = repo_name_slug(group_name)
 
             # check for parent of self
             parent_of_self = lambda: (
                 old_data['group_id'] == int(group_parent_id)
                 if group_parent_id else False
             )
             if edit and parent_of_self():
                 msg = M(self, 'group_parent_id', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(group_parent_id=msg)
                 )
 
             old_gname = None
             if edit:
                 old_gname = RepoGroup.get(old_data.get('group_id')).group_name
 
             if old_gname != group_name or not edit:
 
                 # check group
                 gr = RepoGroup.query()\
                       .filter(RepoGroup.group_name == slug)\
                       .filter(RepoGroup.group_parent_id == group_parent_id)\
                       .scalar()
 
                 if gr:
                     msg = M(self, 'group_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
                     )
 
                 # check for same repo
                 repo = Repository.query()\
                       .filter(Repository.repo_name == slug)\
                       .scalar()
 
                 if repo:
                     msg = M(self, 'repo_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
                     )
 
     return _validator
 
 
 def ValidPassword():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password':
                 _(u'Invalid characters (non-ascii) in password')
         }
 
         def validate_python(self, value, state):
             try:
                 (value or '').decode('ascii')
             except UnicodeError:
                 msg = M(self, 'invalid_password', state)
                 raise formencode.Invalid(msg, value, state,)
     return _validator
 
 
 def ValidPasswordsMatch():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'password_mismatch': _(u'Passwords do not match'),
         }
 
         def validate_python(self, value, state):
 
             pass_val = value.get('password') or value.get('new_password')
             if pass_val != value['password_confirmation']:
                 msg = M(self, 'password_mismatch', state)
                 raise formencode.Invalid(msg, value, state,
                      error_dict=dict(password_confirmation=msg)
                 )
     return _validator
 
 
 def ValidAuth():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password': _(u'invalid password'),
             'invalid_username': _(u'invalid user name'),
             'disabled_account': _(u'Your account is disabled')
         }
 
         def validate_python(self, value, state):
             from rhodecode.lib.auth import authenticate
 
             password = value['password']
             username = value['username']
 
             if not authenticate(username, password):
                 user = User.get_by_username(username)
                 if user and not user.active:
                     log.warning('user %s is disabled' % username)
                     msg = M(self, 'disabled_account', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(username=msg)
                     )
                 else:
                     log.warning('user %s failed to authenticate' % username)
                     msg = M(self, 'invalid_username', state)
                     msg2 = M(self, 'invalid_password', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(username=msg, password=msg2)
                     )
     return _validator
 
 
 def ValidAuthToken():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_token': _(u'Token mismatch')
         }
 
         def validate_python(self, value, state):
             if value != authentication_token():
                 msg = M(self, 'invalid_token', state)
                 raise formencode.Invalid(msg, value, state)
     return _validator
 
 
 def ValidRepoName(edit=False, old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_repo_name':
                 _(u'Repository name %(repo)s is disallowed'),
             'repository_exists':
                 _(u'Repository named %(repo)s already exists'),
             'repository_in_group_exists': _(u'Repository "%(repo)s" already '
                                             'exists in group "%(group)s"'),
             'same_group_exists': _(u'Repository group with name "%(repo)s" '
                                    'already exists')
         }
 
         def _to_python(self, value, state):
             repo_name = repo_name_slug(value.get('repo_name', ''))
             repo_group = value.get('repo_group')
             if repo_group:
                 gr = RepoGroup.get(repo_group)
                 group_path = gr.full_path
                 group_name = gr.group_name
                 # value needs to be aware of group name in order to check
                 # db key This is an actual just the name to store in the
                 # database
                 repo_name_full = group_path + RepoGroup.url_sep() + repo_name
             else:
                 group_name = group_path = ''
                 repo_name_full = repo_name
 
             value['repo_name'] = repo_name
             value['repo_name_full'] = repo_name_full
             value['group_path'] = group_path
             value['group_name'] = group_name
             return value
 
         def validate_python(self, value, state):
 
             repo_name = value.get('repo_name')
             repo_name_full = value.get('repo_name_full')
             group_path = value.get('group_path')
             group_name = value.get('group_name')
 
             if repo_name in [ADMIN_PREFIX, '']:
                 msg = M(self, 'invalid_repo_name', state, repo=repo_name)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_name=msg)
                 )
 
             rename = old_data.get('repo_name') != repo_name_full
             create = not edit
             if rename or create:
 
                 if group_path != '':
                     if Repository.get_by_repo_name(repo_name_full):
                         msg = M(self, 'repository_in_group_exists', state,
                                 repo=repo_name, group=group_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
                         )
                 elif RepoGroup.get_by_group_name(repo_name_full):
                         msg = M(self, 'same_group_exists', state,
                                 repo=repo_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
                         )
 
                 elif Repository.get_by_repo_name(repo_name_full):
                         msg = M(self, 'repository_exists', state,
                                 repo=repo_name)
                         raise formencode.Invalid(msg, value, state,
                             error_dict=dict(repo_name=msg)
                         )
             return value
     return _validator
 
 
 def ValidForkName(*args, **kwargs):
     return ValidRepoName(*args, **kwargs)
 
 
 def SlugifyName():
     class _validator(formencode.validators.FancyValidator):
 
         def _to_python(self, value, state):
             return repo_name_slug(value)
 
         def validate_python(self, value, state):
             pass
 
     return _validator
 
 
 def ValidCloneUri():
     from rhodecode.lib.utils import make_ui
 
     def url_handler(repo_type, url, ui=None):
         if repo_type == 'hg':
             from rhodecode.lib.vcs.backends.hg.repository import MercurialRepository
             from mercurial.httppeer import httppeer
             if url.startswith('http'):
                 ## initially check if it's at least the proper URL
                 ## or does it pass basic auth
                 MercurialRepository._check_url(url)
                 httppeer(ui, url)._capabilities()
             elif url.startswith('svn+http'):
                 from hgsubversion.svnrepo import svnremoterepo
                 svnremoterepo(ui, url).capabilities
             elif url.startswith('git+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url))
 
         elif repo_type == 'git':
             from rhodecode.lib.vcs.backends.git.repository import GitRepository
             if url.startswith('http'):
                 ## initially check if it's at least the proper URL
                 ## or does it pass basic auth
                 GitRepository._check_url(url)
             elif url.startswith('svn+http'):
                 raise NotImplementedError()
             elif url.startswith('hg+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url))
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'clone_uri': _(u'invalid clone url'),
             'invalid_clone_uri': _(u'Invalid clone url, provide a '
                                     'valid clone http(s)/svn+http(s) url')
         }
 
         def validate_python(self, value, state):
             repo_type = value.get('repo_type')
             url = value.get('clone_uri')
 
             if not url:
                 pass
             else:
                 try:
                     url_handler(repo_type, url, make_ui('db', clear_session=False))
                 except Exception:
                     log.exception('Url validation failed')
                     msg = M(self, 'clone_uri')
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(clone_uri=msg)
                     )
     return _validator
 
 
 def ValidForkType(old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_fork_type': _(u'Fork have to be the same type as parent')
         }
 
         def validate_python(self, value, state):
             if old_data['repo_type'] != value:
                 msg = M(self, 'invalid_fork_type', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(repo_type=msg)
                 )
     return _validator
 
 
 def CanWriteGroup(old_data=None):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _(u"You don't have permissions "
                                    "to create repository in this group"),
             'permission_denied_root': _(u"no permission to create repository "
                                         "in root location")
         }
 
         def _to_python(self, value, state):
             #root location
             if value in [-1, "-1"]:
                 return None
             return value
 
         def validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr else None  # None means ROOT location
             val = HasReposGroupPermissionAny('group.write', 'group.admin')
             can_create_repos = HasPermissionAny('hg.admin', 'hg.create.repository')
             forbidden = not val(gr_name, 'can write into group validator')
             value_changed = True  # old_data['repo_group'].get('group_id') != safe_int(value)
             if value_changed:  # do check if we changed the value
                 #parent group need to be existing
                 if gr and forbidden:
                     msg = M(self, 'permission_denied', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
                     )
                 ## check if we can write to root location !
                 elif gr is None and not can_create_repos():
                     msg = M(self, 'permission_denied_root', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(repo_type=msg)
                     )
 
     return _validator
 
 
 def CanCreateGroup(can_create_in_root=False):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'permission_denied': _(u"You don't have permissions "
                                    "to create a group in this location")
         }
 
         def to_python(self, value, state):
             #root location
             if value in [-1, "-1"]:
                 return None
             return value
 
         def validate_python(self, value, state):
             gr = RepoGroup.get(value)
             gr_name = gr.group_name if gr else None  # None means ROOT location
 
             if can_create_in_root and gr is None:
                 #we can create in root, we're fine no validations required
                 return
 
             forbidden_in_root = gr is None and not can_create_in_root
             val = HasReposGroupPermissionAny('group.admin')
             forbidden = not val(gr_name, 'can create group validator')
             if forbidden_in_root or forbidden:
                 msg = M(self, 'permission_denied', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(group_parent_id=msg)
                 )
 
     return _validator
 
 
 def ValidPerms(type_='repo'):
     if type_ == 'group':
         EMPTY_PERM = 'group.none'
     elif type_ == 'repo':
         EMPTY_PERM = 'repository.none'
 
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'perm_new_member_name':
                 _(u'This username or user group name is not valid')
         }
 
         def to_python(self, value, state):
             perms_update = OrderedSet()
             perms_new = OrderedSet()
             # build a list of permission to update and new permission to create
 
             #CLEAN OUT ORG VALUE FROM NEW MEMBERS, and group them using
             new_perms_group = defaultdict(dict)
             for k, v in value.copy().iteritems():
                 if k.startswith('perm_new_member'):
                     del value[k]
                     _type, part = k.split('perm_new_member_')
                     args = part.split('_')
                     if len(args) == 1:
                         new_perms_group[args[0]]['perm'] = v
                     elif len(args) == 2:
                         _key, pos = args
                         new_perms_group[pos][_key] = v
 
             # fill new permissions in order of how they were added
             for k in sorted(map(int, new_perms_group.keys())):
                 perm_dict = new_perms_group[str(k)]
                 new_member = perm_dict.get('name')
                 new_perm = perm_dict.get('perm')
                 new_type = perm_dict.get('type')
                 if new_member and new_perm and new_type:
                     perms_new.add((new_member, new_perm, new_type))
 
             for k, v in value.iteritems():
                 if k.startswith('u_perm_') or k.startswith('g_perm_'):
                     member = k[7:]
                     t = {'u': 'user',
                          'g': 'users_group'
                     }[k[0]]
                     if member == 'default':
                         if str2bool(value.get('repo_private')):
                             # set none for default when updating to
                             # private repo protects agains form manipulation
                             v = EMPTY_PERM
                     perms_update.add((member, v, t))
 
             value['perms_updates'] = list(perms_update)
             value['perms_new'] = list(perms_new)
 
             # update permissions
             for k, v, t in perms_new:
                 try:
                     if t is 'user':
                         self.user_db = User.query()\
                             .filter(User.active == True)\
                             .filter(User.username == k).one()
                     if t is 'users_group':
                         self.user_db = UserGroup.query()\
                             .filter(UserGroup.users_group_active == True)\
                             .filter(UserGroup.users_group_name == k).one()
 
                 except Exception:
                     log.exception('Updated permission failed')
                     msg = M(self, 'perm_new_member_type', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(perm_new_member_name=msg)
                     )
             return value
     return _validator
 
 
 def ValidSettings():
     class _validator(formencode.validators.FancyValidator):
         def _to_python(self, value, state):
             # settings  form for users that are not admin
             # can't edit certain parameters, it's extra backup if they mangle
             # with forms
 
             forbidden_params = [
                 'user', 'repo_type', 'repo_enable_locking',
                 'repo_enable_downloads', 'repo_enable_statistics'
             ]
 
             for param in forbidden_params:
                 if param in value:
                     del value[param]
             return value
 
         def validate_python(self, value, state):
             pass
     return _validator
 
 
 def ValidPath():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_path': _(u'This is not a valid path')
         }
 
         def validate_python(self, value, state):
             if not os.path.isdir(value):
                 msg = M(self, 'invalid_path', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(paths_root_path=msg)
                 )
     return _validator
 
 
 def UniqSystemEmail(old_data={}):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'email_taken': _(u'This e-mail address is already taken')
         }
 
         def _to_python(self, value, state):
             return value.lower()
 
         def validate_python(self, value, state):
             if (old_data.get('email') or '').lower() != value:
                 user = User.get_by_email(value, case_insensitive=True)
                 if user:
                     msg = M(self, 'email_taken', state)
                     raise formencode.Invalid(msg, value, state,
                         error_dict=dict(email=msg)
                     )
     return _validator
 
 
 def ValidSystemEmail():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'non_existing_email': _(u'e-mail "%(email)s" does not exist.')
         }
 
         def _to_python(self, value, state):
             return value.lower()
 
         def validate_python(self, value, state):
             user = User.get_by_email(value, case_insensitive=True)
             if user is None:
                 msg = M(self, 'non_existing_email', state, email=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(email=msg)
                 )
 
     return _validator
 
 
 def LdapLibValidator():
     class _validator(formencode.validators.FancyValidator):
         messages = {
 
         }
 
         def validate_python(self, value, state):
             try:
                 import ldap
                 ldap  # pyflakes silence !
             except ImportError:
                 raise LdapImportError()
 
     return _validator
 
 
 def AttrLoginValidator():
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_cn':
                   _(u'The LDAP Login attribute of the CN must be specified - '
                     'this is the name of the attribute that is equivalent '
                     'to "username"')
         }
-
-        def validate_python(self, value, state):
-            if not value or not isinstance(value, (str, unicode)):
-                msg = M(self, 'invalid_cn', state)
-                raise formencode.Invalid(msg, value, state,
-                    error_dict=dict(ldap_attr_login=msg)
-                )
+        messages['empty'] = messages['invalid_cn']
 
     return _validator
 
 
 def NotReviewedRevisions(repo_id):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'rev_already_reviewed':
                   _(u'Revisions %(revs)s are already part of pull request '
                     'or have set status')
         }
 
         def validate_python(self, value, state):
             # check revisions if they are not reviewed, or a part of another
             # pull request
             statuses = ChangesetStatus.query()\
                 .filter(ChangesetStatus.revision.in_(value))\
                 .filter(ChangesetStatus.repo_id == repo_id)\
                 .all()
 
             errors = []
             for cs in statuses:
                 if cs.pull_request_id:
                     errors.append(['pull_req', cs.revision[:12]])
                 elif cs.status:
                     errors.append(['status', cs.revision[:12]])
 
             if errors:
                 revs = ','.join([x[1] for x in errors])
                 msg = M(self, 'rev_already_reviewed', state, revs=revs)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(revisions=revs)
                 )
 
     return _validator
 
 
 def ValidIp():
     class _validator(CIDR):
         messages = dict(
             badFormat=_('Please enter a valid IPv4 or IpV6 address'),
             illegalBits=_('The network size (bits) must be within the range'
                 ' of 0-32 (not %(bits)r)'))
 
         def to_python(self, value, state):
             v = super(_validator, self).to_python(value, state)
             v = v.strip()
             net = ipaddr.IPNetwork(address=v)
             if isinstance(net, ipaddr.IPv4Network):
                 #if IPv4 doesn't end with a mask, add /32
                 if '/' not in value:
                     v += '/32'
             if isinstance(net, ipaddr.IPv6Network):
                 #if IPv6 doesn't end with a mask, add /128
                 if '/' not in value:
                     v += '/128'
             return v
 
         def validate_python(self, value, state):
             try:
                 addr = value.strip()
                 #this raises an ValueError if address is not IpV4 or IpV6
                 ipaddr.IPNetwork(address=addr)
             except ValueError:
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
 
     return _validator
 
 
 def FieldKey():
     class _validator(formencode.validators.FancyValidator):
         messages = dict(
             badFormat=_('Key name can only consist of letters, '
                         'underscore, dash or numbers'),)
 
         def validate_python(self, value, state):
             if not re.match('[a-zA-Z0-9_-]+$', value):
                 raise formencode.Invalid(self.message('badFormat', state),
                                          value, state)
     return _validator

rhodecode/tests/__init__.py

@@ -1,175 +1,184 b''
 """Pylons application test package
 
 This package assumes the Pylons environment is already loaded, such as
 when this script is imported from the `nosetests --with-pylons=test.ini`
 command.
 
 This module initializes the application via ``websetup`` (`paster
 setup-app`) and provides the base testing objects.
 
 nosetests -x - fail on first error
 nosetests rhodecode.tests.functional.test_admin_settings:TestSettingsController.test_my_account
 nosetests --pdb --pdb-failures
 nosetests --with-coverage --cover-package=rhodecode.model.validators rhodecode.tests.test_validators
 
 optional FLAGS:
     RC_WHOOSH_TEST_DISABLE=1 - skip whoosh index building and tests
     RC_NO_TMP_PATH=1 - disable new temp path for tests, used mostly for test_vcs_operations
 
 """
 import os
 import time
 import logging
 import datetime
 import hashlib
 import tempfile
 from os.path import join as jn
 
 from unittest import TestCase
 from tempfile import _RandomNameSequence
 
 from paste.deploy import loadapp
 from paste.script.appinstall import SetupCommand
 from pylons import config, url
 from routes.util import URLGenerator
 from webtest import TestApp
 from nose.plugins.skip import SkipTest
 
 from rhodecode import is_windows
 from rhodecode.model.meta import Session
 from rhodecode.model.db import User
 from rhodecode.tests.nose_parametrized import parameterized
 
 import pylons.test
 from rhodecode.lib.utils2 import safe_unicode, safe_str
 
 
 os.environ['TZ'] = 'UTC'
 if not is_windows:
     time.tzset()
 
 log = logging.getLogger(__name__)
 
 __all__ = [
     'parameterized', 'environ', 'url', 'get_new_dir', 'TestController',
-    'SkipTest',
+    'SkipTest', 'ldap_lib_installed',
     'TESTS_TMP_PATH', 'HG_REPO', 'GIT_REPO', 'NEW_HG_REPO', 'NEW_GIT_REPO',
     'HG_FORK', 'GIT_FORK', 'TEST_USER_ADMIN_LOGIN', 'TEST_USER_ADMIN_PASS',
     'TEST_USER_REGULAR_LOGIN', 'TEST_USER_REGULAR_PASS',
     'TEST_USER_REGULAR_EMAIL', 'TEST_USER_REGULAR2_LOGIN',
     'TEST_USER_REGULAR2_PASS', 'TEST_USER_REGULAR2_EMAIL', 'TEST_HG_REPO',
     'TEST_HG_REPO_CLONE', 'TEST_HG_REPO_PULL', 'TEST_GIT_REPO',
     'TEST_GIT_REPO_CLONE', 'TEST_GIT_REPO_PULL', 'HG_REMOTE_REPO',
     'GIT_REMOTE_REPO', 'SCM_TESTS',
 ]
 
 # Invoke websetup with the current config file
 # SetupCommand('setup-app').run([config_file])
 
 environ = {}
 
 #SOME GLOBALS FOR TESTS
 
 TESTS_TMP_PATH = jn('/', 'tmp', 'rc_test_%s' % _RandomNameSequence().next())
 TEST_USER_ADMIN_LOGIN = 'test_admin'
 TEST_USER_ADMIN_PASS = 'test12'
 TEST_USER_ADMIN_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR_LOGIN = 'test_regular'
 TEST_USER_REGULAR_PASS = 'test12'
 TEST_USER_REGULAR_EMAIL = '[e-mail]'
 
 TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 TEST_USER_REGULAR2_PASS = 'test12'
 TEST_USER_REGULAR2_EMAIL = '[e-mail]'
 
 HG_REPO = 'vcs_test_hg'
 GIT_REPO = 'vcs_test_git'
 
 NEW_HG_REPO = 'vcs_test_hg_new'
 NEW_GIT_REPO = 'vcs_test_git_new'
 
 HG_FORK = 'vcs_test_hg_fork'
 GIT_FORK = 'vcs_test_git_fork'
 
 ## VCS
 SCM_TESTS = ['hg', 'git']
 uniq_suffix = str(int(time.mktime(datetime.datetime.now().timetuple())))
 
 GIT_REMOTE_REPO = 'git://github.com/codeinn/vcs.git'
 
 TEST_GIT_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 TEST_GIT_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcsgitclone%s' % uniq_suffix)
 TEST_GIT_REPO_PULL = jn(TESTS_TMP_PATH, 'vcsgitpull%s' % uniq_suffix)
 
 
 HG_REMOTE_REPO = 'http://bitbucket.org/marcinkuzminski/vcs'
 
 TEST_HG_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 TEST_HG_REPO_CLONE = jn(TESTS_TMP_PATH, 'vcshgclone%s' % uniq_suffix)
 TEST_HG_REPO_PULL = jn(TESTS_TMP_PATH, 'vcshgpull%s' % uniq_suffix)
 
 TEST_DIR = tempfile.gettempdir()
 TEST_REPO_PREFIX = 'vcs-test'
 
 # cached repos if any !
 # comment out to get some other repos from bb or github
 GIT_REMOTE_REPO = jn(TESTS_TMP_PATH, GIT_REPO)
 HG_REMOTE_REPO = jn(TESTS_TMP_PATH, HG_REPO)
 
+#skip ldap tests if LDAP lib is not installed
+ldap_lib_installed = False
+try:
+    import ldap
+    ldap_lib_installed = True
+except ImportError:
+    # means that python-ldap is not installed
+    pass
+
 
 def get_new_dir(title):
     """
     Returns always new directory path.
     """
     from rhodecode.tests.vcs.utils import get_normalized_path
     name = TEST_REPO_PREFIX
     if title:
         name = '-'.join((name, title))
     hex = hashlib.sha1(str(time.time())).hexdigest()
     name = '-'.join((name, hex))
     path = os.path.join(TEST_DIR, name)
     return get_normalized_path(path)
 
 
 class TestController(TestCase):
 
     def __init__(self, *args, **kwargs):
         wsgiapp = pylons.test.pylonsapp
         config = wsgiapp.config
 
         self.app = TestApp(wsgiapp)
         url._push_object(URLGenerator(config['routes.map'], environ))
         self.Session = Session
         self.index_location = config['app_conf']['index_dir']
         TestCase.__init__(self, *args, **kwargs)
 
     def log_user(self, username=TEST_USER_ADMIN_LOGIN,
                  password=TEST_USER_ADMIN_PASS):
         self._logged_username = username
         response = self.app.post(url(controller='login', action='index'),
                                  {'username': username,
                                   'password': password})
 
         if 'invalid user name' in response.body:
             self.fail('could not login using %s %s' % (username, password))
 
         self.assertEqual(response.status, '302 Found')
         ses = response.session['rhodecode_user']
         self.assertEqual(ses.get('username'), username)
         response = response.follow()
         self.assertEqual(ses.get('is_authenticated'), True)
 
         return response.session['rhodecode_user']
 
     def _get_logged_user(self):
         return User.get_by_username(self._logged_username)
 
     def checkSessionFlash(self, response, msg):
         self.assertTrue('flash' in response.session,
                         msg='Response session:%r have no flash'
                         % response.session)
         if not msg in response.session['flash'][0][1]:
             msg = u'msg `%s` not found in session flash: got `%s` instead' % (
                       msg, response.session['flash'][0][1])
             self.fail(safe_str(msg))

rhodecode/tests/functional/test_admin_ldap_settings.py

@@ -1,85 +1,98 b''
 from rhodecode.tests import *
 from rhodecode.model.db import RhodeCodeSetting
 
-skip_ldap_test = False
-try:
-    import ldap
-except ImportError:
-    # means that python-ldap is not installed
-    skip_ldap_test = True
-    pass
-
-
 class TestLdapSettingsController(TestController):
 
     def test_index(self):
         self.log_user()
         response = self.app.get(url(controller='admin/ldap_settings',
                                     action='index'))
         response.mustcontain('LDAP administration')
 
     def test_ldap_save_settings(self):
         self.log_user()
-        if skip_ldap_test:
+        if ldap_lib_installed:
             raise SkipTest('skipping due to missing ldap lib')
 
         test_url = url(controller='admin/ldap_settings',
                        action='ldap_settings')
 
         response = self.app.post(url=test_url,
             params={'ldap_host' : u'dc.example.com',
                     'ldap_port' : '999',
                     'ldap_tls_kind' : 'PLAIN',
                     'ldap_tls_reqcert' : 'NEVER',
                     'ldap_dn_user':'test_user',
                     'ldap_dn_pass':'test_pass',
                     'ldap_base_dn':'test_base_dn',
                     'ldap_filter':'test_filter',
                     'ldap_search_scope':'BASE',
                     'ldap_attr_login':'test_attr_login',
                     'ldap_attr_firstname':'ima',
                     'ldap_attr_lastname':'tester',
                     'ldap_attr_email':'[e-mail]' })
 
         new_settings = RhodeCodeSetting.get_ldap_settings()
         self.assertEqual(new_settings['ldap_host'], u'dc.example.com',
                          'fail db write compare')
 
         self.checkSessionFlash(response,
                                'LDAP settings updated successfully')
 
-    def test_ldap_error_form(self):
+    def test_ldap_error_form_wrong_port_number(self):
         self.log_user()
-        if skip_ldap_test:
+        if ldap_lib_installed:
             raise SkipTest('skipping due to missing ldap lib')
 
         test_url = url(controller='admin/ldap_settings',
                        action='ldap_settings')
 
         response = self.app.post(url=test_url,
             params={'ldap_host' : '',
-                    'ldap_port' : 'i-should-be-number',
+                    'ldap_port': 'i-should-be-number',  # bad port num
+                    'ldap_tls_kind': 'PLAIN',
+                    'ldap_tls_reqcert': 'NEVER',
+                    'ldap_dn_user': '',
+                    'ldap_dn_pass': '',
+                    'ldap_base_dn': '',
+                    'ldap_filter': '',
+                    'ldap_search_scope': 'BASE',
+                    'ldap_attr_login': '',
+                    'ldap_attr_firstname': '',
+                    'ldap_attr_lastname': '',
+                    'ldap_attr_email': ''})
+
+        response.mustcontain("""<span class="error-message">"""
+                             """Please enter a number</span><br />""")
+
+    def test_ldap_error_form(self):
+        self.log_user()
+        if ldap_lib_installed:
+            raise SkipTest('skipping due to missing ldap lib')
+
+        test_url = url(controller='admin/ldap_settings',
+                       action='ldap_settings')
+
+        response = self.app.post(url=test_url,
+            params={'ldap_host': 'Host',
+                    'ldap_port': '123',
                     'ldap_tls_kind' : 'PLAIN',
                     'ldap_tls_reqcert' : 'NEVER',
                     'ldap_dn_user':'',
                     'ldap_dn_pass':'',
                     'ldap_base_dn':'',
                     'ldap_filter':'',
                     'ldap_search_scope':'BASE',
                     'ldap_attr_login':'', #  <----- missing required input
                     'ldap_attr_firstname':'',
                     'ldap_attr_lastname':'',
                     'ldap_attr_email':'' })
 
         response.mustcontain("""<span class="error-message">The LDAP Login"""
                              """ attribute of the CN must be specified""")
 
-
-        response.mustcontain("""<span class="error-message">Please """
-                             """enter a number</span>""")
-
     def test_ldap_login(self):
         pass
 
     def test_ldap_login_incorrect(self):
         pass

rhodecode/tests/test_validators.py

@@ -1,248 +1,252 b''
 # -*- coding: utf-8 -*-
 import unittest
 import formencode
 
 from rhodecode.tests import *
 
 from rhodecode.model import validators as v
 from rhodecode.model.users_group import UserGroupModel
 
 from rhodecode.model.meta import Session
 from rhodecode.model.repos_group import ReposGroupModel
 from rhodecode.config.routing import ADMIN_PREFIX
 from rhodecode.model.db import ChangesetStatus, Repository
 from rhodecode.model.changeset_status import ChangesetStatusModel
 
 
 class TestReposGroups(unittest.TestCase):
 
     def setUp(self):
         pass
 
     def tearDown(self):
         Session.remove()
 
     def test_Message_extractor(self):
         validator = v.ValidUsername()
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
 
         class StateObj(object):
             pass
 
         self.assertRaises(formencode.Invalid,
                           validator.to_python, 'default', StateObj)
 
     def test_ValidUsername(self):
         validator = v.ValidUsername()
 
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
         self.assertRaises(formencode.Invalid, validator.to_python, 'new_user')
         self.assertRaises(formencode.Invalid, validator.to_python, '.,')
         self.assertRaises(formencode.Invalid, validator.to_python,
                           TEST_USER_ADMIN_LOGIN)
         self.assertEqual('test', validator.to_python('test'))
 
         validator = v.ValidUsername(edit=True, old_data={'user_id': 1})
 
     def test_ValidRepoUser(self):
         validator = v.ValidRepoUser()
         self.assertRaises(formencode.Invalid, validator.to_python, 'nouser')
         self.assertEqual(TEST_USER_ADMIN_LOGIN,
                          validator.to_python(TEST_USER_ADMIN_LOGIN))
 
     def test_ValidUserGroup(self):
         validator = v.ValidUserGroup()
         self.assertRaises(formencode.Invalid, validator.to_python, 'default')
         self.assertRaises(formencode.Invalid, validator.to_python, '.,')
 
         gr = UserGroupModel().create('test')
         gr2 = UserGroupModel().create('tes2')
         Session.commit()
         self.assertRaises(formencode.Invalid, validator.to_python, 'test')
         assert gr.users_group_id != None
         validator = v.ValidUserGroup(edit=True,
                                     old_data={'users_group_id':
                                               gr2.users_group_id})
 
         self.assertRaises(formencode.Invalid, validator.to_python, 'test')
         self.assertRaises(formencode.Invalid, validator.to_python, 'TesT')
         self.assertRaises(formencode.Invalid, validator.to_python, 'TEST')
         UserGroupModel().delete(gr)
         UserGroupModel().delete(gr2)
         Session.commit()
 
     def test_ValidReposGroup(self):
         validator = v.ValidReposGroup()
         model = ReposGroupModel()
         self.assertRaises(formencode.Invalid, validator.to_python,
                           {'group_name': HG_REPO, })
         gr = model.create(group_name='test_gr', group_description='desc',
                           parent=None,
                           just_db=True,
                           owner=TEST_USER_ADMIN_LOGIN)
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'group_name': gr.group_name, })
 
         validator = v.ValidReposGroup(edit=True,
                                       old_data={'group_id':  gr.group_id})
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {
                                         'group_name': gr.group_name + 'n',
                                         'group_parent_id': gr.group_id
                                         })
         model.delete(gr)
 
     def test_ValidPassword(self):
         validator = v.ValidPassword()
         self.assertEqual('lol', validator.to_python('lol'))
         self.assertEqual(None, validator.to_python(None))
         self.assertRaises(formencode.Invalid, validator.to_python, 'ąćżź')
 
     def test_ValidPasswordsMatch(self):
         validator = v.ValidPasswordsMatch()
         self.assertRaises(formencode.Invalid,
                     validator.to_python, {'password': 'pass',
                                           'password_confirmation': 'pass2'})
 
         self.assertRaises(formencode.Invalid,
                     validator.to_python, {'new_password': 'pass',
                                           'password_confirmation': 'pass2'})
 
         self.assertEqual({'new_password': 'pass',
                           'password_confirmation': 'pass'},
                     validator.to_python({'new_password': 'pass',
                                          'password_confirmation': 'pass'}))
 
         self.assertEqual({'password': 'pass',
                           'password_confirmation': 'pass'},
                     validator.to_python({'password': 'pass',
                                          'password_confirmation': 'pass'}))
 
     def test_ValidAuth(self):
         validator = v.ValidAuth()
         valid_creds = {
             'username': TEST_USER_REGULAR2_LOGIN,
             'password': TEST_USER_REGULAR2_PASS,
         }
         invalid_creds = {
             'username': 'err',
             'password': 'err',
         }
         self.assertEqual(valid_creds, validator.to_python(valid_creds))
         self.assertRaises(formencode.Invalid,
                           validator.to_python, invalid_creds)
 
     def test_ValidAuthToken(self):
         validator = v.ValidAuthToken()
         # this is untestable without a threadlocal
 #        self.assertRaises(formencode.Invalid,
 #                          validator.to_python, 'BadToken')
         validator
 
     def test_ValidRepoName(self):
         validator = v.ValidRepoName()
 
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': ''})
 
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': HG_REPO})
 
         gr = ReposGroupModel().create(group_name='group_test',
                                       group_description='desc',
                                       parent=None,
                                       owner=TEST_USER_ADMIN_LOGIN)
         self.assertRaises(formencode.Invalid,
                           validator.to_python, {'repo_name': gr.group_name})
 
         #TODO: write an error case for that ie. create a repo withinh a group
 #        self.assertRaises(formencode.Invalid,
 #                          validator.to_python, {'repo_name': 'some',
 #                                                'repo_group': gr.group_id})
 
     def test_ValidForkName(self):
         # this uses ValidRepoName validator
         assert True
 
     @parameterized.expand([
         ('test', 'test'), ('lolz!', 'lolz'), ('  aavv', 'aavv'),
         ('ala ma kota', 'ala-ma-kota'), ('@nooo', 'nooo'),
         ('$!haha lolz !', 'haha-lolz'), ('$$$$$', ''), ('{}OK!', 'OK'),
         ('/]re po', 're-po')])
     def test_SlugifyName(self, name, expected):
         validator = v.SlugifyName()
         self.assertEqual(expected, validator.to_python(name))
 
     def test_ValidCloneUri(self):
             #TODO: write this one
             pass
 
     def test_ValidForkType(self):
             validator = v.ValidForkType(old_data={'repo_type': 'hg'})
             self.assertEqual('hg', validator.to_python('hg'))
             self.assertRaises(formencode.Invalid, validator.to_python, 'git')
 
     def test_ValidPerms(self):
             #TODO: write this one
             pass
 
     def test_ValidSettings(self):
         validator = v.ValidSettings()
         self.assertEqual({'pass': 'pass'},
                          validator.to_python(value={'user': 'test',
                                                     'pass': 'pass'}))
 
         self.assertEqual({'user2': 'test', 'pass': 'pass'},
                          validator.to_python(value={'user2': 'test',
                                                     'pass': 'pass'}))
 
     def test_ValidPath(self):
             validator = v.ValidPath()
             self.assertEqual(TESTS_TMP_PATH,
                              validator.to_python(TESTS_TMP_PATH))
             self.assertRaises(formencode.Invalid, validator.to_python,
                               '/no_such_dir')
 
     def test_UniqSystemEmail(self):
         validator = v.UniqSystemEmail(old_data={})
 
         self.assertEqual('[e-mail]',
                          validator.to_python('[e-mail]'))
 
         email = TEST_USER_REGULAR2_EMAIL
         self.assertRaises(formencode.Invalid, validator.to_python, email)
 
     def test_ValidSystemEmail(self):
         validator = v.ValidSystemEmail()
         email = TEST_USER_REGULAR2_EMAIL
 
         self.assertEqual(email, validator.to_python(email))
         self.assertRaises(formencode.Invalid, validator.to_python, 'err')
 
     def test_LdapLibValidator(self):
+        if ldap_lib_installed:
+            validator = v.LdapLibValidator()
+            self.assertEqual("DN", validator.to_python('DN'))
+        else:
         validator = v.LdapLibValidator()
         self.assertRaises(v.LdapImportError, validator.to_python, 'err')
 
     def test_AttrLoginValidator(self):
         validator = v.AttrLoginValidator()
-        self.assertRaises(formencode.Invalid, validator.to_python, 123)
+        self.assertEqual('DN_attr', validator.to_python('DN_attr'))
 
     def test_NotReviewedRevisions(self):
         repo_id = Repository.get_by_repo_name(HG_REPO).repo_id
         validator = v.NotReviewedRevisions(repo_id)
         rev = '0' * 40
         # add status for a rev, that should throw an error because it is already
         # reviewed
         new_status = ChangesetStatus()
         new_status.author = ChangesetStatusModel()._get_user(TEST_USER_ADMIN_LOGIN)
         new_status.repo = ChangesetStatusModel()._get_repo(HG_REPO)
         new_status.status = ChangesetStatus.STATUS_APPROVED
         new_status.comment = None
         new_status.revision = rev
         Session().add(new_status)
         Session().commit()
         try:
             self.assertRaises(formencode.Invalid, validator.to_python, [rev])
         finally:
             Session().delete(new_status)
             Session().commit()