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: Dec 30, 2010

:author: marcink

:copyright: (c) 2013 RhodeCode GmbH, and others.

:license: GPLv3, see LICENSE.md for more details.

"""

import tempfile

import time

from os.path import join as jn

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from kallithea.tests import *

from kallithea.model.db import User, Repository, UserIpMap, CacheInvalidation

from kallithea.model.meta import Session

from kallithea.model.repo import RepoModel

from kallithea.model.user import UserModel

DEBUG = True

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)

        if DEBUG:

            print '*** CMD %s ***' % command

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

        stdout, stderr = p.communicate()

        if DEBUG:

            print 'stdout:', repr(stdout)

            print 'stderr:', repr(stderr)

        return stdout, stderr

def _get_tmp_dir():

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

        stdout, stderr = _add_files_and_push('hg', DEST, user='bad',

                                             passwd='name')

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        stdout, stderr = _add_files_and_push('git', DEST, user='bad',

                                             passwd='name')

        assert 'fatal: Authentication failed' in stderr

    def test_push_back_to_wrong_url_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO, dest=DEST)

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

        stdout, stderr = _add_files_and_push('hg', DEST,

        assert 'HTTP Error 404: Not Found' in stderr

    def test_push_back_to_wrong_url_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

        stdout, stderr = Command('/tmp').execute('git clone', clone_url)

        stdout, stderr = _add_files_and_push('git', DEST,

        assert 'not found' in stderr

    def test_clone_and_create_lock_hg(self):

        # enable locking

        r = Repository.get_by_repo_name(HG_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

        clone_url = _construct_url(HG_REPO)

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

        #check if lock was made

        r = Repository.get_by_repo_name(HG_REPO)

        assert r.locked[0] == User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

    def test_clone_and_create_lock_git(self):

        # enable locking

        r = Repository.get_by_repo_name(GIT_REPO)

        r.enable_locking = True

        Session().add(r)

        Session().commit()

        # clone

## Exit if no app can be loaded.

#need-app = true

## Set lazy mode (load apps in workers instead of master).

#lazy = true

## scaling ##

## set cheaper algorithm to use, if not set default will be used

#cheaper-algo = spare

## minimum number of workers to keep at all times

#cheaper = 1

## number of workers to spawn at startup

#cheaper-initial = 1

## maximum number of workers that can be spawned

#workers = 4

## how many workers should be spawned at a time

#cheaper-step = 1

## COMMON ##

host = 127.0.0.1

## middleware for hosting the WSGI application under a URL prefix

#[filter:proxy-prefix]

#use = egg:PasteDeploy#prefix

#prefix = /<your-prefix>

[app:main]

use = egg:kallithea

## enable proxy prefix middleware

#filter-with = proxy-prefix

full_stack = true

static_files = true

## Available Languages:

## cs de fr hu ja nl_BE pl pt_BR ru sk zh_CN zh_TW

lang =

cache_dir = %(here)s/data

index_dir = %(here)s/data/index

## perform a full repository scan on each server start, this should be

## set to false after first startup, to allow faster server restarts.

#initial_repo_scan = false

initial_repo_scan = true