Utilities library for RhodeCode

    :created_on: Apr 18, 2010

    :author: marcink

    :copyright: (C) 2009-2011 Marcin Kuzminski <marcin@python-works.com>    

    :license: GPLv3, see COPYING for more details.

"""

# 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; version 2

# of the License or (at your opinion) any later version of the license.

# 

# 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, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,

# MA  02110-1301, USA.

import os

import logging

import datetime

import traceback

from UserDict import DictMixin

from mercurial import ui, config, hg

from mercurial.error import RepoError

import paste

import beaker

from paste.script.command import Command, BadCommand

from vcs.backends.base import BaseChangeset

from vcs.utils.lazy import LazyProperty

from rhodecode.model import meta

from rhodecode.model.caching_query import FromCache

from rhodecode.model.db import Repository, User, RhodeCodeUi, UserLog, Group

from rhodecode.model.repo import RepoModel

from rhodecode.model.user import UserModel

log = logging.getLogger(__name__)

def get_repo_slug(request):

    return request.environ['pylons.routes_dict'].get('repo_name')

def action_logger(user, action, repo, ipaddr='', sa=None):

    """

    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:

        um = UserModel()

        if hasattr(user, 'user_id'):

            user_obj = user

        elif isinstance(user, basestring):

            user_obj = um.get_by_username(user, cache=False)

        else:

            raise Exception('You have to provide user object or username')

        rm = RepoModel()

        if hasattr(repo, 'repo_id'):

            repo_obj = rm.get(repo.repo_id, cache=False)

            repo_name = repo_obj.repo_name

        elif  isinstance(repo, basestring):

            repo_name = repo.lstrip('/')

            repo_obj = rm.get_by_repo_name(repo_name, cache=False)

        else:

            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)

        sa.commit()

        log.info('Adding user %s, action %s on %s', user_obj, action, repo)

    except:

        log.error(traceback.format_exc())

        sa.rollback()

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 

    """

    from vcs.utils.helpers import get_scm

    from vcs.exceptions import VCSError

    if path.endswith('/'):

        #add ending slash for better results

        path = path[:-1]

    def _get_repos(p):

        for dirpath in os.listdir(p):

            if os.path.isfile(os.path.join(p, dirpath)):

                continue

            cur_path = os.path.join(p, dirpath)

            try:

                scm_info = get_scm(cur_path)

                yield scm_info[1].split(path)[-1].lstrip('/'), scm_info

            except VCSError:

                if not recursive:

                    continue

                #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 check_repo_fast(repo_name, base_path):

    """

    Check given path for existence of directory

    :param repo_name:

    :param base_path:

    :return False: if this directory is present

    """

    if os.path.isdir(os.path.join(base_path, repo_name)):return False

    return True

def check_repo(repo_name, base_path, verify=True):

    repo_path = os.path.join(base_path, repo_name)

    try:

        if not check_repo_fast(repo_name, base_path):

            return False

        r = hg.repository(ui.ui(), repo_path)

        if verify:

            hg.verify(r)

        #here we hnow that repo exists it was verified

        log.info('%s repo is already created', repo_name)

        return False

    except RepoError:

        #it means that there is no valid repo there...

        log.info('%s repo is free for creation', repo_name)

        return True

def ask_ok(prompt, retries=4, complaint='Yes or no, please!'):

    while True:

        ok = raw_input(prompt)

        if ok in ('y', 'ye', 'yes'): return True

        if ok in ('n', 'no', 'nop', 'nope'): return False

        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):

    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)

            return False

        log.debug('reading hgrc from %s', path)

        cfg = config.config()

        cfg.read(path)

        for section in ui_sections:

            for k, v in cfg.items(section):

                log.debug('settings ui from file[%s]%s:%s', section, k, v)

                baseui.setconfig(section, k, v)

    elif read_from == 'db':

        sa = meta.Session()

        ret = sa.query(RhodeCodeUi)\

            .options(FromCache("sql_cache_short",

                               "get_hg_ui_settings")).all()

        hg_ui = ret

        for ui_ in hg_ui:

            if ui_.ui_active:

                log.debug('settings ui from db[%s]%s:%s', ui_.ui_section,

                          ui_.ui_key, ui_.ui_value)

                baseui.setconfig(ui_.ui_section, ui_.ui_key, ui_.ui_value)

        meta.Session.remove()

    return baseui

def set_rhodecode_config(config):

    """Updates pylons config with new settings from database

    :param config:

    """

    from rhodecode.model.settings import SettingsModel

    hgsettings = SettingsModel().get_app_settings()

    for k, v in hgsettings.items():

        config[k] = v

def invalidate_cache(cache_key, *args):

    """Puts cache invalidation task into db for 

    further global cache invalidation

    """

    from rhodecode.model.scm import ScmModel

    if cache_key.startswith('get_repo_cached_'):

        name = cache_key.split('get_repo_cached_')[-1]

        ScmModel().mark_for_invalidation(name)

class EmptyChangeset(BaseChangeset):

    """

    An dummy empty changeset. It's possible to pass hash when creating

    an EmptyChangeset

    """

    def __init__(self, cs='0' * 40):

        self._empty_cs = cs

        self.revision = -1

        self.message = ''

        self.author = ''

        self.date = ''

    @LazyProperty

    def raw_id(self):

        """Returns raw string identifying this changeset, useful for web

        representation.

        """

        return self._empty_cs

    @LazyProperty

    def short_id(self):

        return self.raw_id[:12]

    def get_file_changeset(self, path):

        return self

    def get_file_content(self, path):

        return u''

    def get_file_size(self, path):

        return 0

def map_groups(groups):

    """Checks for groups existence, and creates groups structures.

    It returns last group in structure

    :param groups: list of groups structure

    """

    sa = meta.Session()

    parent = None

    group = None

    for lvl, group_name in enumerate(groups[:-1]):

        group = sa.query(Group).filter(Group.group_name == group_name).scalar()

        if group is None:

            group = Group(group_name, parent)

            sa.add(group)

            sa.commit()

        parent = group

    return group

def repo2db_mapper(initial_repo_list, remove_obsolete=False):

    """maps all repos given in initial_repo_list, non existing repositories

    are created, if remove_obsolete is True it also check for db entries

    that are not in initial_repo_list and removes them.

    :param initial_repo_list: list of repositories found by scanning methods

    :param remove_obsolete: check for obsolete entries in database

    """

    sa = meta.Session()

    rm = RepoModel()

    user = sa.query(User).filter(User.admin == True).first()

    for name, repo in initial_repo_list.items():

        group = map_groups(name.split('/'))

        if not rm.get_by_repo_name(name, cache=False):

            log.info('repository %s not found creating default', name)

            form_data = {

                         'repo_name':name,

                         'repo_type':repo.alias,

                         'description':repo.description \

                            if repo.description != 'unknown' else \

                                        '%s repository' % name,

                         'private':False,

                         'group_id':getattr(group, 'group_id', None)

                         }

            rm.create(form_data, user, just_db=True)

    if remove_obsolete:

        #remove from database those repositories that are not in the filesystem

        for repo in sa.query(Repository).all():

            if repo.repo_name not in initial_repo_list.keys():

                sa.delete(repo)

                sa.commit()

class OrderedDict(dict, DictMixin):

    def __init__(self, *args, **kwds):

        if len(args) > 1:

            raise TypeError('expected at most 1 arguments, got %d' % len(args))

        try:

            self.__end

        except AttributeError:

            self.clear()

        self.update(*args, **kwds)

    def clear(self):

        self.__end = end = []

        end += [None, end, end]         # sentinel node for doubly linked list

        self.__map = {}                 # key --> [key, prev, next]

        dict.clear(self)

    def __setitem__(self, key, value):

        if key not in self:

            end = self.__end

            curr = end[1]

            curr[2] = end[1] = self.__map[key] = [key, curr, end]

        dict.__setitem__(self, key, value)

    def __delitem__(self, key):

        dict.__delitem__(self, key)

        key, prev, next = self.__map.pop(key)

        prev[2] = next

        next[1] = prev

    def __iter__(self):

        end = self.__end

        curr = end[2]

        while curr is not end:

# -*- coding: utf-8 -*-

"""

    rhodecode.tests.test_hg_operations

    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    Test suite for making push/pull operations

    :created_on: Dec 30, 2010

    :copyright: (c) 2010 by marcink.

    :license: LICENSE_NAME, see LICENSE_FILE for more details.

"""

import os

import shutil

import logging

from subprocess import Popen, PIPE

from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO

USER = 'test_admin'

PASS = 'test12'

HOST = '127.0.0.1:5000'

DEBUG = True

log = logging.getLogger(__name__)

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    def execute(self, cmd, *args):

        """Runs command on the system with given ``args``.

        """

        command = cmd + ' ' + ' '.join(args)

        log.debug('Executing %s' % command)

        if DEBUG:

            print command

        p = Popen(command, shell=True, stdout=PIPE, stderr=PIPE, cwd=self.cwd)

        stdout, stderr = p.communicate()

        if DEBUG:

            print stdout, stderr

        return stdout, stderr

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

# TESTS

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

def test_clone():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

    stdout, stderr = Command(cwd).execute('hg clone', clone_url)

    assert """adding file changes""" in stdout, 'no messages about cloning'

    assert """abort""" not in stderr , 'got error from clone'

def test_clone_anonymous_ok():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

    stdout, stderr = Command(cwd).execute('hg clone', clone_url)

    print stdout, stderr

    assert """adding file changes""" in stdout, 'no messages about cloning'

    assert """abort""" not in stderr , 'got error from clone'

def test_clone_wrong_credentials():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        #print 'made dirs %s' % jn(path)

    except OSError:

        raise

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s %(dest)s' % \

                  {'user':USER + 'error',

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':path}

    stdout, stderr = Command(cwd).execute('hg clone', clone_url)

    assert """abort: authorization failed""" in stderr , 'no error from wrong credentials'

def test_pull():

    pass

def test_push():

    modified_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py')

    for i in xrange(5):

        cmd = """echo 'added_line%s' >> %s""" % (i, modified_file)

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'changed file %s' %s """ % (i, modified_file)

        Command(cwd).execute(cmd)

    Command(cwd).execute('hg push %s' % jn(TESTS_TMP_PATH, HG_REPO))

    test_clone()

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    Command(cwd).execute('touch %s' % added_file)

    Command(cwd).execute('hg add %s' % added_file)

        cmd = """echo 'added_line%s' >> %s""" % (i, added_file)

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file)

        Command(cwd).execute(cmd)

    push_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

