conditions=dict(method=["POST"], function=check_repo))

    rmap.connect("edit_repo_fields", "/{repo_name:.*?}/settings/fields",

                 controller='admin/repos', action="edit_fields",

                 conditions=dict(method=["GET"], function=check_repo))

    rmap.connect('create_repo_fields', "/{repo_name:.*?}/settings/fields/new",

                 controller='admin/repos', action="create_repo_field",

                 conditions=dict(method=["POST"], function=check_repo))

    rmap.connect('delete_repo_fields', "/{repo_name:.*?}/settings/fields/{field_id}/delete",

                 controller='admin/repos', action="delete_repo_field",

                 conditions=dict(method=["POST"], function=check_repo))

    rmap.connect("edit_repo_advanced", "/{repo_name:.*?}/settings/advanced",

                 controller='admin/repos', action="edit_advanced",

                 conditions=dict(method=["GET"], function=check_repo))

    rmap.connect("edit_repo_advanced_journal", "/{repo_name:.*?}/settings/advanced/journal",

                 controller='admin/repos', action="edit_advanced_journal",

                 conditions=dict(method=["POST"], function=check_repo))

    rmap.connect("edit_repo_advanced_fork", "/{repo_name:.*?}/settings/advanced/fork",

                 controller='admin/repos', action="edit_advanced_fork",

                 conditions=dict(method=["POST"], function=check_repo))

    rmap.connect("edit_repo_remote", "/{repo_name:.*?}/settings/remote",

                 controller='admin/repos', action="edit_remote",

                 conditions=dict(method=["GET"], function=check_repo))

    rmap.connect("edit_repo_remote_update", "/{repo_name:.*?}/settings/remote",

                 controller='admin/repos', action="edit_remote",

                 conditions=dict(method=["POST"], function=check_repo))

    rmap.connect("edit_repo_statistics", "/{repo_name:.*?}/settings/statistics",

                 controller='admin/repos', action="edit_statistics",

                 conditions=dict(method=["GET"], function=check_repo))

    rmap.connect("edit_repo_statistics_update", "/{repo_name:.*?}/settings/statistics",

                 controller='admin/repos', action="edit_statistics",

                 conditions=dict(method=["POST"], function=check_repo))

    # still working url for backward compat.

    rmap.connect('raw_changeset_home_depraced',

                 '/{repo_name:.*?}/raw-changeset/{revision}',

                 controller='changeset', action='changeset_raw',

                 revision='tip', conditions=dict(function=check_repo))

    ## new URLs

    rmap.connect('changeset_raw_home',

                 '/{repo_name:.*?}/changeset-diff/{revision}',

                 controller='changeset', action='changeset_raw',

        """

        Mark given repository as a fork of another

        :param repo_name:

        """

        try:

            fork_id = request.POST.get('id_fork_of')

            repo = ScmModel().mark_as_fork(repo_name, fork_id,

                                           request.authuser.username)

            fork = repo.fork.repo_name if repo.fork else _('Nothing')

            Session().commit()

            h.flash(_('Marked repository %s as fork of %s') % (repo_name, fork),

                    category='success')

        except RepositoryError as e:

            log.error(traceback.format_exc())

            h.flash(e, category='error')

        except Exception as e:

            log.error(traceback.format_exc())

            h.flash(_('An error occurred during this operation'),

                    category='error')

        raise HTTPFound(location=url('edit_repo_advanced', repo_name=repo_name))

    @HasRepoPermissionLevelDecorator('admin')

    def edit_remote(self, repo_name):

        c.repo_info = self._load_repo()

        c.active = 'remote'

        if request.POST:

            try:

                ScmModel().pull_changes(repo_name, request.authuser.username, request.ip_addr)

                h.flash(_('Pulled from remote location'), category='success')

            except Exception as e:

                log.error(traceback.format_exc())

                h.flash(_('An error occurred during pull from remote location'),

                        category='error')

            raise HTTPFound(location=url('edit_repo_remote', repo_name=c.repo_name))

        return render('admin/repos/repo_edit.html')

    @HasRepoPermissionLevelDecorator('admin')

    def edit_statistics(self, repo_name):

        c.repo_info = self._load_repo()

        repo = c.repo_info.scm_instance

        if c.repo_info.stats:

            # this is on what revision we ended up so we add +1 for count

            last_rev = c.repo_info.stats.stat_on_revision + 1

        else:

            last_rev = 0

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

Database Models for Kallithea

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

:author: marcink

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

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

"""

import base64

import collections

import datetime

import functools

import hashlib

import logging

import os

import time

import traceback

import ipaddr

import sqlalchemy

from sqlalchemy import Boolean, Column, DateTime, Float, ForeignKey, Index, Integer, LargeBinary, String, Unicode, UnicodeText, UniqueConstraint

from sqlalchemy.ext.hybrid import hybrid_property

from sqlalchemy.orm import class_mapper, joinedload, relationship, validates

from tg.i18n import lazy_ugettext as _

from webob.exc import HTTPNotFound

import kallithea

from kallithea.lib import ext_json

from kallithea.lib.exceptions import DefaultUserException

from kallithea.lib.utils2 import (Optional, ascii_bytes, aslist, get_changeset_safe, get_clone_url, remove_prefix, safe_bytes, safe_int, safe_str, str2bool,

                                  urlreadable)

from kallithea.lib.vcs import get_backend

from kallithea.lib.vcs.backends.base import EmptyChangeset

from kallithea.lib.vcs.utils.helpers import get_scm

from kallithea.model.meta import Base, Session

URL_SEP = '/'

log = logging.getLogger(__name__)

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

# BASE CLASSES

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

    @property

    def groups_with_parents(self):

        groups = []

        group = self.group

        while group is not None:

            groups.append(group)

            group = group.parent_group

            assert group not in groups, group # avoid recursion on bad db content

        groups.reverse()

        return groups

    @property

    def repo_full_path(self):

        """

        Returns base full path for the repository - where it actually

        exists on a filesystem.

        """

        p = [kallithea.CONFIG['base_path']]

        # we need to split the name by / since this is how we store the

        # names in the database, but that eventually needs to be converted

        # into a valid system path

        p += self.repo_name.split(URL_SEP)

        return os.path.join(*p)

    def get_new_name(self, repo_name):

        """

        returns new full repository name based on assigned group and new new

        :param group_name:

        """

        path_prefix = self.group.full_path_splitted if self.group else []

        return URL_SEP.join(path_prefix + [repo_name])

    @property

    def _ui(self):

        """

        Creates an db based ui object for this repository

        """

        from kallithea.lib.utils import make_ui

        return make_ui()

    @classmethod

    def is_valid(cls, repo_name):

        """

        returns True if given repo name is a valid filesystem repository

        :param cls:

        :param repo_name:

        grouped = {}

        for stat in statuses.all():

            pr_id = pr_nice_id = pr_repo = None

            if stat.pull_request:

                pr_id = stat.pull_request.pull_request_id

                pr_nice_id = PullRequest.make_nice_id(pr_id)

                pr_repo = stat.pull_request.other_repo.repo_name

            grouped[stat.revision] = [str(stat.status), stat.status_lbl,

                                      pr_id, pr_repo, pr_nice_id,

                                      stat.author]

        return grouped

    def _repo_size(self):

        from kallithea.lib import helpers as h

        log.debug('calculating repository size...')

        return h.format_byte_size(self.scm_instance.size)

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

    # SCM CACHE INSTANCE

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

    def set_invalidate(self):

        """

        """

    @property

    def scm_instance(self):

        if self._scm_instance is None:

        return self._scm_instance

    def scm_instance_no_cache(self):

        repo_full_path = self.repo_full_path

        alias = get_scm(repo_full_path)[0]

        log.debug('Creating instance of %s repository from %s',

                  alias, self.repo_full_path)

        backend = get_backend(alias)

        if alias == 'hg':

        else:

    def __json__(self):

        return dict(

            repo_id=self.repo_id,

            repo_name=self.repo_name,

            landing_rev=self.landing_rev,

        )

class RepoGroup(Base, BaseDbModel):

    __tablename__ = 'groups'

    __table_args__ = (

        _table_args_default_dict,

    )

    SEP = ' » '

    group_id = Column(Integer(), primary_key=True)

    group_name = Column(Unicode(255), nullable=False, unique=True) # full path

    parent_group_id = Column('group_parent_id', Integer(), ForeignKey('groups.group_id'), nullable=True)

    group_description = Column(Unicode(10000), nullable=False)

    owner_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

class UserFollowing(Base, BaseDbModel):

    __tablename__ = 'user_followings'

    __table_args__ = (

        UniqueConstraint('user_id', 'follows_repository_id', name='uq_user_followings_user_repo'),

        UniqueConstraint('user_id', 'follows_user_id', name='uq_user_followings_user_user'),

        _table_args_default_dict,

    )

    user_following_id = Column(Integer(), primary_key=True)

    user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=False)

    follows_repository_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=True)

    follows_user_id = Column(Integer(), ForeignKey('users.user_id'), nullable=True)

    follows_from = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    user = relationship('User', primaryjoin='User.user_id==UserFollowing.user_id')

    follows_user = relationship('User', primaryjoin='User.user_id==UserFollowing.follows_user_id')

    follows_repository = relationship('Repository', order_by=lambda: sqlalchemy.func.lower(Repository.repo_name))

    @classmethod

    def get_repo_followers(cls, repo_id):

        return cls.query().filter(cls.follows_repository_id == repo_id)

