def __inject_extras(self, repo_path, baseui, extras={}):

# -*- 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 sys

import shutil

import logging

from os.path import join as jn

from os.path import dirname as dn

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from paste.deploy import appconfig

from pylons import config

from sqlalchemy import engine_from_config

from rhodecode.lib.utils import add_cache

from rhodecode.model import init_model

from rhodecode.model import meta

from rhodecode.model.db import User, Repository

from rhodecode.lib.auth import get_crypt_password

from rhodecode.tests import TESTS_TMP_PATH, NEW_HG_REPO, HG_REPO

from rhodecode.config.environment import load_environment

rel_path = dn(dn(dn(os.path.abspath(__file__))))

conf = appconfig('config:development.ini', relative_to=rel_path)

load_environment(conf.global_conf, conf.local_conf)

add_cache(conf)

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

def get_session():

    engine = engine_from_config(conf, 'sqlalchemy.db1.')

    init_model(engine)

    sa = meta.Session()

    return sa

def create_test_user(force=True):

    print 'creating test user'

    sa = get_session()

    user = sa.query(User).filter(User.username == USER).scalar()

    if force and user is not None:

        print 'removing current user'

        for repo in sa.query(Repository).filter(Repository.user == user).all():

            sa.delete(repo)

        sa.delete(user)

        sa.commit()

    if user is None or force:

        print 'creating new one'

        new_usr = User()

        new_usr.username = USER

        new_usr.password = get_crypt_password(PASS)

        new_usr.email = 'mail@mail.com'

        new_usr.name = 'test'

        new_usr.lastname = 'lasttestname'

        new_usr.active = True

        new_usr.admin = True

        sa.add(new_usr)

        sa.commit()

    print 'done'

def create_test_repo(force=True):

    print 'creating test repo'

    from rhodecode.model.repo import RepoModel

    sa = get_session()

    user = sa.query(User).filter(User.username == USER).scalar()

    if user is None:

        raise Exception('user not found')

    repo = sa.query(Repository).filter(Repository.repo_name == HG_REPO).scalar()

    if repo is None:

        print 'repo not found creating'

        form_data = {'repo_name':HG_REPO,

                     'repo_type':'hg',

                     'private':False,

                     'clone_uri':'' }

        rm = RepoModel(sa)

        rm.base_path = '/home/hg'

        rm.create(form_data, user)

    print 'done'

def set_anonymous_access(enable=True):

    sa = get_session()

    user = sa.query(User).filter(User.username == 'default').one()

    user.active = enable

    sa.add(user)

    sa.commit()

def get_anonymous_access():

    sa = get_session()

    return sa.query(User).filter(User.username == 'default').one().active

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

# TESTS

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

def test_clone_with_credentials(no_errors=False,repo=HG_REPO):

    cwd = path = jn(TESTS_TMP_PATH, 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':repo,

                   'dest':path+_RandomNameSequence().next()}

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

    if no_errors is False:

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

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

if __name__ == '__main__':

    try:

        create_test_user(force=False)

        for i in range(int(sys.argv[2])):

            test_clone_with_credentials(repo=sys.argv[1])

    except Exception,e:

        sys.exit('stop on %s' % e)

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

print 'DEBUG:', DEBUG

    def __wrapp(*args, **kwargs):

        print '###%s###' % func.__name__

            res = func(*args, **kwargs)

        except Exception, e:

            print ('###############\n-'

                   '--%s failed %s--\n'

                   '###############\n' % (func.__name__, e))

            sys.exit()

        print '++OK++'

    print '\tcreating test user'

        print '\tremoving current user'

        print '\tcreating new one'

    print '\tdone'

        print '\trepo not found creating'

    sa.expire(user)

    import time;time.sleep(3)

    print '\tanonymous access is now:', enable

    print '\tchecking if anonymous access is enabled'

        print '\tenabled, disabling it '

    print '\tchecking if anonymous access is enabled'

        print '\tnot enabled, enabling it '

        print '\tdisabling anonymous access'

    print '\tchecking if anonymous access is enabled'

        print '\tenabled, disabling it '

        raise Exception('Failure')

        print '\tmade dirs %s' % jn(path)

@test_wrapp

def get_logs():

    sa = get_session()

    return len(sa.query(UserLog).all())

@test_wrapp

def test_logs(initial):

    sa = get_session()

    logs = sa.query(UserLog).all()

    operations = 7

    if initial + operations != len(logs):

        raise Exception("missing number of logs %s vs %s" % (initial, len(logs)))

    initial_logs = get_logs()

    test_clone_anonymous()

    test_push_wrong_credentials()

    test_logs(initial_logs)