kallithea Changeset - 401fe08bc6b8

Changeset - 401fe08bc6b8

Parent rev.

Child rev.

[Not reviewed]

default

0 5 0

Thomas De Schampheleire - 7 years ago 2018-06-09 21:24:45
thomas.de_schampheleire@nokia.com

utils: move repo_name_slug to utils2 to prevent import cycle on setup_db

After commit 57a733313e4f, 'gearbox setup-db -c my.ini' fails with an import
cycle as follows:

Traceback (most recent call last):
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/bin/gearbox", line 11, in <module>
sys.exit(main())
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/lib/python2.7/site-packages/gearbox/main.py", line 199, in main
return gearbox.run(args)
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/lib/python2.7/site-packages/gearbox/main.py", line 145, in run
return self._run_subcommand(remainder)
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/lib/python2.7/site-packages/gearbox/main.py", line 149, in _run_subcommand
subcommand = self.command_manager.find_command(argv)
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/lib/python2.7/site-packages/gearbox/commandmanager.py", line 78, in find_command
cmd_factory = cmd_ep.resolve()
File "/home/tdescham/repo/contrib/kallithea/venv/kallithea-release/lib/python2.7/site-packages/pkg_resources/__init__.py", line 2324, in resolve
module = __import__(self.module_name, fromlist=['__name__'], level=0)
File "/home/tdescham/repo/contrib/kallithea/kallithea-release/kallithea/lib/paster_commands/setup_db.py", line 27, in <module>
from kallithea.lib.db_manage import DbManage
File "/home/tdescham/repo/contrib/kallithea/kallithea-release/kallithea/lib/db_manage.py", line 47, in <module>
from kallithea.model.repo_group import RepoGroupModel
File "/home/tdescham/repo/contrib/kallithea/kallithea-release/kallithea/model/repo_group.py", line 35, in <module>
import kallithea.lib.utils
File "/home/tdescham/repo/contrib/kallithea/kallithea-release/kallithea/lib/utils.py", line 48, in <module>
from kallithea.model.repo_group import RepoGroupModel
ImportError: cannot import name RepoGroupModel

i.e. kallithea.model.repo_group wants to import kallithea.lib.utils which
in turn wants to import kallithea.model.repo_group.

In fact there exists kallithea.lib.utils and kallithea.lib.utils2.
The current split is that 'utils2' contains 'simple' utilities, none of which
depend on kallithea models, controllers, ... In contrast, 'utils' does rely
on such kallithea classes.

As kallithea.model.repo_group was only include kallithea.lib.utils for its
repo_name_slug method, which has no dependency on other kallithea classes,
move that method (and its dependent recursive_replace) to
kallithea.lib.utils2 instead. This fixes the import cycle.

5 files changed with 48 insertions and 46 deletions:

kallithea/lib/utils.py

kallithea/lib/utils2.py

kallithea/model/repo.py

kallithea/model/repo_group.py

kallithea/model/validators.py

0 comments (0 inline, 0 general)

kallithea/lib/utils.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.utils
 ~~~~~~~~~~~~~~~~~~~
 Utilities library for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Apr 18, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import re
 import logging
 import datetime
 import traceback
 import beaker
 from tg import request, response
 from tg.i18n import ugettext as _
 from webhelpers.text import collapse, remove_formatting, strip_tags
 from beaker.cache import _cache_decorate
 from kallithea.lib.vcs.utils.hgcompat import ui, config
 from kallithea.lib.vcs.utils.helpers import get_scm
 from kallithea.lib.vcs.exceptions import VCSError
 from kallithea.lib.exceptions import HgsubversionImportError
 from kallithea.model import meta
 from kallithea.model.db import Repository, User, Ui, \
     UserLog, RepoGroup, Setting, UserGroup
 from kallithea.model.repo_group import RepoGroupModel
 from kallithea.lib.utils2 import safe_str, safe_unicode, get_current_authuser
 from kallithea.lib.vcs.utils.fakemod import create_module
 log = logging.getLogger(__name__)
 REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')
 def recursive_replace(str_, replace=' '):
     """
     Recursive replace of given sign to just one instance
     :param str_: given string
     :param replace: char to find and replace multiple instances
     Examples::
     >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')
     'Mighty-Mighty-Bo-sstones'
     """
     if str_.find(replace * 2) == -1:
         return str_
     else:
         str_ = str_.replace(replace * 2, replace)
         return recursive_replace(str_, replace)
 def repo_name_slug(value):
     """
     Return slug of name of repository
     This function is called on each creation/modification
     of repository to prevent bad names in repo
     """
     slug = remove_formatting(value)
     slug = strip_tags(slug)
     for c in """`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:
         slug = slug.replace(c, '-')
     slug = recursive_replace(slug, '-')
     slug = collapse(slug, '-')
     return slug
 #==============================================================================
 # PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS
 #==============================================================================
 def get_repo_slug(request):
     _repo = request.environ['pylons.routes_dict'].get('repo_name')
     if _repo:
         _repo = _repo.rstrip('/')
     return _repo
 def get_repo_group_slug(request):
     _group = request.environ['pylons.routes_dict'].get('group_name')
     if _group:
         _group = _group.rstrip('/')
     return _group
 def get_user_group_slug(request):
     _group = request.environ['pylons.routes_dict'].get('id')
     _group = UserGroup.get(_group)
     if _group:
         return _group.users_group_name
     return None
 def _extract_id_from_repo_name(repo_name):
     if repo_name.startswith('/'):
         repo_name = repo_name.lstrip('/')
     by_id_match = re.match(r'^_(\d{1,})', repo_name)
     if by_id_match:
         return by_id_match.groups()[0]
 def get_repo_by_id(repo_name):
     """
     Extracts repo_name by id from special urls. Example url is _11/repo_name
     :param repo_name:
     :return: repo_name if matched else None
     """
     _repo_id = _extract_id_from_repo_name(repo_name)
     if _repo_id:
         from kallithea.model.db import Repository
         repo = Repository.get(_repo_id)
         if repo:
             # TODO: return repo instead of reponame? or would that be a layering violation?
             return repo.repo_name
     return None
 def action_logger(user, action, repo, ipaddr='', commit=False):
     """
     Action logger for various actions made by users
     :param user: user that made this action, can be a unique username string or
         object containing user_id attribute
     :param action: action to log, should be on of predefined unique actions for
         easy translations
     :param repo: string name of repository or object containing repo_id,
         that action was made on
     :param ipaddr: optional IP address from what the action was made
     """
     # if we don't get explicit IP address try to get one from registered user
     # in tmpl context var
     if not ipaddr:
         ipaddr = getattr(get_current_authuser(), 'ip_addr', '')
     if getattr(user, 'user_id', None):
         user_obj = User.get(user.user_id)
     elif isinstance(user, basestring):
         user_obj = User.get_by_username(user)
     else:
         raise Exception('You have to provide a user object or a username')
     if getattr(repo, 'repo_id', None):
         repo_obj = Repository.get(repo.repo_id)
         repo_name = repo_obj.repo_name
     elif isinstance(repo, basestring):
         repo_name = repo.lstrip('/')
         repo_obj = Repository.get_by_repo_name(repo_name)
     else:
         repo_obj = None
         repo_name = u''
     user_log = UserLog()
     user_log.user_id = user_obj.user_id
     user_log.username = user_obj.username
     user_log.action = safe_unicode(action)
     user_log.repository = repo_obj
     user_log.repository_name = repo_name
     user_log.action_date = datetime.datetime.now()
     user_log.user_ip = ipaddr
     meta.Session().add(user_log)
     log.info('Logging action:%s on %s by user:%s ip:%s',
              action, safe_unicode(repo), user_obj, ipaddr)
     if commit:
         meta.Session().commit()
 def get_filesystem_repos(path):
     """
     Scans given path for repos and return (name,(type,path)) tuple
     :param path: path to scan for repositories
     :param recursive: recursive search and return names with subdirs in front
     """
     # remove ending slash for better results
     path = safe_str(path.rstrip(os.sep))
     log.debug('now scanning in %s', path)
     def isdir(*n):
         return os.path.isdir(os.path.join(*n))
     for root, dirs, _files in os.walk(path):
         recurse_dirs = []
         for subdir in dirs:
             # skip removed repos
             if REMOVED_REPO_PAT.match(subdir):
                 continue
             # skip .<something> dirs TODO: rly? then we should prevent creating them ...
             if subdir.startswith('.'):
                 continue
             cur_path = os.path.join(root, subdir)
             if isdir(cur_path, '.git'):
                 log.warning('ignoring non-bare Git repo: %s', cur_path)
                 continue
             if (isdir(cur_path, '.hg') or
                 isdir(cur_path, '.svn') or
                 isdir(cur_path, 'objects') and (isdir(cur_path, 'refs') or
                                                 os.path.isfile(os.path.join(cur_path, 'packed-refs')))):
                 if not os.access(cur_path, os.R_OK) or not os.access(cur_path, os.X_OK):
                     log.warning('ignoring repo path without access: %s', cur_path)
                     continue
                 if not os.access(cur_path, os.W_OK):
                     log.warning('repo path without write access: %s', cur_path)
                 try:
                     scm_info = get_scm(cur_path)
                     assert cur_path.startswith(path)
                     repo_path = cur_path[len(path) + 1:]
                     yield repo_path, scm_info
                     continue # no recursion
                 except VCSError:
                     # We should perhaps ignore such broken repos, but especially
                     # the bare git detection is unreliable so we dive into it
                     pass
             recurse_dirs.append(subdir)
         dirs[:] = recurse_dirs
 def is_valid_repo_uri(repo_type, url, ui):
     """Check if the url seems like a valid remote repo location - raise an Exception if any problems"""
     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'):
             try:
                 from hgsubversion.svnrepo import svnremoterepo
             except ImportError:
                 raise HgsubversionImportError(_('Unable to activate hgsubversion support. '
                                                 'The "hgsubversion" library is missing'))
             svnremoterepo(ui, url).svn.uuid
         elif url.startswith('git+http'):
             raise NotImplementedError()
         else:
             raise Exception('URI %s not allowed' % (url,))
     elif repo_type == 'git':
         from kallithea.lib.vcs.backends.git.repository import GitRepository
         if url.startswith('http') or url.startswith('git'):
             # 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'):

kallithea/lib/utils2.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.lib.utils2
 ~~~~~~~~~~~~~~~~~~~~
 Some simple helper functions
 Some simple helper functions.
 Note: all these functions should be independent of Kallithea classes, i.e.
 models, controllers, etc.  to prevent import cycles.
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 5, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import re
 import sys
 import time
 import uuid
 import datetime
 import urllib
 import binascii
 import webob
 import urlobject
 from webhelpers.text import collapse, remove_formatting, strip_tags
 from tg.i18n import ugettext as _, ungettext
 from kallithea.lib.vcs.utils.lazy import LazyProperty
 from kallithea.lib.compat import json
 def str2bool(_str):
     """
     returns True/False value from given string, it tries to translate the
     string into boolean
     :param _str: string value to translate into boolean
     :rtype: boolean
     :returns: boolean from given string
     """
     if _str is None:
         return False
     if _str in (True, False):
         return _str
     _str = str(_str).strip().lower()
     return _str in ('t', 'true', 'y', 'yes', 'on', '1')
 def aslist(obj, sep=None, strip=True):
     """
     Returns given string separated by sep as list
     :param obj:
     :param sep:
     :param strip:
     """
     if isinstance(obj, (basestring)):
         lst = obj.split(sep)
         if strip:
             lst = [v.strip() for v in lst]
         return lst
     elif isinstance(obj, (list, tuple)):
         return obj
     elif obj is None:
         return []
     else:
         return [obj]
 def convert_line_endings(line, mode):
     """
     Converts a given line  "line end" according to given mode
     Available modes are::
 - Unix
 - Mac
 - DOS
     :param line: given line to convert
     :param mode: mode to convert to
     :rtype: str
     :return: converted line according to mode
     """
     from string import replace
     if mode == 0:
             line = replace(line, '\r\n', '\n')
             line = replace(line, '\r', '\n')
     elif mode == 1:
             line = replace(line, '\r\n', '\r')
             line = replace(line, '\n', '\r')
     elif mode == 2:
             line = re.sub("\r(?!\n)|(?<!\r)\n", "\r\n", line)
     return line
 def detect_mode(line, default):
     """
     Detects line break for given line, if line break couldn't be found
     given default value is returned
     :param line: str line
     :param default: default
     :rtype: int
     :return: value of line end on of 0 - Unix, 1 - Mac, 2 - DOS
     """
     if line.endswith('\r\n'):
         return 2
     elif line.endswith('\n'):
         return 0
     elif line.endswith('\r'):
         return 1
     else:
         return default
 def generate_api_key():
     """
     Generates a random (presumably unique) API key.
     This value is used in URLs and "Bearer" HTTP Authorization headers,
     which in practice means it should only contain URL-safe characters
     (RFC 3986):
         unreserved = ALPHA / DIGIT / "-" / "." / "_" / "~"
     """
     # Hexadecimal certainly qualifies as URL-safe.
     return binascii.hexlify(os.urandom(20))
 def safe_int(val, default=None):
     """
     Returns int() of val if val is not convertable to int use default
     instead
     :param val:
     :param default:
     """
     try:
         val = int(val)
     except (ValueError, TypeError):
         val = default
     return val
 def safe_unicode(str_, from_encoding=None):
     """
     safe unicode function. Does few trick to turn str_ into unicode
     In case of UnicodeDecode error we try to return it with encoding detected
     by chardet library if it fails fallback to unicode with errors replaced
     :param str_: string to decode
     :rtype: unicode
     :returns: unicode object
     """
     if isinstance(str_, unicode):
         return str_
     if not from_encoding:
         import kallithea
         DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
                                                         'utf-8'), sep=',')
         from_encoding = DEFAULT_ENCODINGS
     if not isinstance(from_encoding, (list, tuple)):
         from_encoding = [from_encoding]
     try:
         return unicode(str_)
     except UnicodeDecodeError:
         pass
     for enc in from_encoding:
         try:
             return unicode(str_, enc)
         except UnicodeDecodeError:
             pass
     try:
         import chardet
         encoding = chardet.detect(str_)['encoding']
         if encoding is None:
             raise Exception()
         return str_.decode(encoding)
     except (ImportError, UnicodeDecodeError, Exception):
         return unicode(str_, from_encoding[0], 'replace')
 def safe_str(unicode_, to_encoding=None):
     """
     safe str function. Does few trick to turn unicode_ into string
     In case of UnicodeEncodeError we try to return it with encoding detected
     by chardet library if it fails fallback to string with errors replaced
     :param unicode_: unicode to encode
     :rtype: str
     :returns: str object
     """
     # if it's not basestr cast to str
     if not isinstance(unicode_, basestring):
         return str(unicode_)
     if isinstance(unicode_, str):
         return unicode_
     if not to_encoding:
         import kallithea
         DEFAULT_ENCODINGS = aslist(kallithea.CONFIG.get('default_encoding',
                                                         'utf-8'), sep=',')
         to_encoding = DEFAULT_ENCODINGS
     if not isinstance(to_encoding, (list, tuple)):
@@ @@ -457,192 +460,228 @@ def get_changeset_safe(repo, rev): @@
 def datetime_to_time(dt):
     if dt:
         return time.mktime(dt.timetuple())
 def time_to_datetime(tm):
     if tm:
         if isinstance(tm, basestring):
             try:
                 tm = float(tm)
             except ValueError:
                 return
         return datetime.datetime.fromtimestamp(tm)
 # Must match regexp in kallithea/public/js/base.js MentionsAutoComplete()
 # Check char before @ - it must not look like we are in an email addresses.
 # Matching is greedy so we don't have to look beyond the end.
 MENTIONS_REGEX = re.compile(r'(?:^|(?<=[^a-zA-Z0-9]))@([a-zA-Z0-9][-_.a-zA-Z0-9]*[a-zA-Z0-9])')
 def extract_mentioned_usernames(text):
     r"""
     Returns list of (possible) usernames @mentioned in given text.
     >>> extract_mentioned_usernames('@1-2.a_X,@1234 not@not @ddd@not @n @ee @ff @gg, @gg;@hh @n\n@zz,')
     ['1-2.a_X', '1234', 'ddd', 'ee', 'ff', 'gg', 'hh', 'zz']
     """
     return MENTIONS_REGEX.findall(text)
 def extract_mentioned_users(text):
     """ Returns set of actual database Users @mentioned in given text. """
     from kallithea.model.db import User
     result = set()
     for name in extract_mentioned_usernames(text):
         user = User.get_by_username(name, case_insensitive=True)
         if user is not None and not user.is_default_user:
             result.add(user)
     return result
 class AttributeDict(dict):
     def __getattr__(self, attr):
         return self.get(attr, None)
     __setattr__ = dict.__setitem__
     __delattr__ = dict.__delitem__
 def fix_PATH(os_=None):
     """
     Get current active python path, and append it to PATH variable to fix issues
     of subprocess calls and different python versions
     """
     if os_ is None:
         import os
     else:
         os = os_
     cur_path = os.path.split(sys.executable)[0]
     if not os.environ['PATH'].startswith(cur_path):
         os.environ['PATH'] = '%s:%s' % (cur_path, os.environ['PATH'])
 def obfuscate_url_pw(engine):
     from sqlalchemy.engine import url as sa_url
     from sqlalchemy.exc import ArgumentError
     try:
         _url = sa_url.make_url(engine or '')
     except ArgumentError:
         return engine
     if _url.password:
         _url.password = 'XXXXX'
     return str(_url)
 def get_server_url(environ):
     req = webob.Request(environ)
     return req.host_url + req.script_name
 def _extract_extras(env=None):
     """
     Extracts the Kallithea extras data from os.environ, and wraps it into named
     AttributeDict object
     """
     if not env:
         env = os.environ
     try:
         extras = json.loads(env['KALLITHEA_EXTRAS'])
     except KeyError:
         extras = {}
     try:
         for k in ['username', 'repository', 'locked_by', 'scm', 'make_lock',
                   'action', 'ip']:
             extras[k]
     except KeyError as e:
         raise Exception('Missing key %s in os.environ %s' % (e, extras))
     return AttributeDict(extras)
 def _set_extras(extras):
     # RC_SCM_DATA can probably be removed in the future, but for compatibility now...
     os.environ['KALLITHEA_EXTRAS'] = os.environ['RC_SCM_DATA'] = json.dumps(extras)
 def get_current_authuser():
     """
     Gets kallithea user from threadlocal tmpl_context variable if it's
     defined, else returns None.
     """
     from tg import tmpl_context
     if hasattr(tmpl_context, 'authuser'):
         return tmpl_context.authuser
     return None
 class OptionalAttr(object):
     """
     Special Optional Option that defines other attribute. Example::
         def test(apiuser, userid=Optional(OAttr('apiuser')):
             user = Optional.extract(userid)
             # calls
     """
     def __init__(self, attr_name):
         self.attr_name = attr_name
     def __repr__(self):
         return '<OptionalAttr:%s>' % self.attr_name
     def __call__(self):
         return self
 # alias
 OAttr = OptionalAttr
 class Optional(object):
     """
     Defines an optional parameter::
         param = param.getval() if isinstance(param, Optional) else param
         param = param() if isinstance(param, Optional) else param
     is equivalent of::
         param = Optional.extract(param)
     """
     def __init__(self, type_):
         self.type_ = type_
     def __repr__(self):
         return '<Optional:%s>' % self.type_.__repr__()
     def __call__(self):
         return self.getval()
     def getval(self):
         """
         returns value from this Optional instance
         """
         if isinstance(self.type_, OAttr):
             # use params name
             return self.type_.attr_name
         return self.type_
     @classmethod
     def extract(cls, val):
         """
         Extracts value from Optional() instance
         :param val:
         :return: original value if it's not Optional instance else
             value of instance
         """
         if isinstance(val, cls):
             return val.getval()
         return val
 def urlreadable(s, _cleanstringsub=re.compile('[^-a-zA-Z0-9./]+').sub):
     return _cleanstringsub('_', safe_str(s)).rstrip('_')
 def recursive_replace(str_, replace=' '):
     """
     Recursive replace of given sign to just one instance
     :param str_: given string
     :param replace: char to find and replace multiple instances
     Examples::
     >>> recursive_replace("Mighty---Mighty-Bo--sstones",'-')
     'Mighty-Mighty-Bo-sstones'
     """
     if str_.find(replace * 2) == -1:
         return str_
     else:
         str_ = str_.replace(replace * 2, replace)
         return recursive_replace(str_, replace)
 def repo_name_slug(value):
     """
     Return slug of name of repository
     This function is called on each creation/modification
     of repository to prevent bad names in repo
     """
     slug = remove_formatting(value)
     slug = strip_tags(slug)
     for c in """`?=[]\;'"<>,/~!@#$%^&*()+{}|: """:
         slug = slug.replace(c, '-')
     slug = recursive_replace(slug, '-')
     slug = collapse(slug, '-')
     return slug

kallithea/model/repo.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo
 ~~~~~~~~~~~~~~~~~~~~
 Repository model for kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jun 5, 2010
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import shutil
 import logging
 import traceback
 from datetime import datetime
 from sqlalchemy.orm import subqueryload
-import kallithea.lib.utils
+import kallithea.lib.utils2
 from kallithea.lib.utils import make_ui, is_valid_repo_uri
 from kallithea.lib.vcs.backends import get_backend
 from kallithea.lib.utils2 import LazyProperty, safe_str, safe_unicode, \
     remove_prefix, obfuscate_url_pw, get_current_authuser
 from kallithea.lib.caching_query import FromCache
 from kallithea.lib.hooks import log_delete_repository
 from kallithea.model.db import Repository, UserRepoToPerm, UserGroupRepoToPerm, \
     UserRepoGroupToPerm, UserGroupRepoGroupToPerm, User, Permission, Session, \
     Statistics, UserGroup, Ui, RepoGroup, RepositoryField
 from kallithea.lib import helpers as h
 from kallithea.lib.auth import HasRepoPermissionLevel, HasUserGroupPermissionLevel
 from kallithea.lib.exceptions import AttachedForksError
 from kallithea.model.scm import UserGroupList
 log = logging.getLogger(__name__)
 class RepoModel(object):
     URL_SEPARATOR = Repository.url_sep()
     def _create_default_perms(self, repository, private):
         # create default permission
         default = 'repository.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('repository.'):
                 default = p.permission.permission_name
                 break
         default_perm = 'repository.none' if private else default
         repo_to_perm = UserRepoToPerm()
         repo_to_perm.permission = Permission.get_by_key(default_perm)
         repo_to_perm.repository = repository
         repo_to_perm.user_id = def_user.user_id
         Session().add(repo_to_perm)
         return repo_to_perm
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.query().filter(Ui.ui_key == '/').one()
         return q.ui_value
     def get(self, repo_id, cache=False):
         repo = Repository.query() \
             .filter(Repository.repo_id == repo_id)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_id))
         return repo.scalar()
     def get_repo(self, repository):
         return Repository.guess_instance(repository)
     def get_by_repo_name(self, repo_name, cache=False):
         repo = Repository.query() \
             .filter(Repository.repo_name == repo_name)
         if cache:
             repo = repo.options(FromCache("sql_cache_short",
                                           "get_repo_%s" % repo_name))
         return repo.scalar()
     def get_all_user_repos(self, user):
         """
         Gets all repositories that user have at least read access
         :param user:
         """
         from kallithea.lib.auth import AuthUser
         user = User.guess_instance(user)
         repos = AuthUser(dbuser=user).permissions['repositories']
         access_check = lambda r: r[1] in ['repository.read',
                                           'repository.write',
                                           'repository.admin']
         repos = [x[0] for x in filter(access_check, repos.items())]
         return Repository.query().filter(Repository.repo_name.in_(repos))
     def get_users_js(self):
         users = User.query() \
             .filter(User.active == True) \
             .order_by(User.name, User.lastname) \
             .all()
         return [
+            {
                 'id': u.user_id,
                 'fname': h.escape(u.name),
                 'lname': h.escape(u.lastname),
                 'nname': u.username,
                 'gravatar_lnk': h.gravatar_url(u.email, size=28, default='default'),
                 'gravatar_size': 14,
             } for u in users]
     @classmethod
     def _render_datatable(cls, tmpl, *args, **kwargs):
         import kallithea
         from tg import tmpl_context as c, request, app_globals
         from tg.i18n import ugettext as _
         _tmpl_lookup = app_globals.mako_lookup
         template = _tmpl_lookup.get_template('data_table/_dt_elements.html')
         tmpl = template.get_def(tmpl)
         kwargs.update(dict(_=_, h=h, c=c, request=request))
         return tmpl.render(*args, **kwargs)
     def get_repos_as_dict(self, repos_list, repo_groups_list=None,
                           admin=False,
                           short_name=False):
         """Return repository list for use by DataTable.
         repos_list: list of repositories - but will be filtered for read permission.
         repo_groups_list: added at top of list without permission check.
         admin: return data for action column.
         """
         _render = self._render_datatable
         from tg import tmpl_context as c
         def repo_lnk(name, rtype, rstate, private, fork_of):
             return _render('repo_name', name, rtype, rstate, private, fork_of,
                            short_name=short_name)
         def last_change(last_change):
             return _render("last_change", last_change)
         def rss_lnk(repo_name):
             return _render("rss", repo_name)
         def atom_lnk(repo_name):
             return _render("atom", repo_name)
         def last_rev(repo_name, cs_cache):
             return _render('revision', repo_name, cs_cache.get('revision'),
                            cs_cache.get('raw_id'), cs_cache.get('author'),
                            cs_cache.get('message'))
         def desc(desc):
             return h.urlify_text(desc, truncate=80, stylize=c.visual.stylify_metalabels)
         def state(repo_state):
             return _render("repo_state", repo_state)
         def repo_actions(repo_name):
             return _render('repo_actions', repo_name)
         def owner_actions(owner_id, username):
             return _render('user_name', owner_id, username)
         repos_data = []
         for gr in repo_groups_list or []:
             repos_data.append(dict(
                 raw_name='\0' + gr.name, # sort before repositories
                 just_name=gr.name,
                 name=_render('group_name_html', group_name=gr.group_name, name=gr.name),
                 desc=gr.group_description))
         for repo in repos_list:
             if not HasRepoPermissionLevel('read')(repo.repo_name, 'get_repos_as_dict check'):
                 continue
             cs_cache = repo.changeset_cache
             row = {
                 "raw_name": repo.repo_name,
                 "just_name": repo.just_name,
                 "name": repo_lnk(repo.repo_name, repo.repo_type,
                                  repo.repo_state, repo.private, repo.fork),
                 "last_change_iso": repo.last_db_change.isoformat(),
                 "last_change": last_change(repo.last_db_change),
                 "last_changeset": last_rev(repo.repo_name, cs_cache),
                 "last_rev_raw": cs_cache.get('revision'),
                 "desc": desc(repo.description),
                 "owner": h.person(repo.owner),
                 "state": state(repo.repo_state),
                 "rss": rss_lnk(repo.repo_name),
                 "atom": atom_lnk(repo.repo_name),
+            }
             if admin:
                 row.update({
                     "action": repo_actions(repo.repo_name),
                     "owner": owner_actions(repo.owner_id,
                                            h.person(repo.owner))
                 })
             repos_data.append(row)
         return {
             "sort": "name",
             "dir": "asc",
             "records": repos_data
+        }
     def _get_defaults(self, repo_name):
         """
         Gets information about repository, and returns a dict for
         usage in forms
         :param repo_name:
         """
         repo_info = Repository.get_by_repo_name(repo_name)
         if repo_info is None:
             return None
         defaults = repo_info.get_dict()
         defaults['repo_name'] = repo_info.just_name
         defaults['repo_group'] = repo_info.group_id
         for strip, k in [(0, 'repo_type'), (1, 'repo_enable_downloads'),
                          (1, 'repo_description'), (1, 'repo_enable_locking'),
                          (1, 'repo_landing_rev'), (0, 'clone_uri'),
                          (1, 'repo_private'), (1, 'repo_enable_statistics')]:
             attr = k
             if strip:
                 attr = remove_prefix(k, 'repo_')
             val = defaults[attr]
             if k == 'repo_landing_rev':
                 val = ':'.join(defaults[attr])
             defaults[k] = val
             if k == 'clone_uri':
                 defaults['clone_uri_hidden'] = repo_info.clone_uri_hidden
         # fill owner
         if repo_info.owner:
             defaults.update({'owner': repo_info.owner.username})
         else:
             replacement_user = User.query().filter(User.admin ==
                                                    True).first().username
             defaults.update({'owner': replacement_user})
         # fill repository users
         for p in repo_info.repo_to_perm:
             defaults.update({'u_perm_%s' % p.user.username:
                                  p.permission.permission_name})
         # fill repository groups
         for p in repo_info.users_group_to_perm:
             defaults.update({'g_perm_%s' % p.users_group.users_group_name:
                                  p.permission.permission_name})
         return defaults
     def update(self, repo, **kwargs):
         try:
             cur_repo = Repository.guess_instance(repo)
             org_repo_name = cur_repo.repo_name
             if 'owner' in kwargs:
                 cur_repo.owner = User.get_by_username(kwargs['owner'])
             if 'repo_group' in kwargs:
                 assert kwargs['repo_group'] != u'-1', kwargs # RepoForm should have converted to None
                 cur_repo.group = RepoGroup.get(kwargs['repo_group'])
                 cur_repo.repo_name = cur_repo.get_new_name(cur_repo.just_name)
             log.debug('Updating repo %s with params:%s', cur_repo, kwargs)
             for k in ['repo_enable_downloads',
                       'repo_description',
                       'repo_enable_locking',
                       'repo_landing_rev',
                       'repo_private',
                       'repo_enable_statistics',
                       ]:
                 if k in kwargs:
                     setattr(cur_repo, remove_prefix(k, 'repo_'), kwargs[k])
             clone_uri = kwargs.get('clone_uri')
             if clone_uri is not None and clone_uri != cur_repo.clone_uri_hidden:
                 # clone_uri is modified - if given a value, check it is valid
                 if clone_uri != '':
                     # will raise exception on error
                     is_valid_repo_uri(cur_repo.repo_type, clone_uri, make_ui('db', clear_session=False))
                 cur_repo.clone_uri = clone_uri
             if 'repo_name' in kwargs:
                 repo_name = kwargs['repo_name']
-                if kallithea.lib.utils.repo_name_slug(repo_name) != repo_name:
+                if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                     raise Exception('invalid repo name %s' % repo_name)
                 cur_repo.repo_name = cur_repo.get_new_name(repo_name)
             # if private flag is set, reset default permission to NONE
             if kwargs.get('repo_private'):
                 EMPTY_PERM = 'repository.none'
                 RepoModel().grant_user_permission(
                     repo=cur_repo, user='default', perm=EMPTY_PERM
+                )
                 # handle extra fields
             for field in filter(lambda k: k.startswith(RepositoryField.PREFIX),
                                 kwargs):
                 k = RepositoryField.un_prefix_key(field)
                 ex_field = RepositoryField.get_by_key_name(key=k, repo=cur_repo)
                 if ex_field:
                     ex_field.field_value = kwargs[field]
             if org_repo_name != cur_repo.repo_name:
                 # rename repository
                 self._rename_filesystem_repo(old=org_repo_name, new=cur_repo.repo_name)
             return cur_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _create_repo(self, repo_name, repo_type, description, owner,
                      private=False, clone_uri=None, repo_group=None,
                      landing_rev='rev:tip', fork_of=None,
                      copy_fork_permissions=False, enable_statistics=False,
                      enable_locking=False, enable_downloads=False,
                      copy_group_permissions=False, state=Repository.STATE_PENDING):
         """
         Create repository inside database with PENDING state. This should only be
         executed by create() repo, with exception of importing existing repos.
         """
         from kallithea.model.scm import ScmModel
         owner = User.guess_instance(owner)
         fork_of = Repository.guess_instance(fork_of)
         repo_group = RepoGroup.guess_instance(repo_group)
         try:
             repo_name = safe_unicode(repo_name)
             description = safe_unicode(description)
             # repo name is just a name of repository
             # while repo_name_full is a full qualified name that is combined
             # with name and path of group
             repo_name_full = repo_name
             repo_name = repo_name.split(self.URL_SEPARATOR)[-1]
-            if kallithea.lib.utils.repo_name_slug(repo_name) != repo_name:
+            if kallithea.lib.utils2.repo_name_slug(repo_name) != repo_name:
                 raise Exception('invalid repo name %s' % repo_name)
             new_repo = Repository()
             new_repo.repo_state = state
             new_repo.enable_statistics = False
             new_repo.repo_name = repo_name_full
             new_repo.repo_type = repo_type
             new_repo.owner = owner
             new_repo.group = repo_group
             new_repo.description = description or repo_name
             new_repo.private = private
             if clone_uri:
                 # will raise exception on error
                 is_valid_repo_uri(repo_type, clone_uri, make_ui('db', clear_session=False))
             new_repo.clone_uri = clone_uri
             new_repo.landing_rev = landing_rev
             new_repo.enable_statistics = enable_statistics
             new_repo.enable_locking = enable_locking
             new_repo.enable_downloads = enable_downloads
             if repo_group:
                 new_repo.enable_locking = repo_group.enable_locking
             if fork_of:
                 parent_repo = fork_of
                 new_repo.fork = parent_repo
             Session().add(new_repo)
             if fork_of and copy_fork_permissions:
                 repo = fork_of
                 user_perms = UserRepoToPerm.query() \
                     .filter(UserRepoToPerm.repository == repo).all()
                 group_perms = UserGroupRepoToPerm.query() \
                     .filter(UserGroupRepoToPerm.repository == repo).all()
                 for perm in user_perms:
                     UserRepoToPerm.create(perm.user, new_repo, perm.permission)
                 for perm in group_perms:
                     UserGroupRepoToPerm.create(perm.users_group, new_repo,
                                                perm.permission)
             elif repo_group and copy_group_permissions:
                 user_perms = UserRepoGroupToPerm.query() \
                     .filter(UserRepoGroupToPerm.group == repo_group).all()
                 group_perms = UserGroupRepoGroupToPerm.query() \
                     .filter(UserGroupRepoGroupToPerm.group == repo_group).all()
                 for perm in user_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = Permission.get_by_key(perm_name)
                     UserRepoToPerm.create(perm.user, new_repo, perm_obj)
                 for perm in group_perms:
                     perm_name = perm.permission.permission_name.replace('group.', 'repository.')
                     perm_obj = Permission.get_by_key(perm_name)
                     UserGroupRepoToPerm.create(perm.users_group, new_repo, perm_obj)
             else:
                 self._create_default_perms(new_repo, private)
             # now automatically start following this repository as owner
             ScmModel().toggle_following_repo(new_repo.repo_id, owner.user_id)
             # we need to flush here, in order to check if database won't
             # throw any exceptions, create filesystem dirs at the very end
             Session().flush()
             return new_repo
         except Exception:
             log.error(traceback.format_exc())
             raise
     def create(self, form_data, cur_user):
         """
         Create repository using celery tasks
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo(form_data, cur_user)
     def _update_permissions(self, repo, perms_new=None, perms_updates=None,
                             check_perms=True):
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         # update permissions
         for member, perm, member_type in perms_updates:
             if member_type == 'user':
                 # this updates existing one
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
             # set new permissions
         for member, perm, member_type in perms_new:
             if member_type == 'user':
                 self.grant_user_permission(
                     repo=repo, user=member, perm=perm
+                )
             else:
                 # check if we have permissions to alter this usergroup's access
                 if not check_perms or HasUserGroupPermissionLevel('read')(member):
                     self.grant_user_group_permission(
                         repo=repo, group_name=member, perm=perm
+                    )
     def create_fork(self, form_data, cur_user):
         """
         Simple wrapper into executing celery task for fork creation
         :param form_data:
         :param cur_user:
         """
         from kallithea.lib.celerylib import tasks
         return tasks.create_repo_fork(form_data, cur_user)
     def delete(self, repo, forks=None, fs_remove=True, cur_user=None):
         """
         Delete given repository, forks parameter defines what do do with
         attached forks. Throws AttachedForksError if deleted repo has attached
         forks
         :param repo:
         :param forks: str 'delete' or 'detach'
         :param fs_remove: remove(archive) repo from filesystem
         """
         if not cur_user:
             cur_user = getattr(get_current_authuser(), 'username', None)
         repo = Repository.guess_instance(repo)
         if repo is not None:
             if forks == 'detach':
                 for r in repo.forks:
                     r.fork = None
             elif forks == 'delete':
                 for r in repo.forks:
                     self.delete(r, forks='delete')
             elif [f for f in repo.forks]:
                 raise AttachedForksError()
             old_repo_dict = repo.get_dict()
             try:
                 Session().delete(repo)
                 if fs_remove:
                     self._delete_filesystem_repo(repo)
                 else:
                     log.debug('skipping removal from filesystem')
                 log_delete_repository(old_repo_dict,
                                       deleted_by=cur_user)
             except Exception:
                 log.error(traceback.format_exc())
                 raise
     def grant_user_permission(self, repo, user, perm):
         """
         Grant permission for user on given repository, or update existing one
         if found
         :param repo: Instance of Repository, repository_id, or repository name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         user = User.guess_instance(user)
         repo = Repository.guess_instance(repo)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = UserRepoToPerm.query() \
             .filter(UserRepoToPerm.user == user) \
             .filter(UserRepoToPerm.repository == repo) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserRepoToPerm()
             Session().add(obj)
         obj.repository = repo
         obj.user = user
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, user, repo)
         return obj

kallithea/model/repo_group.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 kallithea.model.repo_group
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 repo group model for Kallithea
 This file was forked by the Kallithea project in July 2014.
 Original author and date, and relevant copyright and licensing information is below:
 :created_on: Jan 25, 2011
 :author: marcink
 :copyright: (c) 2013 RhodeCode GmbH, and others.
 :license: GPLv3, see LICENSE.md for more details.
 """
 import os
 import logging
 import traceback
 import shutil
 import datetime
-import kallithea.lib.utils
+import kallithea.lib.utils2
 from kallithea.lib.utils2 import LazyProperty
 from kallithea.model.db import RepoGroup, Session, Ui, UserRepoGroupToPerm, \
     User, Permission, UserGroupRepoGroupToPerm, UserGroup, Repository
 log = logging.getLogger(__name__)
 class RepoGroupModel(object):
     @LazyProperty
     def repos_path(self):
         """
         Gets the repositories root path from database
         """
         q = Ui.get_by_key('paths', '/')
         return q.ui_value
     def _create_default_perms(self, new_group):
         # create default permission
         default_perm = 'group.read'
         def_user = User.get_default_user()
         for p in def_user.user_perms:
             if p.permission.permission_name.startswith('group.'):
                 default_perm = p.permission.permission_name
                 break
         repo_group_to_perm = UserRepoGroupToPerm()
         repo_group_to_perm.permission = Permission.get_by_key(default_perm)
         repo_group_to_perm.group = new_group
         repo_group_to_perm.user_id = def_user.user_id
         Session().add(repo_group_to_perm)
         return repo_group_to_perm
     def _create_group(self, group_name):
         """
         makes repository group on filesystem
         :param repo_name:
         :param parent_id:
         """
         create_path = os.path.join(self.repos_path, group_name)
         log.debug('creating new group in %s', create_path)
         if os.path.isdir(create_path):
             raise Exception('That directory already exists !')
         os.makedirs(create_path)
         log.debug('Created group in %s', create_path)
     def _rename_group(self, old, new):
         """
         Renames a group on filesystem
         :param group_name:
         """
         if old == new:
             log.debug('skipping group rename')
             return
         log.debug('renaming repository group from %s to %s', old, new)
         old_path = os.path.join(self.repos_path, old)
         new_path = os.path.join(self.repos_path, new)
         log.debug('renaming repos paths from %s to %s', old_path, new_path)
         if os.path.isdir(new_path):
             raise Exception('Was trying to rename to already '
                             'existing dir %s' % new_path)
         shutil.move(old_path, new_path)
     def _delete_group(self, group, force_delete=False):
         """
         Deletes a group from a filesystem
         :param group: instance of group from database
         :param force_delete: use shutil rmtree to remove all objects
         """
         paths = group.full_path.split(RepoGroup.url_sep())
         paths = os.sep.join(paths)
         rm_path = os.path.join(self.repos_path, paths)
         log.info("Removing group %s", rm_path)
         # delete only if that path really exists
         if os.path.isdir(rm_path):
             if force_delete:
                 shutil.rmtree(rm_path)
             else:
                 # archive that group
                 _now = datetime.datetime.now()
                 _ms = str(_now.microsecond).rjust(6, '0')
                 _d = 'rm__%s_GROUP_%s' % (_now.strftime('%Y%m%d_%H%M%S_' + _ms),
                                           group.name)
                 shutil.move(rm_path, os.path.join(self.repos_path, _d))
     def create(self, group_name, group_description, owner, parent=None,
                just_db=False, copy_permissions=False):
         try:
-            if kallithea.lib.utils.repo_name_slug(group_name) != group_name:
+            if kallithea.lib.utils2.repo_name_slug(group_name) != group_name:
                 raise Exception('invalid repo group name %s' % group_name)
             owner = User.guess_instance(owner)
             parent_group = RepoGroup.guess_instance(parent)
             new_repo_group = RepoGroup()
             new_repo_group.owner = owner
             new_repo_group.group_description = group_description or group_name
             new_repo_group.parent_group = parent_group
             new_repo_group.group_name = new_repo_group.get_new_name(group_name)
             Session().add(new_repo_group)
             # create an ADMIN permission for owner except if we're super admin,
             # later owner should go into the owner field of groups
             if not owner.is_admin:
                 self.grant_user_permission(repo_group=new_repo_group,
                                            user=owner, perm='group.admin')
             if parent_group and copy_permissions:
                 # copy permissions from parent
                 user_perms = UserRepoGroupToPerm.query() \
                     .filter(UserRepoGroupToPerm.group == parent_group).all()
                 group_perms = UserGroupRepoGroupToPerm.query() \
                     .filter(UserGroupRepoGroupToPerm.group == parent_group).all()
                 for perm in user_perms:
                     # don't copy over the permission for user who is creating
                     # this group, if he is not super admin he get's admin
                     # permission set above
                     if perm.user != owner or owner.is_admin:
                         UserRepoGroupToPerm.create(perm.user, new_repo_group, perm.permission)
                 for perm in group_perms:
                     UserGroupRepoGroupToPerm.create(perm.users_group, new_repo_group, perm.permission)
             else:
                 self._create_default_perms(new_repo_group)
             if not just_db:
                 # we need to flush here, in order to check if database won't
                 # throw any exceptions, create filesystem dirs at the very end
                 Session().flush()
                 self._create_group(new_repo_group.group_name)
             return new_repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def _update_permissions(self, repo_group, perms_new=None,
                             perms_updates=None, recursive=None,
                             check_perms=True):
         from kallithea.model.repo import RepoModel
         from kallithea.lib.auth import HasUserGroupPermissionLevel
         if not perms_new:
             perms_new = []
         if not perms_updates:
             perms_updates = []
         def _set_perm_user(obj, user, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_permission(repo_group=obj, user=user, perm=perm)
             elif isinstance(obj, Repository):
                 user = User.guess_instance(user)
                 # private repos will not allow to change the default permissions
                 # using recursive mode
                 if obj.private and user.is_default_user:
                     return
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_permission(
                     repo=obj, user=user, perm=perm
+                )
         def _set_perm_group(obj, users_group, perm):
             if isinstance(obj, RepoGroup):
                 self.grant_user_group_permission(repo_group=obj,
                                                   group_name=users_group,
                                                   perm=perm)
             elif isinstance(obj, Repository):
                 # we set group permission but we have to switch to repo
                 # permission
                 perm = perm.replace('group.', 'repository.')
                 RepoModel().grant_user_group_permission(
                     repo=obj, group_name=users_group, perm=perm
+                )
         # start updates
         updates = []
         log.debug('Now updating permissions for %s in recursive mode:%s',
                   repo_group, recursive)
         for obj in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(obj, RepoGroup) and not obj == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(obj, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 obj = repo_group
                 # also we do a break at the end of this loop.
             # update permissions
             for member, perm, member_type in perms_updates:
                 ## set for user
                 if member_type == 'user':
                     # this updates also current one if found
                     _set_perm_user(obj, user=member, perm=perm)
                 ## set for user group
                 else:
                     # check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             # set new permissions
             for member, perm, member_type in perms_new:
                 if member_type == 'user':
                     _set_perm_user(obj, user=member, perm=perm)
                 else:
                     # check if we have permissions to alter this usergroup's access
                     if not check_perms or HasUserGroupPermissionLevel('read')(member):
                         _set_perm_group(obj, users_group=member, perm=perm)
             updates.append(obj)
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
         return updates
     def update(self, repo_group, kwargs):
         try:
             repo_group = RepoGroup.guess_instance(repo_group)
             old_path = repo_group.full_path
             # change properties
             if 'group_description' in kwargs:
                 repo_group.group_description = kwargs['group_description']
             if 'parent_group_id' in kwargs:
                 repo_group.parent_group_id = kwargs['parent_group_id']
             if 'enable_locking' in kwargs:
                 repo_group.enable_locking = kwargs['enable_locking']
             if 'parent_group_id' in kwargs:
                 assert kwargs['parent_group_id'] != u'-1', kwargs # RepoGroupForm should have converted to None
                 repo_group.parent_group = RepoGroup.get(kwargs['parent_group_id'])
             if 'group_name' in kwargs:
                 group_name = kwargs['group_name']
-                if kallithea.lib.utils.repo_name_slug(group_name) != group_name:
+                if kallithea.lib.utils2.repo_name_slug(group_name) != group_name:
                     raise Exception('invalid repo group name %s' % group_name)
                 repo_group.group_name = repo_group.get_new_name(group_name)
             new_path = repo_group.full_path
             Session().add(repo_group)
             # iterate over all members of this groups and do fixes
             # set locking if given
             # if obj is a repoGroup also fix the name of the group according
             # to the parent
             # if obj is a Repo fix it's name
             # this can be potentially heavy operation
             for obj in repo_group.recursive_groups_and_repos():
                 # set the value from it's parent
                 obj.enable_locking = repo_group.enable_locking
                 if isinstance(obj, RepoGroup):
                     new_name = obj.get_new_name(obj.name)
                     log.debug('Fixing group %s to new name %s' \
                                 % (obj.group_name, new_name))
                     obj.group_name = new_name
                 elif isinstance(obj, Repository):
                     # we need to get all repositories from this new group and
                     # rename them accordingly to new group path
                     new_name = obj.get_new_name(obj.just_name)
                     log.debug('Fixing repo %s to new name %s' \
                                 % (obj.repo_name, new_name))
                     obj.repo_name = new_name
             self._rename_group(old_path, new_path)
             return repo_group
         except Exception:
             log.error(traceback.format_exc())
             raise
     def delete(self, repo_group, force_delete=False):
         repo_group = RepoGroup.guess_instance(repo_group)
         try:
             Session().delete(repo_group)
             self._delete_group(repo_group, force_delete)
         except Exception:
             log.error('Error removing repo_group %s', repo_group)
             raise
     def add_permission(self, repo_group, obj, obj_type, perm, recursive):
         from kallithea.model.repo import RepoModel
         repo_group = RepoGroup.guess_instance(repo_group)
         perm = Permission.guess_instance(perm)
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
             if isinstance(el, RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().grant_user_permission(el, user=obj, perm=perm)
                 elif obj_type == 'user_group':
                     RepoGroupModel().grant_user_group_permission(el, group_name=obj, perm=perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, Repository):
                 # for repos we need to hotfix the name of permission
                 _perm = perm.permission_name.replace('group.', 'repository.')
                 if obj_type == 'user':
                     RepoModel().grant_user_permission(el, user=obj, perm=_perm)
                 elif obj_type == 'user_group':
                     RepoModel().grant_user_group_permission(el, group_name=obj, perm=_perm)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
     def delete_permission(self, repo_group, obj, obj_type, recursive):
         """
         Revokes permission for repo_group for given obj(user or users_group),
         obj_type can be user or user group
         :param repo_group:
         :param obj: user or user group id
         :param obj_type: user or user group type
         :param recursive: recurse to all children of group
         """
         from kallithea.model.repo import RepoModel
         repo_group = RepoGroup.guess_instance(repo_group)
         for el in repo_group.recursive_groups_and_repos():
             # iterated obj is an instance of a repos group or repository in
             # that group, recursive option can be: none, repos, groups, all
             if recursive == 'all':
                 pass
             elif recursive == 'repos':
                 # skip groups, other than this one
                 if isinstance(el, RepoGroup) and not el == repo_group:
                     continue
             elif recursive == 'groups':
                 # skip repos
                 if isinstance(el, Repository):
                     continue
             else:  # recursive == 'none': # DEFAULT don't apply to iterated objects
                 el = repo_group
                 # also we do a break at the end of this loop.
             if isinstance(el, RepoGroup):
                 if obj_type == 'user':
                     RepoGroupModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     RepoGroupModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             elif isinstance(el, Repository):
                 if obj_type == 'user':
                     RepoModel().revoke_user_permission(el, user=obj)
                 elif obj_type == 'user_group':
                     RepoModel().revoke_user_group_permission(el, group_name=obj)
                 else:
                     raise Exception('undefined object type %s' % obj_type)
             else:
                 raise Exception('el should be instance of Repository or '
                                 'RepositoryGroup got %s instead' % type(el))
             # if it's not recursive call for all,repos,groups
             # break the loop and don't proceed with other changes
             if recursive not in ['all', 'repos', 'groups']:
                 break
     def grant_user_permission(self, repo_group, user, perm):
         """
         Grant permission for user on given repository group, or update
         existing one if found
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         :param perm: Instance of Permission, or permission_name
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         user = User.guess_instance(user)
         permission = Permission.guess_instance(perm)
         # check if we have that permission already
         obj = UserRepoGroupToPerm.query() \
             .filter(UserRepoGroupToPerm.user == user) \
             .filter(UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is None:
             # create new !
             obj = UserRepoGroupToPerm()
             Session().add(obj)
         obj.group = repo_group
         obj.user = user
         obj.permission = permission
         log.debug('Granted perm %s to %s on %s', perm, user, repo_group)
         return obj
     def revoke_user_permission(self, repo_group, user):
         """
         Revoke permission for user on given repository group
         :param repo_group: Instance of RepoGroup, repositories_group_id,
             or repositories_group name
         :param user: Instance of User, user_id or username
         """
         repo_group = RepoGroup.guess_instance(repo_group)
         user = User.guess_instance(user)
         obj = UserRepoGroupToPerm.query() \
             .filter(UserRepoGroupToPerm.user == user) \
             .filter(UserRepoGroupToPerm.group == repo_group) \
             .scalar()
         if obj is not None:
             Session().delete(obj)
             log.debug('Revoked perm on %s on %s', repo_group, user)

kallithea/model/validators.py

➞

Show inline comments

 # -*- coding: utf-8 -*-
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
+#
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
+#
 # You should have received a copy of the GNU General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 Set of generic validators
 """
 import os
 import re
 import formencode
 import logging
 from collections import defaultdict
 from tg.i18n import ugettext as _
 from sqlalchemy import func
 from webhelpers.pylonslib.secure_form import authentication_token
 import sqlalchemy
 from formencode.validators import (
     UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set,
     NotEmpty, IPAddress, CIDR, String, FancyValidator
+)
 from kallithea.lib.compat import OrderedSet
 from kallithea.lib import ipaddr
 from kallithea.lib.utils import repo_name_slug, is_valid_repo_uri
 from kallithea.lib.utils2 import str2bool, aslist
 from kallithea.lib.utils import is_valid_repo_uri
 from kallithea.lib.utils2 import str2bool, aslist, repo_name_slug
 from kallithea.model.db import RepoGroup, Repository, UserGroup, User
 from kallithea.lib.exceptions import LdapImportError
 from kallithea.config.routing import ADMIN_PREFIX
 from kallithea.lib.auth import HasRepoGroupPermissionLevel, HasPermissionAny
 # silence warnings and pylint
 UnicodeString, OneOf, Int, Number, Regex, Email, Bool, StringBoolean, Set, \
     NotEmpty, IPAddress, CIDR, String, FancyValidator
 log = logging.getLogger(__name__)
 def UniqueListFromString():
     class _UniqueListFromString(formencode.FancyValidator):
         """
         Split value on ',' and make unique while preserving order
         """
         messages = dict(
             empty=_('Value cannot be an empty list'),
             missing_value=_('Value cannot be an empty list'),
+        )
         def _to_python(self, value, state):
             value = aslist(value, ',')
             seen = set()
             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=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'username_exists': _('Username "%(username)s" already exists'),
             'system_invalid_username':
                 _('Username "%(username)s" cannot be used'),
             'invalid_username':
                 _('Username may only contain alphanumeric characters '
                   'underscores, periods or dashes and must begin with an '
                   'alphanumeric character or underscore')
+        }
         def validate_python(self, value, state):
             if value in ['default', 'new_user']:
                 msg = self.message('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 = self.message('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 = self.message('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': _('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 = self.message('invalid_username', state, username=value)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(username=msg)
+                )
     return _validator
 def ValidUserGroup(edit=False, old_data=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'invalid_group': _('Invalid user group name'),
             'group_exist': _('User group "%(usergroup)s" already exists'),
             'invalid_usergroup_name':
                 _('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 = self.message('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 = self.message('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 = self.message('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=None):
     old_data = old_data or {}
     class _validator(formencode.validators.FancyValidator):
         messages = {
             'parent_group_id': _('Cannot assign this group as parent'),
             'group_exists': _('Group "%(group_name)s" already exists'),
             'repo_exists':
                 _('Repository with name "%(group_name)s" already exists')
+        }
         def validate_python(self, value, state):
             # TODO WRITE VALIDATIONS
             group_name = value.get('group_name')
             parent_group_id = value.get('parent_group_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'] == parent_group_id
                 if parent_group_id else False
+            )
             if edit and parent_of_self():
                 msg = self.message('parent_group_id', state)
                 raise formencode.Invalid(msg, value, state,
                     error_dict=dict(parent_group_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(func.lower(RepoGroup.group_name) == func.lower(slug)) \
                       .filter(RepoGroup.parent_group_id == parent_group_id) \
                       .scalar()
                 if gr is not None:
                     msg = self.message('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(func.lower(Repository.repo_name) == func.lower(slug)) \
                       .scalar()
                 if repo is not None:
                     msg = self.message('repo_exists', state, group_name=slug)
                     raise formencode.Invalid(msg, value, state,
                             error_dict=dict(group_name=msg)
+                    )

0 comments (0 inline, 0 general)