class ChangesetComment(Base, BaseDbModel):

    __tablename__ = 'changeset_comments'

    __table_args__ = (

        Index('cc_revision_idx', 'revision'),

        Index('cc_pull_request_id_idx', 'pull_request_id'),

        _table_args_default_dict,

    )

    comment_id = Column(Integer(), primary_key=True)

    repo_id = Column(Integer(), ForeignKey('repositories.repo_id'), nullable=False)

    revision = Column(String(40), nullable=True)

    pull_request_id = Column(Integer(), ForeignKey('pull_requests.pull_request_id'), nullable=True)

    line_no = Column(Unicode(10), nullable=True)

    f_path = Column(Unicode(1000), nullable=True)

    author_id = Column('user_id', Integer(), ForeignKey('users.user_id'), nullable=False)

    text = Column(UnicodeText(), nullable=False)

    created_on = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    modified_at = Column(DateTime(timezone=False), nullable=False, default=datetime.datetime.now)

    author = relationship('User')

    repo = relationship('Repository')

    # status_change is frequently used directly in templates - make it a lazy

    # join to avoid fetching each related ChangesetStatus on demand.

    # There will only be one ChangesetStatus referencing each comment so the join will not explode.

    ${_('Settings')}

</%def>

<%block name="header_menu">

    ${self.menu('repositories')}

