.. _contributing:

=========================

Contributing to Kallithea

=========================

Kallithea is developed and maintained by its users. Please join us and scratch

your own itch.

Infrastructure

--------------

The main repository is hosted on Our Own Kallithea (aka OOK) at

https://kallithea-scm.org/repos/kallithea/, our self-hosted instance

of Kallithea.

For now, we use Bitbucket_ for `pull requests`_ and `issue tracking`_. The

issue tracker is for tracking bugs, not for support, discussion, or ideas --

please use the `mailing list`_ or :ref:`IRC <readme>` to reach the community.

We use Weblate_ to translate the user interface messages into languages other

than English. Join our project on `Hosted Weblate`_ to help us.

To register, you can use your Bitbucket or GitHub account. See :ref:`translations`

for more details.

Getting started

---------------

To get started with development::

        hg clone https://kallithea-scm.org/repos/kallithea

        cd kallithea

        virtualenv ../kallithea-venv

        source ../kallithea-venv/bin/activate

        pip install --upgrade pip setuptools

        pip install -e .

        gearbox make-config my.ini

        gearbox setup-db -c my.ini --user=user --email=user@example.com --password=password --repos=/tmp

        gearbox serve -c my.ini --reload &

        firefox http://127.0.0.1:5000/

You can also start out by forking https://bitbucket.org/conservancy/kallithea

on Bitbucket_ and create a local clone of your own fork.

Running tests

-------------

After finishing your changes make sure all tests pass cleanly. Install the test

dependencies, then run the testsuite by invoking ``py.test`` from the

project root::

    pip install -r dev_requirements.txt

    py.test

Note that testing on Python 2.6 also requires ``unittest2``.

You can also use ``tox`` to run the tests with all supported Python versions

(currently Python 2.6--2.7).

When running tests, Kallithea uses `kallithea/tests/test.ini` and populates the

SQLite database specified there.

It is possible to avoid recreating the full test database on each invocation of

the tests, thus eliminating the initial delay. To achieve this, run the tests as::

    gearbox serve -c kallithea/tests/test.ini --pid-file=test.pid --daemon

    KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 py.test

    kill -9 $(cat test.pid)

In these commands, the following variables are used::

    KALLITHEA_WHOOSH_TEST_DISABLE=1 - skip whoosh index building and tests

    KALLITHEA_NO_TMP_PATH=1 - disable new temp path for tests, used mostly for testing_vcs_operations

You can run individual tests by specifying their path as argument to py.test.

py.test also has many more options, see `py.test -h`. Some useful options

are::

    -k EXPRESSION         only run tests which match the given substring

                          expression. An expression is a python evaluable

                          expression where all names are substring-matched

                          against test names and their parent classes. Example:

    -x, --exitfirst       exit instantly on first error or failed test.

    --lf                  rerun only the tests that failed at the last run (or

                          all if none failed)

    --ff                  run all tests but run the last failures first. This

                          may re-order tests and thus lead to repeated fixture

                          setup/teardown

    --pdb                 start the interactive Python debugger on errors.

    -s, --capture=no      don't capture stdout (any stdout output will be

                          printed immediately)

Contribution guidelines

-----------------------

Kallithea is GPLv3 and we assume all contributions are made by the

committer/contributor and under GPLv3 unless explicitly stated. We do care a

lot about preservation of copyright and license information for existing code

that is brought into the project.

Contributions will be accepted in most formats -- such as pull requests on

Bitbucket, something hosted on your own Kallithea instance, or patches sent by

email to the `kallithea-general`_ mailing list.

When contributing via Bitbucket, please make your fork of

https://bitbucket.org/conservancy/kallithea/ `non-publishing`_ -- it is one of

the settings on "Repository details" page. This ensures your commits are in

"draft" phase and makes it easier for you to address feedback and for project

maintainers to integrate your changes.

.. _non-publishing: https://www.mercurial-scm.org/wiki/Phases#Publishing_Repository

Make sure to test your changes both manually and with the automatic tests

before posting.

We care about quality and review and keeping a clean repository history. We

might give feedback that requests polishing contributions until they are

"perfect". We might also rebase and collapse and make minor adjustments to your

changes when we apply them.

We try to make sure we have consensus on the direction the project is taking.

Everything non-sensitive should be discussed in public -- preferably on the

mailing list.  We aim at having all non-trivial changes reviewed by at least

