kallithea Changeset - 221d6a002601

Changeset - 221d6a002601

Parent rev.

Child rev.

[Not reviewed]

default

0 4 0

Mads Kiilerich - 10 years ago 2015-06-09 22:53:24
madski@unity3d.com

repos: cleanup of stuff related to clone_uri

4 files changed with 8 insertions and 27 deletions:

kallithea/model/validators.py

kallithea/templates/admin/repos/repo_add_base.html

kallithea/templates/admin/repos/repo_edit_settings.html

kallithea/tests/functional/test_admin_repos.py

0 comments (0 inline, 0 general)

kallithea/model/validators.py

➞

Show inline comments

@@ @@ -90,771 +90,771 @@ def UniqueListFromString(): @@
             return [c for c in value if not (c in seen or seen.add(c))]
         def empty_value(self, value):
             return []
     return _UniqueListFromString
 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 or underscore')
+        }
         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 ValidRegex(msg=None):
     class _validator(formencode.validators.Regex):
         messages = dict(invalid=msg or _('The input is not valid'))
     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 sqlalchemy.exc.InvalidRequestError: # NoResultFound/MultipleResultsFound
                 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 ValidRepoGroup(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 ValidOldPassword(username):
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_password': _(u'Invalid old password')
+        }
         def validate_python(self, value, state):
             from kallithea.lib import auth_modules
             if not auth_modules.authenticate(username, value, ''):
                 msg = M(self, 'invalid_password', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(current_password=msg)
+                )
     return _validator
 def ValidPasswordsMatch(passwd='new_password', passwd_confirmation='password_confirmation'):
     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(passwd)
             if pass_val != value[passwd_confirmation]:
                 msg = M(self, 'password_mismatch', state)
                 raise formencode.Invalid(msg, value, state,
                      error_dict={passwd:msg, passwd_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 kallithea.lib import auth_modules
             password = value['password']
             username = value['username']
             if not auth_modules.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 kallithea.lib.utils import make_ui
     def url_handler(repo_type, url, ui):
         if repo_type == 'hg':
             from kallithea.lib.vcs.backends.hg.repository import MercurialRepository
             if url.startswith('http') or url.startswith('ssh'):
                 # initially check if it's at least the proper URL
                 # or does it pass basic auth
                 MercurialRepository._check_url(url, ui)
             elif url.startswith('svn+http'):
                 from hgsubversion.svnrepo import svnremoterepo
                 svnremoterepo(ui, url).svn.uuid
             elif url.startswith('git+http'):
                 raise NotImplementedError()
             else:
                 raise Exception('clone from URI %s not allowed' % (url,))
         elif repo_type == 'git':
             from kallithea.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)/ssh URL')
             'clone_uri': _(u'Invalid repository URL'),
             'invalid_clone_uri': _(u'Invalid repository URL. It must be a '
                                     'valid http, https, ssh, svn+http or svn+https 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 has 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
             # create repositories with write permission on group is set to true
             create_on_write = HasPermissionAny('hg.create.write_on_repogroup.true')()
             group_admin = HasRepoGroupPermissionAny('group.admin')(gr_name,
                                             'can write into group validator')
             group_write = HasRepoGroupPermissionAny('group.write')(gr_name,
                                             'can write into group validator')
             forbidden = not (group_admin or (group_write and create_on_write))
             can_create_repos = HasPermissionAny('hg.admin', 'hg.create.repository')
             gid = (old_data['repo_group'].get('group_id')
                    if (old_data and 'repo_group' in old_data) else None)
             value_changed = gid != safe_int(value)
             new = not old_data
             # do check if we changed the value, there's a case that someone got
             # revoked write permissions to a repository, he still created, we
             # don't need to check permission if he didn't change the value of
             # groups in form box
             if value_changed or new:
                 #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 = HasRepoGroupPermissionAny('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_ == 'repo_group':
         EMPTY_PERM = 'group.none'
     elif type_ == 'repo':
         EMPTY_PERM = 'repository.none'
     elif type_ == 'user_group':
         EMPTY_PERM = 'usergroup.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 == User.DEFAULT_USER:
                         if str2bool(value.get('repo_private')):
                             # set none for default when updating to
                             # private repo protects against 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.UnicodeString):
         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"')
+        }
         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

kallithea/templates/admin/repos/repo_add_base.html

➞

Show inline comments

 ## -*- coding: utf-8 -*-
 ${h.form(url('repos'))}
 <div class="form">
     <!-- fields -->
     <div class="fields">
         <div class="field">
             <div class="label">
                 <label for="repo_name">${_('Name')}:</label>
             </div>
             <div class="input">
                 ${h.text('repo_name',class_="small")}
                 <div style="margin: 6px 0px 0px 0px">
                     <a id="remote_clone_toggle" href="#">
                         <i class="icon-download-cloud"></i>
                         ${_('Import existing repository ?')}
                     </a>
                 </div>
                 %if not c.authuser.is_admin:
                     ${h.hidden('user_created',True)}
                 %endif
             </div>
          </div>
-        <div id="remote_clone" class="field" style="display: none">
         <div id="remote_clone" class="field">
             <div class="label">
                 <label for="clone_uri">${_('Clone from')}:</label>
             </div>
             <div class="input">
                 ${h.text('clone_uri',class_="small")}
                 <span class="help-block">
                     ${_('Optional URL from which repository should be cloned.')}
                 </span>
             </div>
         </div>
         <div class="field">
             <div class="label label-textarea">
                 <label for="repo_description">${_('Description')}:</label>
             </div>
             <div class="textarea-repo editor">
                 ${h.textarea('repo_description')}
                 <span class="help-block">${_('Keep it short and to the point. Use a README file for longer descriptions.')}</span>
             </div>
         </div>
         <div class="field">
              <div class="label">
                  <label for="repo_group">${_('Repository group')}:</label>
              </div>
              <div class="input">
                  ${h.select('repo_group',request.GET.get('parent_group'),c.repo_groups,class_="medium")}
                  <span class="help-block">${_('Optionally select a group to put this repository into.')}</span>
              </div>
         </div>
         <div id="copy_perms" class="field">
             <div class="label label-checkbox">
                 <label for="repo_copy_permissions">${_('Copy parent group permissions')}:</label>
             </div>
             <div class="checkboxes">
                 ${h.checkbox('repo_copy_permissions',value="True")}
                 <span class="help-block">${_('Copy permission set from parent repository group.')}</span>
             </div>
         </div>
         <div class="field">
             <div class="label">
                 <label for="repo_type">${_('Type')}:</label>
             </div>
             <div class="input">
                 ${h.select('repo_type','hg',c.backends,class_="small")}
                 <span class="help-block">${_('Type of repository to create.')}</span>
             </div>
         </div>
         <div class="field">
             <div class="label">
                 <label for="repo_landing_rev">${_('Landing revision')}:</label>
             </div>
             <div class="input">
                 ${h.select('repo_landing_rev','',c.landing_revs,class_="medium")}
                 <span class="help-block">${_('Default revision for files page, downloads, full text search index and readme generation')}</span>
             </div>
         </div>
         <div class="field">
             <div class="label label-checkbox">
                 <label for="repo_private">${_('Private repository')}:</label>
             </div>
             <div class="checkboxes">
                 ${h.checkbox('repo_private',value="True")}
                 <span class="help-block">${_('Private repositories are only visible to people explicitly added as collaborators.')}</span>
             </div>
         </div>
         <div class="buttons">
           ${h.submit('add',_('Add'),class_="btn")}
         </div>
     </div>
 </div>
 <script>
     $(document).ready(function(){
         var setCopyPermsOption = function(group_val){
             if(group_val != "-1"){
                 $('#copy_perms').show();
+            }
             else{
                 $('#copy_perms').hide();
+            }
+        }
         $('#remote_clone_toggle').on('click', function(e){
             $('#remote_clone').show();
             e.preventDefault();
         });
         if($('#remote_clone input').hasClass('error')){
             $('#remote_clone').show();
+        }
         if($('#remote_clone input').val()){
             $('#remote_clone').show();
+        }
         $("#repo_group").select2({
             'dropdownAutoWidth': true
         });
         setCopyPermsOption($('#repo_group').val());
         $("#repo_group").on("change", function(e) {
             setCopyPermsOption(e.val);
         });
         $("#repo_type").select2({
             'minimumResultsForSearch': -1
         });
         $("#repo_landing_rev").select2({
             'minimumResultsForSearch': -1
         });
         $('#repo_name').focus();
     });
 </script>
 ${h.end_form()}

kallithea/templates/admin/repos/repo_edit_settings.html

➞

Show inline comments

 ${h.form(url('repo', repo_name=c.repo_info.repo_name),method='put')}
     <div class="form">
         <!-- fields -->
         <div class="fields">
             <div class="field">
                 <div class="label">
                     <label for="repo_name">${_('Name')}:</label>
                 </div>
                 <div class="input">
                     ${h.text('repo_name',class_="medium")}
                     <span class="help-block">${_('Permanent Repository ID')}: `_${c.repo_info.repo_id}` <span><a id="show_more_clone_id" href="#">${_('What is that?')}</a></span></span>
                     <span id="clone_id" class="help-block" style="display: none">
                         ${_('URL by id')}: `${c.repo_info.clone_url(with_id=True)}` </br>
                         ${_('''In case this repository is renamed or moved into another group the repository URL changes.
                                Using the above permanent URL guarantees that this repository always will be accessible on that URL.
                                This is useful for CI systems, or any other cases that you need to hardcode the URL into a 3rd party service.''')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label">
                     <label for="clone_uri">${_('Clone URL')}:</label>
                 </div>
                 <div class="input">
                   %if c.repo_info.clone_uri:
                     <div id="clone_uri_hidden" style="font-size: 14px">
                         <span id="clone_uri_hidden_value">${c.repo_info.clone_uri_hidden}</span>
                         <span style="cursor: pointer; padding: 0px 0px 5px 0px" id="edit_clone_uri"><i class="icon-edit"></i>${_('Edit')}</span>
                     </div>
                     <div id="alter_clone_uri" style="display: none">
-                        ${h.text('clone_uri',class_="medium",  placeholder=_('new value'))}
+                        ${h.text('clone_uri',class_="medium", placeholder=_('New URL'))}
                     </div>
                   %else:
                     ## not set yet, display form to set it
                     ${h.text('clone_uri',class_="medium")}
                     ${h.hidden('clone_uri_change', 'NEW')}
                   %endif
                   <span id="alter_clone_uri_help_block" class="help-block">
                       ${_('URL used for doing remote pulls.')}
                   </span>
                 </div>
             </div>
             <div class="field">
                 <div class="label">
                     <label for="repo_group">${_('Repository group')}:</label>
                 </div>
                 <div class="input">
                     ${h.select('repo_group','',c.repo_groups,class_="medium")}
                     <span class="help-block">${_('Optionally select a group to put this repository into.')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label">
                     <label for="repo_landing_rev">${_('Landing revision')}:</label>
                 </div>
                 <div class="input">
                     ${h.select('repo_landing_rev','',c.landing_revs,class_="medium")}
                     <span class="help-block">${_('Default revision for files page, downloads, whoosh and readme')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label">
                     <label for="user">${_('Owner')}:</label>
                 </div>
                 <div class="input input-medium ac">
                     <div class="perm_ac">
                        ${h.text('user',class_='yui-ac-input')}
                        <span class="help-block">${_('Change owner of this repository.')}</span>
                        <div id="owner_container"></div>
                     </div>
                 </div>
              </div>
             <div class="field">
                 <div class="label label-textarea">
                     <label for="repo_description">${_('Description')}:</label>
                 </div>
                 <div class="textarea text-area editor">
                     ${h.textarea('repo_description', style="height:165px")}
                     <span class="help-block">${_('Keep it short and to the point. Use a README file for longer descriptions.')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label label-checkbox">
                     <label for="repo_private">${_('Private repository')}:</label>
                 </div>
                 <div class="checkboxes">
                     ${h.checkbox('repo_private',value="True")}
                     <span class="help-block">${_('Private repositories are only visible to people explicitly added as collaborators.')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label label-checkbox">
                     <label for="repo_enable_statistics">${_('Enable statistics')}:</label>
                 </div>
                 <div class="checkboxes">
                     ${h.checkbox('repo_enable_statistics',value="True")}
                     <span class="help-block">${_('Enable statistics window on summary page.')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label label-checkbox">
                     <label for="repo_enable_downloads">${_('Enable downloads')}:</label>
                 </div>
                 <div class="checkboxes">
                     ${h.checkbox('repo_enable_downloads',value="True")}
                     <span class="help-block">${_('Enable download menu on summary page.')}</span>
                 </div>
             </div>
             <div class="field">
                 <div class="label label-checkbox">
                     <label for="repo_enable_locking">${_('Enable locking')}:</label>
                 </div>
                 <div class="checkboxes">
                     ${h.checkbox('repo_enable_locking',value="True")}
                     <span class="help-block">${_('Enable lock-by-pulling on repository.')}</span>
                 </div>
             </div>
             %if c.visual.repository_fields:
               ## EXTRA FIELDS
               %for field in c.repo_fields:
                 <div class="field">
                     <div class="label">
                         <label for="${field.field_key_prefixed}">${field.field_label} (${field.field_key}):</label>
                     </div>
                     <div class="input input-medium">
                         ${h.text(field.field_key_prefixed, field.field_value, class_='medium')}
                         %if field.field_desc:
                           <span class="help-block">${field.field_desc}</span>
                         %endif
                     </div>
                  </div>
               %endfor
             %endif
             <div class="buttons">
               ${h.submit('save',_('Save'),class_="btn")}
               ${h.reset('reset',_('Reset'),class_="btn")}
             </div>
         </div>
     </div>
     ${h.end_form()}
 <script>
     $(document).ready(function(){
         $('#show_more_clone_id').on('click', function(e){
             $('#clone_id').show();
             e.preventDefault();
         });
         $('#edit_clone_uri').on('click', function(e){
           $('#alter_clone_uri').show();
           $('#edit_clone_uri').hide();
           $('#clone_uri_hidden').hide();
           ## store hash of old value for change detection
-          var uri_change =  '<input id="clone_uri_change" name="clone_uri_change" type="hidden" value="${h.md5(c.repo_info.clone_uri or "").hexdigest()}" />';
           var uri_change = '<input id="clone_uri_change" name="clone_uri_change" type="hidden" value="${h.md5(c.repo_info.clone_uri or "").hexdigest()}" />';
           $('#alter_clone_uri_help_block').html($('#alter_clone_uri_help_block').html()+" ("+$('#clone_uri_hidden_value').html()+")");
         });
         $('#repo_landing_rev').select2({
             'dropdownAutoWidth': true
         });
         $('#repo_group').select2({
             'dropdownAutoWidth': true
         });
     });
 </script>

kallithea/tests/functional/test_admin_repos.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 import os
 import mock
 import urllib
 from kallithea.lib import vcs
 from kallithea.model.db import Repository, RepoGroup, UserRepoToPerm, User,\
     Permission
 from kallithea.model.user import UserModel
 from kallithea.tests import *
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.model.repo import RepoModel
 from kallithea.model.meta import Session
 from kallithea.tests.fixture import Fixture, error_function
 fixture = Fixture()
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query()\
                 .filter(UserRepoToPerm.repository ==
                         Repository.get_by_repo_name(repo))\
                 .filter(UserRepoToPerm.user == User.get_by_username(user))\
                 .all()
     return perm
 class _BaseTest(object):
     """
     Write all tests here
     """
     REPO = None
     REPO_TYPE = None
     NEW_REPO = None
     OTHER_TYPE_REPO = None
     OTHER_TYPE = None
     @classmethod
     def setup_class(cls):
         pass
     @classmethod
     def teardown_class(cls):
         pass
     def test_index(self):
         self.log_user()
         response = self.app.get(url('repos'))
     def test_create(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.assertEqual(response.json, {u'result': True})
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name).one()
         self.assertEqual(new_repo.repo_name, repo_name)
         self.assertEqual(new_repo.description, description)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name)
         Session().commit()
     def test_create_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (self.NEW_REPO, non_ascii)
         repo_name_unicode = repo_name.decode('utf8')
         description = 'description for newly created repo' + non_ascii
         description_unicode = description.decode('utf8')
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.assertEqual(response.json, {u'result': True})
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
                                % (urllib.quote(repo_name), repo_name_unicode))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_unicode).one()
         self.assertEqual(new_repo.repo_name, repo_name_unicode)
         self.assertEqual(new_repo.description, description_unicode)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
     def test_create_in_group(self):
         self.log_user()
         ## create GROUP
         group_name = 'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         self.assertEqual(response.json, {u'result': True})
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query()\
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         self.assertEqual(len(inherited_perms), 1)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name_full))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             self.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_in_group_without_needed_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # avoid spurious RepoGroup DetachedInstanceError ...
         authentication_token = self.authentication_token()
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         ## create GROUP
         group_name = 'reg_sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
         group_name_allowed = 'reg_sometest_allowed_%s' % self.REPO_TYPE
         gr_allowed = RepoGroupModel().create(group_name=group_name_allowed,
                                      group_description='test',
                                      owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 _authentication_token=authentication_token))
         response.mustcontain('Invalid value')
         # user is allowed to create in this group
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join([group_name_allowed, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr_allowed.group_id,
                                                 _authentication_token=authentication_token))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         self.assertEqual(response.json, {u'result': True})
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         inherited_perms = UserRepoToPerm.query()\
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         self.assertEqual(len(inherited_perms), 1)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name_full))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             self.fail('no repo %s in filesystem' % repo_name)
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         RepoGroupModel().delete(group_name_allowed)
         Session().commit()
     def test_create_in_group_inherit_permissions(self):
         self.log_user()
         ## create GROUP
         group_name = 'sometest_%s' % self.REPO_TYPE
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(gr, TEST_USER_REGULAR_LOGIN, perm)
         ## add repo permissions
         Session().commit()
         repo_name = 'ingroup_inherited_%s' % self.REPO_TYPE
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 repo_group=gr.group_id,
                                                 repo_copy_permissions=True,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name_full))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name_full, repo_name_full))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_full).one()
         new_repo_id = new_repo.repo_id
         self.assertEqual(new_repo.repo_name, repo_name_full)
         self.assertEqual(new_repo.description, description)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name_full))
         response.mustcontain(repo_name_full)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name_full))
         except vcs.exceptions.VCSError:
             RepoGroupModel().delete(group_name)
             Session().commit()
             self.fail('no repo %s in filesystem' % repo_name)
         #check if inherited permissiona are applied
         inherited_perms = UserRepoToPerm.query()\
             .filter(UserRepoToPerm.repository_id == new_repo_id).all()
         self.assertEqual(len(inherited_perms), 2)
         self.assertTrue(TEST_USER_REGULAR_LOGIN in [x.user.username
                                                     for x in inherited_perms])
         self.assertTrue('repository.write' in [x.permission.permission_name
                                                for x in inherited_perms])
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
     def test_create_remote_repo_wrong_clone_uri(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
-        response.mustcontain('invalid clone URL')
+        response.mustcontain('Invalid repository URL')
     def test_create_remote_repo_wrong_clone_uri_hg_svn(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 clone_uri='svn+http://127.0.0.1/repo',
                                                 _authentication_token=self.authentication_token()))