</%block>

<%def name="main()">

${self.repo_context_bar('options')}

<div class="panel panel-primary">

    <div class="panel-body settings">

        <ul class="nav nav-pills nav-stacked">

          <li class="${'active' if c.active=='settings' else ''}">

              <a href="${h.url('edit_repo', repo_name=c.repo_name)}">${_('Settings')}</a>

          </li>

          <li class="${'active' if c.active=='permissions' else ''}">

              <a href="${h.url('edit_repo_perms', repo_name=c.repo_name)}">${_('Permissions')}</a>

          </li>

          <li class="${'active' if c.active=='advanced' else ''}">

              <a href="${h.url('edit_repo_advanced', repo_name=c.repo_name)}">${_('Advanced')}</a>

          </li>

          <li class="${'active' if c.active=='fields' else ''}">

              <a href="${h.url('edit_repo_fields', repo_name=c.repo_name)}">${_('Extra Fields')}</a>

          </li>

          <li class="${'active' if c.active=='remote' else ''}">

              <a href="${h.url('edit_repo_remote', repo_name=c.repo_name)}">${_('Remote')}</a>

          </li>

          <li class="${'active' if c.active=='statistics' else ''}">

              <a href="${h.url('edit_repo_statistics', repo_name=c.repo_name)}">${_('Statistics')}</a>

          </li>

        </ul>

        <div>

            <%include file="/admin/repos/repo_edit_${c.active}.html"/>

        </div>

    </div>