one other core developer before pushing. Obvious non-controversial changes will

be handled more casually.

For now we just have one official branch ("default") and will keep it so stable

that it can be (and is) used in production. Experimental changes should live

elsewhere (for example in a pull request) until they are ready.

Coding guidelines

-----------------

We don't have a formal coding/formatting standard. We are currently using a mix

of Mercurial's (https://www.mercurial-scm.org/wiki/CodingStyle), pep8, and

consistency with existing code. Run ``scripts/run-all-cleanup`` before

committing to ensure some basic code formatting consistency.

We support both Python 2.6.x and 2.7.x and nothing else. For now we don't care

about Python 3 compatibility.

We try to support the most common modern web browsers. IE9 is still supported

to the extent it is feasible, IE8 is not.

We primarily support Linux and OS X on the server side but Windows should also work.

HTML templates should use 2 spaces for indentation ... but be pragmatic. We

should use templates cleverly and avoid duplication. We should use reasonable

semantic markup with element classes and IDs that can be used for styling and testing.

We should only use inline styles in places where it really is semantic (such as

``display: none``).

JavaScript must use ``;`` between/after statements. Indentation 4 spaces. Inline

multiline functions should be indented two levels -- one for the ``()`` and one for

``{}``.

Variables holding jQuery objects should be named with a leading ``$``.

Commit messages should have a leading short line summarizing the changes. For

