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 datetime

import logging

import os

import re

import sys

import traceback

import urllib.error

from distutils.version import StrictVersion

import beaker.cache

import mercurial.config

import mercurial.ui

import kallithea.config.conf

from kallithea.lib.exceptions import InvalidCloneUriException

from kallithea.lib.utils2 import ascii_bytes, aslist, get_current_authuser, safe_bytes, safe_str

from kallithea.lib.vcs.backends.git.repository import GitRepository

from kallithea.lib.vcs.backends.hg.repository import MercurialRepository

from kallithea.lib.vcs.conf import settings

from kallithea.lib.vcs.exceptions import RepositoryError, VCSError

from kallithea.lib.vcs.utils.fakemod import create_module

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.model import meta

from kallithea.model.db import RepoGroup, Repository, Setting, Ui, User, UserGroup, UserLog

log = logging.getLogger(__name__)

REMOVED_REPO_PAT = re.compile(r'rm__\d{8}_\d{6}_\d{6}_.*')

#==============================================================================

# PERM DECORATOR HELPERS FOR EXTRACTING NAMES FOR PERM CHECKS

#==============================================================================

def get_repo_slug(request):

                    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 InvalidCloneUriException if any problems"""

    if repo_type == 'hg':

        if url.startswith('http') or url.startswith('ssh'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            try:

                MercurialRepository._check_url(url, ui)

            except urllib.error.URLError as e:

                raise InvalidCloneUriException('URI %s URLError: %s' % (url, e))

        elif url.startswith('svn+http'):

            try:

                from hgsubversion.svnrepo import svnremoterepo

            except ImportError:

                raise InvalidCloneUriException('URI type %s not supported - hgsubversion is not available' % (url,))

            svnremoterepo(ui, url).svn.uuid

        elif url.startswith('git+http'):

            raise InvalidCloneUriException('URI type %s not implemented' % (url,))

        else:

            raise InvalidCloneUriException('URI %s not allowed' % (url,))

    elif repo_type == 'git':

        if url.startswith('http') or url.startswith('git'):

            # initially check if it's at least the proper URL

            # or does it pass basic auth

            try:

                GitRepository._check_url(url)

            except urllib.error.URLError as e:

                raise InvalidCloneUriException('URI %s URLError: %s' % (url, e))

        elif url.startswith('svn+http'):

            raise InvalidCloneUriException('URI type %s not implemented' % (url,))

        elif url.startswith('hg+http'):

            raise InvalidCloneUriException('URI type %s not implemented' % (url,))

        else: