@@ -63,206 +63,205 @@ def recursive_replace(str_, replace=' ')
: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
def get_repo_slug(request):
return request.environ['pylons.routes_dict'].get('repo_name')
def action_logger(user, action, repo, ipaddr='', sa=None, 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
:param sa: optional sqlalchemy session
if not sa:
sa = meta.Session
try:
if hasattr(user, 'user_id'):
user_obj = user
elif isinstance(user, basestring):
user_obj = User.get_by_username(user)
raise Exception('You have to provide user object or username')
if hasattr(repo, 'repo_id'):
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)
raise Exception('You have to provide repository to action logger')
user_log = UserLog()
user_log.user_id = user_obj.user_id
user_log.action = action
user_log.repository_id = repo_obj.repo_id
user_log.repository_name = repo_name
user_log.action_date = datetime.datetime.now()
user_log.user_ip = ipaddr
sa.add(user_log)
log.info('Adding user %s, action %s on %s', user_obj, action, repo)
if commit:
sa.commit()
except:
log.error(traceback.format_exc())
raise
def get_repos(path, recursive=False):
Scans given path for repos and return (name,(type,path)) tuple
:param path: path to scann for repositories
:param recursive: recursive search and return names with subdirs in front
if path.endswith(os.sep):
#remove ending slash for better results
path = path[:-1]
# remove ending slash for better results
path = path.rstrip('/')
def _get_repos(p):
if not os.access(p, os.W_OK):
return
for dirpath in os.listdir(p):
if os.path.isfile(os.path.join(p, dirpath)):
continue
cur_path = os.path.join(p, dirpath)
scm_info = get_scm(cur_path)
yield scm_info[1].split(path)[-1].lstrip(os.sep), scm_info
yield scm_info[1].split(path, 1)[-1].lstrip(os.sep), scm_info
except VCSError:
if not recursive:
#check if this dir containts other repos for recursive scan
rec_path = os.path.join(p, dirpath)
if os.path.isdir(rec_path):
for inner_scm in _get_repos(rec_path):
yield inner_scm
return _get_repos(path)
def is_valid_repo(repo_name, base_path):
Returns True if given path is a valid repository False otherwise
:param repo_name:
:param base_path:
:return True: if given path is a valid repository
full_path = os.path.join(base_path, repo_name)
get_scm(full_path)
return True
return False
def is_valid_repos_group(repos_group_name, base_path):
Returns True if given path is a repos group False otherwise
full_path = os.path.join(base_path, repos_group_name)
# check if it's not a repo
if is_valid_repo(repos_group_name, base_path):
# check if it's a valid path
if os.path.isdir(full_path):
def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):
while True:
ok = raw_input(prompt)
if ok in ('y', 'ye', 'yes'):
if ok in ('n', 'no', 'nop', 'nope'):
retries = retries - 1
if retries < 0:
raise IOError
print complaint
#propagated from mercurial documentation
ui_sections = ['alias', 'auth',
'decode/encode', 'defaults',
'diff', 'email',
'extensions', 'format',
'merge-patterns', 'merge-tools',
'hooks', 'http_proxy',
'smtp', 'patch',
'paths', 'profiling',
'server', 'trusted',
'ui', 'web', ]
def make_ui(read_from='file', path=None, checkpaths=True):
"""A function that will read python rc files or database
and make an mercurial ui object from read options
:param path: path to mercurial config file
:param checkpaths: check the path
:param read_from: read from 'file' or 'db'
baseui = ui.ui()
#clean the baseui object
baseui._ocfg = config.config()
baseui._ucfg = config.config()
baseui._tcfg = config.config()
if read_from == 'file':
if not os.path.isfile(path):
log.warning('Unable to read config file %s' % path)
log.debug('reading hgrc from %s', path)
cfg = config.config()
cfg.read(path)
for section in ui_sections:
Status change: