feed = Rss201rev2Feed(title=_desc,

                         link=_link,

                         description=_desc,

                         language=language,

                         ttl=ttl)

        for entry in journal[:feed_nr]:

            user = entry.user

            if user is None:

                #fix deleted users

                user = AttributeDict({'short_contact': entry.username,

                                      'email': '',

                                      'full_contact': ''})

            action, action_extra, ico = h.action_parser(entry, feed=True)

            title = "%s - %s %s" % (user.short_contact, action(),

                                    entry.repository.repo_name)

            desc = action_extra()

            _url = None

            if entry.repository is not None:

                _url = h.canonical_url('changelog_home',

                           repo_name=entry.repository.repo_name)

            feed.add_item(title=title,

                          pubdate=entry.action_date,

                          link=_url or h.canonical_url(''),

                          author_email=user.email,

                          author_name=user.full_contact,

                          description=desc)

        response.content_type = feed.mime_type

        return feed.writeString('utf-8')

    @LoginRequired()

    @NotAnonymous()

    def index(self):

        # Return a rendered template

        p = safe_int(request.GET.get('page'), 1)

        c.user = User.get(request.authuser.user_id)

        c.following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        journal = self._get_journal_data(c.following)

        def url_generator(**kw):

            return url.current(filter=c.search_term, **kw)

        c.journal_pager = Page(journal, page=p, items_per_page=20, url=url_generator)

        c.journal_day_aggregate = self._get_daily_aggregate(c.journal_pager)

        if request.environ.get('HTTP_X_PARTIAL_XHR'):

            return render('journal/journal_data.html')

        repos_list = Repository.query(sorted=True) \

            .filter_by(owner_id=request.authuser.user_id).all()

        repos_data = RepoModel().get_repos_as_dict(repos_list=repos_list,

                                                   admin=True)

        #data used to render the grid

        c.data = repos_data

        return render('journal/journal.html')

    @LoginRequired(api_access=True)

    @NotAnonymous()

    def journal_atom(self):

        """

        Produce an atom-1.0 feed via feedgenerator module

        """

        following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        return self._atom_feed(following, public=False)

    @LoginRequired(api_access=True)

    @NotAnonymous()

    def journal_rss(self):

        """

        Produce an rss feed via feedgenerator module

        """

        following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        return self._rss_feed(following, public=False)

    @LoginRequired()

    @NotAnonymous()

    def toggle_following(self):

        user_id = request.POST.get('follows_user_id')

        if user_id:

            try:

                self.scm_model.toggle_following_user(user_id,

                                            request.authuser.user_id)

                return 'ok'

            except Exception:

                log.error(traceback.format_exc())

                raise HTTPBadRequest()

        repo_id = request.POST.get('follows_repository_id')

        if repo_id:

            try:

                self.scm_model.toggle_following_repo(repo_id,

                                            request.authuser.user_id)

                return 'ok'

            except Exception:

                log.error(traceback.format_exc())

                raise HTTPBadRequest()

        raise HTTPBadRequest()

    @LoginRequired()

    def public_journal(self):

        # Return a rendered template

        p = safe_int(request.GET.get('page'), 1)

        c.following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        journal = self._get_journal_data(c.following)

        c.journal_pager = Page(journal, page=p, items_per_page=20)

        c.journal_day_aggregate = self._get_daily_aggregate(c.journal_pager)

        if request.environ.get('HTTP_X_PARTIAL_XHR'):

            return render('journal/journal_data.html')

        return render('journal/public_journal.html')

    @LoginRequired(api_access=True)

    def public_journal_atom(self):

        """

        Produce an atom-1.0 feed via feedgenerator module

        """

        c.following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        return self._atom_feed(c.following)

    @LoginRequired(api_access=True)

    def public_journal_rss(self):

        """

        Produce an rss2 feed via feedgenerator module

        """

        c.following = UserFollowing.query() \

            .filter(UserFollowing.user_id == request.authuser.user_id) \

            .options(joinedload(UserFollowing.follows_repository)) \

            .all()

        return self._rss_feed(c.following)