bug fixes, put ``(Issue #123)`` at the end of this line.

Use American English grammar and spelling overall. Use `English title case`_ for

page titles, button labels, headers, and 'labels' for fields in forms.

.. _English title case: https://en.wikipedia.org/wiki/Capitalization#Title_case

Template helpers (that is, everything in ``kallithea.lib.helpers``)

should only be referenced from templates. If you need to call a

helper from the Python code, consider moving the function somewhere

else (e.g. to the model).

Notes on the SQLAlchemy session

^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

Each HTTP request runs inside an independent SQLAlchemy session (as well

as in an independent database transaction). Database model objects

(almost) always belong to a particular SQLAlchemy session, which means

that SQLAlchemy will ensure that they're kept in sync with the database

(but also means that they cannot be shared across requests).

Objects can be added to the session using ``Session().add``, but this is

rarely needed:

* When creating a database object by calling the constructor directly,

  it must explicitly be added to the session.

* When creating an object using a factory function (like

  ``create_repo``), the returned object has already (by convention)

  been added to the session, and should not be added again.

* When getting an object from the session (via ``Session().query`` or

  any of the utility functions that look up objects in the database),

  it's already part of the session, and should not be added again.

  SQLAlchemy monitors attribute modifications automatically for all

  objects it knows about and syncs them to the database.

SQLAlchemy also flushes changes to the database automatically; manually

calling ``Session().flush`` is usually only necessary when the Python

code needs the database to assign an "auto-increment" primary key ID to

a freshly created model object (before flushing, the ID attribute will

be ``None``).

TurboGears2 DebugBar

^^^^^^^^^^^^^^^^^^^^

It is possible to enable the TurboGears2-provided DebugBar_, a toolbar overlayed

over the Kallithea web interface, allowing you to see:

* timing information of the current request, including profiling information

* request data, including GET data, POST data, cookies, headers and environment

  variables

* a list of executed database queries, including timing and result values

DebugBar is only activated when ``debug = true`` is set in the configuration

file. This is important, because the DebugBar toolbar will be visible for all

users, and allow them to see information they should not be allowed to see. Like

is anyway the case for ``debug = true``, do not use this in production!

To enable DebugBar, install ``tgext.debugbar`` and ``kajiki`` (typically via

``pip``) and restart Kallithea (in debug mode).

"Roadmap"

---------

We do not have a road map but are waiting for your contributions. Refer to the

wiki_ for some ideas of places we might want to go -- contributions in these

areas are very welcome.

Thank you for your contribution!

--------------------------------

.. _Weblate: http://weblate.org/

.. _issue tracking: https://bitbucket.org/conservancy/kallithea/issues?status=new&status=open

.. _pull requests: https://bitbucket.org/conservancy/kallithea/pull-requests

.. _bitbucket: http://bitbucket.org/

.. _mailing list: http://lists.sfconservancy.org/mailman/listinfo/kallithea-general

.. _kallithea-general: http://lists.sfconservancy.org/mailman/listinfo/kallithea-general

.. _Hosted Weblate: https://hosted.weblate.org/projects/kallithea/kallithea/

.. _wiki: https://bitbucket.org/conservancy/kallithea/wiki/Home

.. _DebugBar: https://github.com/TurboGears/tgext.debugbar

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

# 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, either version 3 of the License, or

# (at your option) any later version.

#

# 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, see <http://www.gnu.org/licenses/>.

"""

kallithea.tests.other.manual_test_vcs_operations

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

Test suite for making push/pull operations.

Run it in two terminals::

 gearbox serve -c kallithea/tests/test.ini

 KALLITHEA_WHOOSH_TEST_DISABLE=1 KALLITHEA_NO_TMP_PATH=1 py.test kallithea/tests/other/manual_test_vcs_operations.py

You must have git > 1.8.1 for tests to work fine

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 os

import re

import tempfile

import time

import pytest

from tempfile import _RandomNameSequence

from subprocess import Popen, PIPE

from kallithea.tests.base import *

from kallithea.tests.fixture import Fixture

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

HOST = '127.0.0.1:4999'  # test host

fixture = Fixture()

class Command(object):

    def __init__(self, cwd):

        self.cwd = cwd

    def execute(self, cmd, *args, **environ):

        """

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

        """

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

        ignoreReturnCode = environ.pop('ignoreReturnCode', False)

        if DEBUG:

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

        testenv = dict(os.environ)

        testenv['LANG'] = 'en_US.UTF-8'

        testenv['LANGUAGE'] = 'en_US:en'

        testenv.update(environ)

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

        stdout, stderr = p.communicate()

        if DEBUG:

            if stdout:

                print 'stdout:', repr(stdout)

            if stderr:

                print 'stderr:', repr(stderr)

        if not ignoreReturnCode:

            assert p.returncode == 0

        return stdout, stderr

def _get_tmp_dir():

    return tempfile.mkdtemp(prefix='rc_integration_test')

def _construct_url(repo, **kwargs):

    """Return a clone url for the provided repo path.

    Optional named parameters: user, passwd and host."""

    params = {

        'user': TEST_USER_ADMIN_LOGIN,

        'passwd': TEST_USER_ADMIN_PASS,

        'host': HOST,

        'cloned_repo': repo,

    }

    params.update(**kwargs)

    if params['user'] and params['passwd']:

        _url = 'http://%(user)s:%(passwd)s@%(host)s/%(cloned_repo)s' % params

    else:

        _url = 'http://(host)s/%(cloned_repo)s' % params

    return _url

def _add_files_and_push(vcs, DEST, ignoreReturnCode=False, **kwargs):

    """

    Generate some files, add it to DEST repo and push back

    vcs is git or hg and defines what VCS we want to make those files for

    :param vcs:

    :param DEST:

    """

    # commit some stuff into this repo

    cwd = os.path.join(DEST)

    #added_file = '%ssetupążźć.py' % _RandomNameSequence().next()

    added_file = '%ssetup.py' % _RandomNameSequence().next()

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

    Command(cwd).execute('%s add %s' % (vcs, added_file))

    email = 'me@example.com'

    if os.name == 'nt':

        author_str = 'User <%s>' % email

    else:

        author_str = 'User ǝɯɐᴎ <%s>' % email

    for i in xrange(kwargs.get('files_no', 3)):

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

        Command(cwd).execute(cmd)

        if vcs == 'hg':

            cmd = """hg commit -m "committed new %s" -u "%s" "%s" """ % (

                i, author_str, added_file

            )

        elif vcs == 'git':

            cmd = """git commit -m "committed new %s" --author "%s" "%s" """ % (

                i, author_str, added_file

            )

        # git commit needs EMAIL on some machines

        Command(cwd).execute(cmd, EMAIL=email)

    # PUSH it back

    _REPO = None

    if vcs == 'hg':

        _REPO = HG_REPO

    elif vcs == 'git':

        _REPO = GIT_REPO

    clone_url = _construct_url(_REPO, **kwargs)

    if 'clone_url' in kwargs:

        clone_url = kwargs['clone_url']

    stdout = stderr = None

    if vcs == 'hg':

        stdout, stderr = Command(cwd).execute('hg push --verbose', clone_url, ignoreReturnCode=ignoreReturnCode)

    elif vcs == 'git':

        stdout, stderr = Command(cwd).execute('git push --verbose', clone_url, "master", ignoreReturnCode=ignoreReturnCode)

    return stdout, stderr

def set_anonymous_access(enable=True):

    user = User.get_default_user()

    user.active = enable

    Session().commit()

    print '\tanonymous access is now:', enable

    if enable != User.get_default_user().active:

        raise Exception('Cannot set anonymous access')

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

# TESTS

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

def _check_proper_git_push(stdout, stderr):

    #WTF Git stderr is output ?!

    assert 'fatal' not in stderr

    assert 'rejected' not in stderr

    assert 'Pushing to' in stderr

    assert 'master -> master' in stderr

@pytest.mark.usefixtures("test_context_fixture")

class TestVCSOperations(TestController):

    @classmethod

    def setup_class(cls):

        #DISABLE ANONYMOUS ACCESS

        set_anonymous_access(False)

    def setup_method(self, method):

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().commit()

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.unlock(r)

        r.enable_locking = False

        Session().commit()

    def test_clone_hg_repo_by_admin(self):

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

        assert 'adding file changes' in stdout

        assert stderr == ''

    def test_clone_git_repo_by_admin(self):

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

        assert 'Cloning into' in stdout + stderr

        assert stderr == '' or stdout == ''

    def test_clone_wrong_credentials_hg(self):

        clone_url = _construct_url(HG_REPO, passwd='bad!')

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'abort: authorization failed' in stderr

    def test_clone_wrong_credentials_git(self):

        clone_url = _construct_url(GIT_REPO, passwd='bad!')

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'fatal: Authentication failed' in stderr

    def test_clone_git_dir_as_hg(self):

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_hg_repo_as_git(self):

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'not found' in stderr

    def test_clone_non_existing_path_hg(self):

        clone_url = _construct_url('trololo')

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'HTTP Error 404: Not Found' in stderr

    def test_clone_non_existing_path_git(self):

        clone_url = _construct_url('trololo')

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        assert 'not found' in stderr

    def test_push_new_file_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = _construct_url(fork_name).split()[0]

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

        assert 'pushing to' in stdout

        assert 'Repository size' in stdout

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

        # commit some stuff into this repo

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        clone_url = _construct_url(fork_name).split()[0]

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

        print [(x.repo_full_path,x.repo_path) for x in Repository.query()] # TODO: what is this for

        _check_proper_git_push(stdout, stderr)

    def test_push_invalidates_cache_hg(self):

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==HG_REPO).scalar()

        if not key:

            key = CacheInvalidation(HG_REPO, HG_REPO)

            Session().add(key)

        key.cache_active = True

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        clone_url = _construct_url(fork_name).split()[0]

        stdout, stderr = _add_files_and_push('hg', DEST, files_no=1, clone_url=clone_url)

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==fork_name).all()

        assert key == []

    def test_push_invalidates_cache_git(self):

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==GIT_REPO).scalar()

        if not key:

            key = CacheInvalidation(GIT_REPO, GIT_REPO)

            Session().add(key)

        key.cache_active = True

        Session().commit()

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

        # commit some stuff into this repo

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        clone_url = _construct_url(fork_name).split()[0]

        stdout, stderr = _add_files_and_push('git', DEST, files_no=1, clone_url=clone_url)

        _check_proper_git_push(stdout, stderr)

        key = CacheInvalidation.query().filter(CacheInvalidation.cache_key

                                               ==fork_name).all()

        assert key == []

    def test_push_wrong_credentials_hg(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

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

                                             passwd='name', ignoreReturnCode=True)

        assert 'abort: authorization failed' in stderr

    def test_push_wrong_credentials_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

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

                                             passwd='name', ignoreReturnCode=True)

        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)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

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

                                    clone_url='http://%s/tmp' % HOST,

                                    ignoreReturnCode = True)

        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)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

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

                                    clone_url='http://%s/tmp' % HOST,

                                    ignoreReturnCode = True)

        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().commit()

        # clone

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

        #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().commit()

        # clone

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

        #check if lock was made

        r = Repository.get_by_repo_name(GIT_REPO)

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

    def test_clone_after_repo_was_locked_hg(self):

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_clone_after_repo_was_locked_git(self):

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #pull fails since repo is locked

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

        msg = ("""The requested URL returned error: 423""")

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_hg(self):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

        #lock repo

        r = Repository.get_by_repo_name(HG_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        Session().commit()

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #push fails repo is locked by other user !

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

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS,

                                             ignoreReturnCode=True)

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (HG_REPO, TEST_USER_ADMIN_LOGIN))

        assert msg in stderr

    def test_push_on_locked_repo_by_other_user_git(self):

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

        #lock repo

        r = Repository.get_by_repo_name(GIT_REPO)

        # let this user actually push !

        RepoModel().grant_user_permission(repo=r, user=TEST_USER_REGULAR_LOGIN,

                                          perm='repository.write')

        Session().commit()

        Repository.lock(r, User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id)

        #push fails repo is locked by other user !

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

                                             user=TEST_USER_REGULAR_LOGIN,

                                             passwd=TEST_USER_REGULAR_PASS,

                                             ignoreReturnCode=True)

        err = 'Repository `%s` locked by user `%s`' % (GIT_REPO, TEST_USER_ADMIN_LOGIN)

        assert err in stderr

        #TODO: fix this somehow later on Git, Git is stupid and even if we throw

        #back 423 to it, it makes ANOTHER request and we fail there with 405 :/

        msg = ("""abort: HTTP Error 423: Repository `%s` locked by user `%s`"""

                % (GIT_REPO, TEST_USER_ADMIN_LOGIN))

        #msg = "405 Method Not Allowed"

        #assert msg in stderr

    def test_push_unlocks_repository_hg(self):

        # enable locking

        fork_name = '%s_fork%s' % (HG_REPO, _RandomNameSequence().next())

        fixture.create_fork(HG_REPO, fork_name)

        r = Repository.get_by_repo_name(fork_name)

        r.enable_locking = True

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(fork_name)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, DEST)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(fork_name)

        uid = User.get_by_username(TEST_USER_ADMIN_LOGIN).user_id

        assert r.locked[0] == uid

        #push is ok and repo is now unlocked

        stdout, stderr = _add_files_and_push('hg', DEST, clone_url=clone_url.split()[0])

        assert str('remote: Released lock on repo `%s`' % fork_name) in stdout

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(fork_name)

        assert r.locked == [None, None]

    #TODO: fix me ! somehow during tests hooks don't get called on Git

    def test_push_unlocks_repository_git(self):

        # enable locking

        fork_name = '%s_fork%s' % (GIT_REPO, _RandomNameSequence().next())

        fixture.create_fork(GIT_REPO, fork_name)

        r = Repository.get_by_repo_name(fork_name)

        r.enable_locking = True

        Session().commit()

        #clone some temp

        DEST = _get_tmp_dir()

        clone_url = _construct_url(fork_name)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, DEST)

        #check for lock repo after clone

        r = Repository.get_by_repo_name(fork_name)

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

        #push is ok and repo is now unlocked

        stdout, stderr = _add_files_and_push('git', DEST, clone_url=clone_url.split()[0])

        _check_proper_git_push(stdout, stderr)

        assert ('remote: Released lock on repo `%s`' % fork_name) in stderr

        #we need to cleanup the Session Here !

        Session.remove()

        r = Repository.get_by_repo_name(fork_name)

        assert r.locked == [None, None]

    def test_ip_restriction_hg(self):

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = _construct_url(HG_REPO)

            stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

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

        finally:

            #release IP restrictions

            for ip in UserIpMap.query():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to wait for the cache in the server process to expire

        time.sleep(1.5)

        clone_url = _construct_url(HG_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('hg clone', clone_url, _get_tmp_dir())

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

        assert 'adding file changes' in stdout

        assert stderr == ''

    def test_ip_restriction_git(self):

        user_model = UserModel()

        try:

            user_model.add_extra_ip(TEST_USER_ADMIN_LOGIN, '10.10.10.10/32')

            Session().commit()

            clone_url = _construct_url(GIT_REPO)

            stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir(), ignoreReturnCode=True)

            # The message apparently changed in Git 1.8.3, so match it loosely.

            assert re.search(r'\b403\b', stderr)

        finally:

            #release IP restrictions

            for ip in UserIpMap.query():

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to wait for the cache in the server process to expire

        time.sleep(1.5)

        clone_url = _construct_url(GIT_REPO)

        stdout, stderr = Command(tempfile.gettempdir()).execute('git clone', clone_url, _get_tmp_dir())

        assert 'Cloning into' in stdout + stderr

        assert stderr == '' or stdout == ''