def test_push_wrong_credentials():

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER + 'xxx',

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO,

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

    modified_file = jn(TESTS_TMP_PATH, HG_REPO, 'setup.py')

    for i in xrange(5):

        cmd = """echo 'added_line%s' >> %s""" % (i, modified_file)

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited %s' %s """ % (i, modified_file)

        Command(cwd).execute(cmd)

    Command(cwd).execute('hg push %s' % clone_url)

def test_push_wrong_path():

    cwd = path = jn(TESTS_TMP_PATH, HG_REPO)

    added_file = jn(path, 'somefile.py')

    try:

        shutil.rmtree(path, ignore_errors=True)

        os.makedirs(path)

        print 'made dirs %s' % jn(path)

    except OSError:

        raise

    Command(cwd).execute("""echo '' > %s""" % added_file)

    Command(cwd).execute("""hg init %s""" % path)

    Command(cwd).execute("""hg add %s""" % added_file)

    for i in xrange(2):

        cmd = """echo 'added_line%s' >> %s""" % (i, added_file)

        Command(cwd).execute(cmd)

        cmd = """hg ci -m 'commited new %s' %s """ % (i, added_file)

        Command(cwd).execute(cmd)

    clone_url = 'http://%(user)s:%(pass)s@%(host)s/%(cloned_repo)s' % \

                  {'user':USER,

                   'pass':PASS,

                   'host':HOST,

                   'cloned_repo':HG_REPO + '_error',

                   'dest':jn(TESTS_TMP_PATH, HG_REPO)}

    stdout, stderr = Command(cwd).execute('hg push %s' % clone_url)

    assert """abort: HTTP Error 403: Forbidden"""  in stderr

if __name__ == '__main__':

    #test_clone_wrong_credentials()

    ##test_clone_anonymous_ok()

    #test_push_wrong_path()

    #test_push_wrong_credentials()