</div>

</%def>

                                  clone_uri=custom_remote_path)

        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_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_create_existing_user(self):

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

                                  username=base.TEST_USER_ADMIN_LOGIN,

                                  email='test@example.com',

                                  password='trololo')

        response = api_call(self, params)

        expected = "user `%s` already exist" % base.TEST_USER_ADMIN_LOGIN

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

    def test_api_create_user_with_existing_email(self):

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

                                  username=base.TEST_USER_ADMIN_LOGIN + 'new',

                                  email=base.TEST_USER_REGULAR_EMAIL,

                                  password='trololo')

        response = api_call(self, params)

        expected = "email `%s` already exist" % base.TEST_USER_REGULAR_EMAIL

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

    def test_api_create_user(self):

        username = 'test_new_api_user'

        email = username + "@example.com"

The tests need Git > 1.8.1.

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 json

import os

import re

import tempfile

import time

import urllib.request

from subprocess import PIPE, Popen

from tempfile import _RandomNameSequence

import pytest

from kallithea import CONFIG

from kallithea.lib.utils2 import ascii_bytes, safe_str

from kallithea.model.meta import Session

from kallithea.model.ssh_key import SshKeyModel

from kallithea.model.user import UserModel

from kallithea.tests import base

from kallithea.tests.fixture import Fixture

DEBUG = True

HOST = '127.0.0.1:4999'  # test host

fixture = Fixture()

# Parameterize different kinds of VCS testing - both the kind of VCS and the

# access method (HTTP/SSH)

# Mixin for using HTTP and SSH URLs

class HttpVcsTest(object):

    @staticmethod

    def repo_url_param(webserver, repo_name, **kwargs):

        return webserver.repo_url(repo_name, **kwargs)

class SshVcsTest(object):

    public_keys = {

            else:

                assert vt.repo_type == 'hg'

                assert "abort: HTTP Error 404: Not Found" in stderr

        else:

            assert issubclass(vt, SshVcsTest)

            if vt.repo_type == 'git':

                assert "abort: Access to './%s' denied" % vt.repo_name in stderr

            else:

                assert "abort: Access to './%s' denied" % vt.repo_name in stdout

        stdout, stderr = Command(dest_dir).execute(vt.repo_type, 'pull', clone_url.replace('/' + vt.repo_name, '/%s/' % vt.repo_name), ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'Already up to date.' in stdout

        else:

            assert vt.repo_type == 'hg'

            assert "no changes found" in stdout

        assert "denied" not in stderr

        assert "denied" not in stdout

        assert "404" not in stdout

    @parametrize_vcs_test

    def test_push_invalidates_cache(self, webserver, testfork, vt):

        pre_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, testfork[vt.repo_type])

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir, files_no=1, clone_url=clone_url)

        if vt.repo_type == 'git':

            _check_proper_git_push(stdout, stderr)

        post_cached_tip = [repo.get_api_data()['last_changeset']['short_id'] for repo in Repository.query().filter(Repository.repo_name == testfork[vt.repo_type])]

        assert pre_cached_tip != post_cached_tip

    @parametrize_vcs_test_http

    def test_push_wrong_credentials(self, webserver, vt):

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)

        clone_url = webserver.repo_url(vt.repo_name, username='bad', password='name')

        stdout, stderr = _add_files_and_push(webserver, vt, dest_dir,

                                             clone_url=clone_url, ignoreReturnCode=True)

        if vt.repo_type == 'git':

            assert 'fatal: Authentication failed' in stderr

        elif vt.repo_type == 'hg':

            assert 'abort: authorization failed' in stderr

    @parametrize_vcs_test

    def test_push_with_readonly_credentials(self, webserver, vt):

        UserLog.query().delete()

        Session().commit()

        dest_dir = _get_tmp_dir()

        clone_url = vt.repo_url_param(webserver, vt.repo_name, username=base.TEST_USER_REGULAR_LOGIN, password=base.TEST_USER_REGULAR_PASS)

        stdout, stderr = Command(base.TESTS_TMP_PATH).execute(vt.repo_type, 'clone', clone_url, dest_dir)