Changeset - ff2ea58debb5
[Not reviewed]
beta
0 5 0
Marcin Kuzminski - 13 years ago 2013-04-04 18:31:15
marcin@python-works.com
fixed ldap tests when ldap lib is installed
5 files changed with 47 insertions and 30 deletions:
0 comments (0 inline, 0 general)
rhodecode/model/forms.py
Show inline comments
 
""" 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
Show inline comments
 
@@ -342,472 +342,466 @@ def ValidRepoName(edit=False, old_data={
 
            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
Show inline comments
 
"""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 = 'test_admin@mail.com'
 

	
 
TEST_USER_REGULAR_LOGIN = 'test_regular'
 
TEST_USER_REGULAR_PASS = 'test12'
 
TEST_USER_REGULAR_EMAIL = 'test_regular@mail.com'
 

	
 
TEST_USER_REGULAR2_LOGIN = 'test_regular2'
 
TEST_USER_REGULAR2_PASS = 'test12'
 
TEST_USER_REGULAR2_EMAIL = 'test_regular2@mail.com'
 

	
 
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
Show inline comments
 
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':'test@example.com' })
 

	
 
        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
Show inline comments
 
# -*- 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('mail@python.org',
 
                         validator.to_python('MaiL@Python.org'))
 

	
 
        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()
0 comments (0 inline, 0 general)