-        response.mustcontain('invalid clone URL')
+        response.mustcontain('Invalid repository URL')
     def test_delete(self):
         self.log_user()
         repo_name = 'vcs_test_new_to_delete_%s' % self.REPO_TYPE
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_name=repo_name,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.checkSessionFlash(response,
                                'Created repository <a href="/%s">%s</a>'
                                % (repo_name, repo_name))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name).one()
         self.assertEqual(new_repo.repo_name, repo_name)
         self.assertEqual(new_repo.description, description)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('repo', repo_name=repo_name))
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name).scalar()
         self.assertEqual(deleted_repo, None)
         self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),
                                   False)
     def test_delete_non_ascii(self):
         self.log_user()
         non_ascii = "ąęł"
         repo_name = "%s%s" % (self.NEW_REPO, non_ascii)
         repo_name_unicode = repo_name.decode('utf8')
         description = 'description for newly created repo' + non_ascii
         description_unicode = description.decode('utf8')
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         ## run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         self.assertEqual(response.json, {u'result': True})
         self.checkSessionFlash(response,
                                u'Created repository <a href="/%s">%s</a>'
                                % (urllib.quote(repo_name), repo_name_unicode))
         # test if the repo was created in the database
         new_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_unicode).one()
         self.assertEqual(new_repo.repo_name, repo_name_unicode)
         self.assertEqual(new_repo.description, description_unicode)
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(self.REPO_TYPE)
         # test if the repository was created on filesystem
         try:
             vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
         except vcs.exceptions.VCSError:
             self.fail('no repo %s in filesystem' % repo_name)
         response = self.app.delete(url('repo', repo_name=repo_name))
         self.checkSessionFlash(response, 'Deleted repository %s' % (repo_name_unicode))
         response.follow()
         #check if repo was deleted from db
         deleted_repo = Session().query(Repository)\
             .filter(Repository.repo_name == repo_name_unicode).scalar()
         self.assertEqual(deleted_repo, None)
         self.assertEqual(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)),
                                   False)
     def test_delete_repo_with_group(self):
         #TODO:
         pass
     def test_delete_browser_fakeout(self):
         response = self.app.post(url('repo', repo_name=self.REPO),
                                  params=dict(_method='delete', _authentication_token=self.authentication_token()))
     def test_show(self):
         self.log_user()
         response = self.app.get(url('repo', repo_name=self.REPO))
     def test_edit(self):
         response = self.app.get(url('edit_repo', repo_name=self.REPO))
     def test_set_private_flag_sets_default_to_none(self):
         self.log_user()
         #initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.read')
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         response = self.app.put(url('repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=1,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, True)
         #now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         response = self.app.put(url('repo', repo_name=self.REPO),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=self.REPO,
                                                 repo_type=self.REPO_TYPE,
                                                 user=TEST_USER_ADMIN_LOGIN,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                msg='Repository %s updated successfully' % (self.REPO))
         self.assertEqual(Repository.get_by_repo_name(self.REPO).private, False)
         #we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=self.REPO)
         self.assertTrue(len(perm), 1)
         self.assertEqual(perm[0].permission.permission_name, 'repository.none')
         #update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().add(perm[0])
         Session().commit()
     def test_set_repo_fork_has_no_self_id(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.get(url('edit_repo_advanced', repo_name=self.REPO))
         opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
         response.mustcontain(no=[opt])
     def test_set_fork_of_other_repo(self):
         self.log_user()
         other_repo = 'other_%s' % self.REPO_TYPE
         fixture.create_repo(other_repo, repo_type=self.REPO_TYPE)
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(other_repo)
         self.checkSessionFlash(response,
             'Marked repo %s as fork of %s' % (repo.repo_name, repo2.repo_name))
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
         opt = """<option value="%s" selected="selected">%s</option>""" % (
                     repo2.repo_id, repo2.repo_name)
         response.mustcontain(opt)
         fixture.destroy_repo(other_repo, forks='detach')
     def test_set_fork_of_other_type_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo2.repo_id, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
             'Cannot set repository as fork of repository with other type')
     def test_set_fork_of_none(self):
         self.log_user()
         ## mark it as None
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=None, _authentication_token=self.authentication_token()))
         repo = Repository.get_by_repo_name(self.REPO)
         repo2 = Repository.get_by_repo_name(self.OTHER_TYPE_REPO)
         self.checkSessionFlash(response,
                                'Marked repo %s as fork of %s'
                                % (repo.repo_name, "Nothing"))
         assert repo.fork is None
     def test_set_fork_of_same_repo(self):
         self.log_user()
         repo = Repository.get_by_repo_name(self.REPO)
         response = self.app.put(url('edit_repo_advanced_fork', repo_name=self.REPO),
                                 params=dict(id_fork_of=repo.repo_id, _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'An error occurred during this operation')
     def test_create_on_top_level_without_permissions(self):
         usr = self.log_user(TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
         user = User.get(usr['user_id'])
         repo_name = self.NEW_REPO+'no_perms'
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         response.mustcontain('no permission to create repository in root location')
         RepoModel().delete(repo_name)
         Session().commit()
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(self):
         self.log_user()
         repo_name = self.NEW_REPO
         description = 'description for newly created repo'
         response = self.app.post(url('repos'),
                         fixture._get_repo_create_params(repo_private=False,
                                                 repo_name=repo_name,
                                                 repo_type=self.REPO_TYPE,
                                                 repo_description=description,
                                                 _authentication_token=self.authentication_token()))
         self.checkSessionFlash(response,
                                'Error creating repository %s' % repo_name)
         # repo must not be in db
         repo = Repository.get_by_repo_name(repo_name)
         self.assertEqual(repo, None)
         # repo must not be in filesystem !
         self.assertFalse(os.path.isdir(os.path.join(TESTS_TMP_PATH, repo_name)))
 class TestAdminReposControllerGIT(TestController, _BaseTest):
     REPO = GIT_REPO
     REPO_TYPE = 'git'
     NEW_REPO = NEW_GIT_REPO
     OTHER_TYPE_REPO = HG_REPO
     OTHER_TYPE = 'hg'
 class TestAdminReposControllerHG(TestController, _BaseTest):
     REPO = HG_REPO
     REPO_TYPE = 'hg'
     NEW_REPO = NEW_HG_REPO
     OTHER_TYPE_REPO = GIT_REPO
     OTHER_TYPE = 'git'

0 comments (0 inline, 0 general)