# -*- 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.lib.auth

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

authentication and permission libraries

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 4, 2010

:author: marcink

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

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

"""

import os

import logging

import traceback

import hashlib

import itertools

import collections

from decorator import decorator

from tg import request, session

from tg.i18n import ugettext as _

from webhelpers.pylonslib import secure_form

from sqlalchemy.orm.exc import ObjectDeletedError

from sqlalchemy.orm import joinedload

from webob.exc import HTTPFound, HTTPBadRequest, HTTPForbidden, HTTPMethodNotAllowed

from kallithea import __platform__, is_windows, is_unix

from kallithea.config.routing import url

from kallithea.lib.vcs.utils.lazy import LazyProperty

from kallithea.model.meta import Session

from kallithea.model.user import UserModel

from kallithea.model.db import User, Repository, Permission, \

    UserToPerm, UserGroupRepoToPerm, UserGroupToPerm, UserGroupMember, \

    RepoGroup, UserGroupRepoGroupToPerm, UserIpMap, UserGroupUserGroupToPerm, \

    UserGroup, UserApiKeys

from kallithea.lib.utils2 import safe_str, safe_unicode, aslist

from kallithea.lib.utils import get_repo_slug, get_repo_group_slug, \

    get_user_group_slug, conditional_cache

from kallithea.lib.caching_query import FromCache

log = logging.getLogger(__name__)

class PasswordGenerator(object):

    """

    This is a simple class for generating password from different sets of

    characters

    usage::

        passwd_gen = PasswordGenerator()

        #print 8-letter password containing only big and small letters

            of alphabet

        passwd_gen.gen_password(8, passwd_gen.ALPHABETS_BIG_SMALL)

    """

    ALPHABETS_NUM = r'''1234567890'''

    ALPHABETS_SMALL = r'''qwertyuiopasdfghjklzxcvbnm'''

    ALPHABETS_BIG = r'''QWERTYUIOPASDFGHJKLZXCVBNM'''

    ALPHABETS_SPECIAL = r'''`-=[]\;',./~!@#$%^&*()_+{}|:"<>?'''

    ALPHABETS_FULL = ALPHABETS_BIG + ALPHABETS_SMALL \

        + ALPHABETS_NUM + ALPHABETS_SPECIAL

    ALPHABETS_ALPHANUM = ALPHABETS_BIG + ALPHABETS_SMALL + ALPHABETS_NUM

    ALPHABETS_BIG_SMALL = ALPHABETS_BIG + ALPHABETS_SMALL

    ALPHABETS_ALPHANUM_BIG = ALPHABETS_BIG + ALPHABETS_NUM

    ALPHABETS_ALPHANUM_SMALL = ALPHABETS_SMALL + ALPHABETS_NUM

    def gen_password(self, length, alphabet=ALPHABETS_FULL):

        assert len(alphabet) <= 256, alphabet

        l = []

        while len(l) < length:

            i = ord(os.urandom(1))

            if i < len(alphabet):

                l.append(alphabet[i])

        return ''.join(l)

def get_crypt_password(password):

    """

    Cryptographic function used for password hashing based on pybcrypt

    or Python's own OpenSSL wrapper on windows

    :param password: password to hash

    """

    if is_windows:

        return hashlib.sha256(password).hexdigest()

    elif is_unix:

        import bcrypt

        return bcrypt.hashpw(safe_str(password), bcrypt.gensalt(10))

    else:

        raise Exception('Unknown or unsupported platform %s' \

                        % __platform__)

def check_password(password, hashed):

    """

    Checks matching password with it's hashed value, runs different

    implementation based on platform it runs on

    :param password: password

    :param hashed: password in hashed form

    """

    if is_windows:

        return hashlib.sha256(password).hexdigest() == hashed

    elif is_unix:

        import bcrypt

        return bcrypt.checkpw(safe_str(password), safe_str(hashed))

    else:

        raise Exception('Unknown or unsupported platform %s' \

                        % __platform__)

def _cached_perms_data(user_id, user_is_admin, user_inherit_default_permissions,

                       explicit, algo):

    RK = 'repositories'

    GK = 'repositories_groups'

    UK = 'user_groups'

    GLOBAL = 'global'

    PERM_WEIGHTS = Permission.PERM_WEIGHTS

    permissions = {RK: {}, GK: {}, UK: {}, GLOBAL: set()}

    def _choose_perm(new_perm, cur_perm):

        new_perm_val = PERM_WEIGHTS[new_perm]

        cur_perm_val = PERM_WEIGHTS[cur_perm]

    @property

    def user_groups_admin(self):

        """

        Returns list of user groups you're an admin of

        """

        return [x[0] for x in self.permissions['user_groups'].iteritems()

                if x[1] == 'usergroup.admin']

    @staticmethod

    def check_ip_allowed(user, ip_addr):

        """

        Check if the given IP address (a `str`) is allowed for the given

        user (an `AuthUser` or `db.User`).

        """

        allowed_ips = AuthUser.get_allowed_ips(user.user_id, cache=True,

            inherit_from_default=user.inherit_default_permissions)

        if check_ip_access(source_ip=ip_addr, allowed_ips=allowed_ips):

            log.debug('IP:%s is in range of %s', ip_addr, allowed_ips)

            return True

        else:

            log.info('Access for IP:%s forbidden, '

                     'not in %s' % (ip_addr, allowed_ips))

            return False

    def __repr__(self):

        return "<AuthUser('id:%s[%s] auth:%s')>" \

            % (self.user_id, self.username, (self.is_authenticated or self.is_default_user))

    def to_cookie(self):

        """ Serializes this login session to a cookie `dict`. """

        return {

            'user_id': self.user_id,

            'is_external_auth': self.is_external_auth,

        }

    @staticmethod

    def from_cookie(cookie):

        """

        Deserializes an `AuthUser` from a cookie `dict`.

        """

        au = AuthUser(

            user_id=cookie.get('user_id'),

            is_external_auth=cookie.get('is_external_auth', False),

        )

        au.is_authenticated = True

        return au

    @classmethod

    def get_allowed_ips(cls, user_id, cache=False, inherit_from_default=False):

        _set = set()

        if inherit_from_default:

            default_ips = UserIpMap.query().filter(UserIpMap.user ==

                                            User.get_default_user(cache=True))

            if cache:

                default_ips = default_ips.options(FromCache("sql_cache_short",

                                                  "get_user_ips_default"))

            # populate from default user

            for ip in default_ips:

                try:

                    _set.add(ip.ip_addr)

                except ObjectDeletedError:

                    # since we use heavy caching sometimes it happens that we get

                    # deleted objects here, we just skip them

                    pass

        user_ips = UserIpMap.query().filter(UserIpMap.user_id == user_id)

        if cache:

            user_ips = user_ips.options(FromCache("sql_cache_short",

                                                  "get_user_ips_%s" % user_id))

        for ip in user_ips:

            try:

                _set.add(ip.ip_addr)

            except ObjectDeletedError:

                # since we use heavy caching sometimes it happens that we get

                # deleted objects here, we just skip them

                pass

        return _set or set(['0.0.0.0/0', '::/0'])

def set_available_permissions(config):

    """

    This function will propagate globals with all available defined

    permission given in db. We don't want to check each time from db for new

    permissions since adding a new permission also requires application restart

    ie. to decorate new views with the newly created permission

    :param config: current config instance

    """

    log.info('getting information about all available permissions')

    try:

        config['available_permissions'] = [x.permission_name for x in all_perms]

    finally:

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

# CHECK DECORATORS

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

def _redirect_to_login(message=None):

    """Return an exception that must be raised. It will redirect to the login

    page which will redirect back to the current URL after authentication.

    The optional message will be shown in a flash message."""

    from kallithea.lib import helpers as h

    if message:

        h.flash(h.literal(message), category='warning')

    p = request.path_qs

    log.debug('Redirecting to login page, origin: %s', p)

    return HTTPFound(location=url('login_home', came_from=p))

# Use as decorator

class LoginRequired(object):

    """Client must be logged in as a valid User (but the "default" user,

    if enabled, is considered valid), or we'll redirect to the login page.

    Also checks that IP address is allowed, and if using API key instead

    of regular cookie authentication, checks that API key access is allowed

    (based on `api_access` parameter and the API view whitelist).

    """

    def __init__(self, api_access=False):

        self.api_access = api_access

    def __call__(self, func):

        return decorator(self.__wrapper, func)

    def __wrapper(self, func, *fargs, **fkwargs):

        controller = fargs[0]

        user = request.authuser

        loc = "%s:%s" % (controller.__class__.__name__, func.__name__)

        log.debug('Checking access for user %s @ %s', user, loc)

        if not AuthUser.check_ip_allowed(user, request.ip_addr):

            raise _redirect_to_login(_('IP %s not allowed') % request.ip_addr)

        # Check if we used an API key to authenticate.

        api_key = user.authenticating_api_key

        if api_key is not None:

            # Check that controller is enabled for API key usage.

            if not self.api_access and not allowed_api_access(loc, api_key=api_key):

                # controller does not allow API access

                log.warning('API access to %s is not allowed', loc)

                raise HTTPForbidden()

            log.info('user %s authenticated with API key ****%s @ %s',

                     user, api_key[-4:], loc)

            return func(*fargs, **fkwargs)

        # CSRF protection: Whenever a request has ambient authority (whether

        # through a session cookie or its origin IP address), it must include

        # the correct token, unless the HTTP method is GET or HEAD (and thus

        # guaranteed to be side effect free. In practice, the only situation

        # where we allow side effects without ambient authority is when the

        # authority comes from an API key; and that is handled above.

        if request.method not in ['GET', 'HEAD']:

            token = request.POST.get(secure_form.token_key)

            if not token or token != secure_form.authentication_token():

                log.error('CSRF check failed')

                raise HTTPForbidden()

        # regular user authentication

        if user.is_authenticated or user.is_default_user:

            log.info('user %s authenticated with regular auth @ %s', user, loc)

            return func(*fargs, **fkwargs)

        else:

            log.warning('user %s NOT authenticated with regular auth @ %s', user, loc)

            raise _redirect_to_login()

# Use as decorator

class NotAnonymous(object):

    """Ensures that client is not logged in as the "default" user, and

    redirects to the login page otherwise. Must be used together with

    LoginRequired."""

    def __call__(self, func):

        return decorator(self.__wrapper, func)

    def __wrapper(self, func, *fargs, **fkwargs):

        cls = fargs[0]

        user = request.authuser

        log.debug('Checking that user %s is not anonymous @%s', user.username, cls)

        if user.is_default_user:

            raise _redirect_to_login(_('You need to be a registered user to '

                                       'perform this action'))

        else:

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        expected = 'Missing non optional `repoid` arg in JSON DATA'

        self._compare_error(id_, expected, given=response.body)

    def test_api_args_is_null(self):

        id_, params = _build_data(self.apikey, 'get_users', )

        params = params.replace('"args": {}', '"args": null')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_is_bad(self):

        id_, params = _build_data(self.apikey, 'get_users', )

        params = params.replace('"args": {}', '"args": 1')

        response = api_call(self, params)

        assert response.status == '200 OK'

    def test_api_args_different_args(self):

        import string

        expected = {

            'ascii_letters': string.ascii_letters,

            'ws': string.whitespace,

            'printables': string.printable

        }

        id_, params = _build_data(self.apikey, 'test', args=expected)

        response = api_call(self, params)

        assert response.status == '200 OK'

        self._compare_ok(id_, expected, response.body)

    def test_api_get_users(self):

        id_, params = _build_data(self.apikey, 'get_users', )

        response = api_call(self, params)

        ret_all = []

        _users = User.query().filter_by(is_default_user=False) \

            .order_by(User.username).all()

        for usr in _users:

            ret = usr.get_api_data()

            ret_all.append(jsonify(ret))

        expected = ret_all

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_user(self):

        id_, params = _build_data(self.apikey, 'get_user',

                                  userid=TEST_USER_ADMIN_LOGIN)

        response = api_call(self, params)

        usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = usr.get_api_data()

        ret['permissions'] = AuthUser(dbuser=usr).permissions

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_user_that_does_not_exist(self):

        id_, params = _build_data(self.apikey, 'get_user',

                                  userid='trololo')

        response = api_call(self, params)

        expected = "user `%s` does not exist" % 'trololo'

        self._compare_error(id_, expected, given=response.body)

    def test_api_get_user_without_giving_userid(self):

        id_, params = _build_data(self.apikey, 'get_user')

        response = api_call(self, params)

        usr = User.get_by_username(TEST_USER_ADMIN_LOGIN)

        ret = usr.get_api_data()

        ret['permissions'] = AuthUser(dbuser=usr).permissions

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_user_without_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user')

        response = api_call(self, params)

        usr = User.get_by_username(self.TEST_USER_LOGIN)

        ret = usr.get_api_data()

        ret['permissions'] = AuthUser(dbuser=usr).permissions

        expected = ret

        self._compare_ok(id_, expected, given=response.body)

    def test_api_get_user_with_giving_userid_non_admin(self):

        id_, params = _build_data(self.apikey_regular, 'get_user',

                                  userid=self.TEST_USER_LOGIN)

        response = api_call(self, params)

        expected = 'userid is not the same as your user'

        self._compare_error(id_, expected, given=response.body)

    def test_api_pull(self):

        repo_name = u'test_pull'

        r = fixture.create_repo(repo_name, repo_type=self.REPO_TYPE)

        r.clone_uri = os.path.join(Ui.get_by_key('paths', '/').ui_value, self.REPO)

        id_, params = _build_data(self.apikey, 'pull',

                                  repoid=repo_name,)

        response = api_call(self, params)

        expected = {'msg': 'Pulled from `%s`' % repo_name,

                    'repository': repo_name}

        self._compare_ok(id_, expected, given=response.body)

        fixture.destroy_repo(repo_name)

    def test_api_pull_error(self):

        id_, params = _build_data(self.apikey, 'pull',

                                  repoid=self.REPO, )

        response = api_call(self, params)

        expected = 'Unable to pull changes from `%s`' % self.REPO

        self._compare_error(id_, expected, given=response.body)

    def test_api_rescan_repos(self):

        id_, params = _build_data(self.apikey, 'rescan_repos')

        response = api_call(self, params)

        expected = {'added': [], 'removed': []}

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(ScmModel, 'repo_scan', crash)

    def test_api_rescann_error(self):

        id_, params = _build_data(self.apikey, 'rescan_repos', )

        response = api_call(self, params)

        expected = 'Error occurred during rescan repositories action'

        self._compare_error(id_, expected, given=response.body)

    def test_api_invalidate_cache(self):

        repo = RepoModel().get_by_repo_name(self.REPO)

        repo.scm_instance_cached()  # seed cache

        id_, params = _build_data(self.apikey, 'invalidate_cache',

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = {

            'msg': "Cache for repository `%s` was invalidated" % (self.REPO,),

            'repository': self.REPO

        }

        self._compare_ok(id_, expected, given=response.body)

    @mock.patch.object(ScmModel, 'mark_for_invalidation', crash)

    def test_api_invalidate_cache_error(self):

        id_, params = _build_data(self.apikey, 'invalidate_cache',

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = 'Error occurred during cache invalidation action'

        self._compare_error(id_, expected, given=response.body)

    def test_api_invalidate_cache_regular_user_no_permission(self):

        repo = RepoModel().get_by_repo_name(self.REPO)

        repo.scm_instance_cached() # seed cache

        id_, params = _build_data(self.apikey_regular, 'invalidate_cache',

                                  repoid=self.REPO)

        response = api_call(self, params)

        expected = "repository `%s` does not exist" % (self.REPO,)

        self._compare_error(id_, expected, given=response.body)

    def test_api_lock_repo_lock_acquire(self):

        id_, params = _build_data(self.apikey, 'lock',

                                  userid=TEST_USER_ADMIN_LOGIN,

                                  repoid=self.REPO,

                                  locked=True)

        response = api_call(self, params)

        expected = {

            'repo': self.REPO, 'locked': True,

            'locked_since': response.json['result']['locked_since'],

            'locked_by': TEST_USER_ADMIN_LOGIN,

            'lock_state_changed': True,

            'msg': ('User `%s` set lock state for repo `%s` to `%s`'

                    % (TEST_USER_ADMIN_LOGIN, self.REPO, True))

        }

        self._compare_ok(id_, expected, given=response.body)

    def test_api_lock_repo_lock_acquire_by_non_admin(self):

        repo_name = u'api_delete_me'

        fixture.create_repo(repo_name, repo_type=self.REPO_TYPE,

                            cur_user=self.TEST_USER_LOGIN)

        try:

            id_, params = _build_data(self.apikey_regular, 'lock',

                                      repoid=repo_name,

                                      locked=True)

            response = api_call(self, params)

            expected = {

                'repo': repo_name,

                'locked': True,

                                  'pullrequest_desc': 'description',

                                  'owner': TEST_USER_ADMIN_LOGIN,

                                  '_authentication_token': self.authentication_token(),

                                  'review_members': [str(invalid_user_id)],

                                 },

                                 status=400)

        response.mustcontain('Invalid reviewer "%s" specified' % invalid_user_id)

    def test_iteration_refs(self):

        # Repo graph excerpt:

        #   o   fb95b340e0d0 webvcs

        #  /:

        # o :   41d2568309a0 default

        # : :

        # : o   5ec21f21aafe webvcs

        # : :

        # : o   9e6119747791 webvcs

        # : :

        # o :   3d1091ee5a53 default

        # :/

        # o     948da46b29c1 default

        self.log_user()

        # create initial PR

        response = self.app.post(

            url(controller='pullrequests', action='create', repo_name=HG_REPO),

            {

                'org_repo': HG_REPO,

                'org_ref': 'rev:9e6119747791:9e6119747791ff886a5abe1193a730b6bf874e1c',

                'other_repo': HG_REPO,

                'other_ref': 'branch:default:3d1091ee5a533b1f4577ec7d8a226bb315fb1336',

                'pullrequest_title': 'title',

                'pullrequest_desc': 'description',

                '_authentication_token': self.authentication_token(),

            },

            status=302)

        pr1_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))

        pr1 = PullRequest.get(pr1_id)

        assert pr1.org_ref == 'branch:webvcs:9e6119747791ff886a5abe1193a730b6bf874e1c'

        assert pr1.other_ref == 'branch:default:948da46b29c125838a717f6a8496eb409717078d'

        Session().rollback() # invalidate loaded PR objects before issuing next request.

        # create PR 2 (new iteration with same ancestor)

        response = self.app.post(

            url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr1_id),

            {

                'updaterev': '5ec21f21aafe95220f1fc4843a4a57c378498b71',

                'pullrequest_title': 'title',

                'pullrequest_desc': 'description',

                'owner': TEST_USER_REGULAR_LOGIN,

                '_authentication_token': self.authentication_token(),

             },

             status=302)

        pr2_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))

        pr1 = PullRequest.get(pr1_id)

        pr2 = PullRequest.get(pr2_id)

        assert pr2_id != pr1_id

        assert pr1.status == PullRequest.STATUS_CLOSED

        assert pr2.org_ref == 'branch:webvcs:5ec21f21aafe95220f1fc4843a4a57c378498b71'

        assert pr2.other_ref == pr1.other_ref

        Session().rollback() # invalidate loaded PR objects before issuing next request.

        # create PR 3 (new iteration with new ancestor)

        response = self.app.post(

            url(controller='pullrequests', action='post', repo_name=HG_REPO, pull_request_id=pr2_id),

            {

                'updaterev': 'fb95b340e0d03fa51f33c56c991c08077c99303e',

                'pullrequest_title': 'title',

                'pullrequest_desc': 'description',

                'owner': TEST_USER_REGULAR_LOGIN,

                '_authentication_token': self.authentication_token(),

             },

             status=302)

        pr3_id = int(re.search('/pull-request/(\d+)/', response.location).group(1))

        pr2 = PullRequest.get(pr2_id)

        pr3 = PullRequest.get(pr3_id)

        assert pr3_id != pr2_id

        assert pr2.status == PullRequest.STATUS_CLOSED

        assert pr3.org_ref == 'branch:webvcs:fb95b340e0d03fa51f33c56c991c08077c99303e'

        assert pr3.other_ref == 'branch:default:41d2568309a05f422cffb8008e599d385f8af439'

@pytest.mark.usefixtures("test_context_fixture") # apply fixture for all test methods

class TestPullrequestsGetRepoRefs(TestController):

    def setup_method(self, method):

        self.repo_name = u'main'

        repo = fixture.create_repo(self.repo_name, repo_type='hg')

        self.repo_scm_instance = repo.scm_instance

        self.c = PullrequestsController()

    def teardown_method(self, method):

        fixture.destroy_repo(u'main')

        Session.remove()

    def test_repo_refs_empty_repo(self):

        # empty repo with no commits, no branches, no bookmarks, just one tag

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

        assert default == 'tag:null:0000000000000000000000000000000000000000'

    def test_repo_refs_one_commit_no_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                content='line1\n', message='commit1', vcs_type='hg',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

        assert default == 'branch:default:%s' % cs0.raw_id

        assert ([('branch:default:%s' % cs0.raw_id, 'default (current tip)')],

                'Branches') in refs

    def test_repo_refs_one_commit_rev_hint(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                content='line1\n', message='commit1', vcs_type='hg',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)

        expected = 'branch:default:%s' % cs0.raw_id

        assert default == expected

        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_no_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                content='line1\n', message='commit1', vcs_type='hg',

                parent=None, newfile=True)

        cs1 = fixture.commit_change(self.repo_name, filename='file2',

                content='line2\n', message='commit2', vcs_type='hg',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance)

        expected = 'branch:default:%s' % cs1.raw_id

        assert default == expected

        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_rev_hints(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                content='line1\n', message='commit1', vcs_type='hg',

                parent=None, newfile=True)

        cs1 = fixture.commit_change(self.repo_name, filename='file2',

                content='line2\n', message='commit2', vcs_type='hg',

                parent=None, newfile=True)

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs0.raw_id)

        expected = 'rev:%s:%s' % (cs0.raw_id, cs0.raw_id)

        assert default == expected

        assert ([(expected, 'Changeset: %s' % cs0.raw_id[0:12])], 'Special') in refs

        assert ([('branch:default:%s' % cs1.raw_id, 'default (current tip)')], 'Branches') in refs

        refs, default = self.c._get_repo_refs(self.repo_scm_instance, rev=cs1.raw_id)

        expected = 'branch:default:%s' % cs1.raw_id

        assert default == expected

        assert ([(expected, 'default (current tip)')], 'Branches') in refs

    def test_repo_refs_two_commits_branch_hint(self):

        cs0 = fixture.commit_change(self.repo_name, filename='file1',

                content='line1\n', message='commit1', vcs_type='hg',

                parent=None, newfile=True)