c.active = 'mapping'

        if request.POST:

            rm_obsolete = request.POST.get('destroy', False)

            install_git_hooks = request.POST.get('hooks', False)

            overwrite_git_hooks = request.POST.get('hooks_overwrite', False);

            invalidate_cache = request.POST.get('invalidate', False)

            log.debug('rescanning repo location with destroy obsolete=%s, '

                      'install git hooks=%s and '

                      'overwrite git hooks=%s' % (rm_obsolete, install_git_hooks, overwrite_git_hooks))

            if invalidate_cache:

                log.debug('invalidating all repositories cache')

                    ScmModel().mark_for_invalidation(repo.repo_name)

            filesystem_repos = ScmModel().repo_scan()

            added, removed = repo2db_mapper(filesystem_repos, rm_obsolete,

                                            install_git_hooks=install_git_hooks,

                                            user=c.authuser.username,

                                            overwrite_git_hooks=overwrite_git_hooks)

            h.flash(h.literal(_('Repositories successfully rescanned. Added: %s. Removed: %s.') %

                (', '.join(h.link_to(safe_unicode(repo_name), h.url('summary_home', repo_name=repo_name))

                 for repo_name in added) or '-',

                 ', '.join(h.escape(safe_unicode(repo_name)) for repo_name in removed) or '-')),

                category='success')

            if not isinstance(userid, Optional) and userid != self.authuser.user_id:

                raise JSONRPCError(

                    'userid is not the same as your user'

                )

        ret = []

        if isinstance(userid, Optional):

            user = None

        else:

            user = get_user_or_error(userid)

        # show all locks

            userid, time_ = r.locked

            if time_:

                _api_data = r.get_api_data()

                # if we use userfilter just show the locks for this user

                if user is not None:

                    if safe_int(userid) == user.user_id:

                        ret.append(_api_data)

                else:

                    ret.append(_api_data)

        return ret

        read access to user group.

        OUTPUT::

            id : <id_given_in_input>

            result : [<user_group_obj>,...]

            error : null

        """

        result = []

        _perms = ('usergroup.read', 'usergroup.write', 'usergroup.admin',)

                                        perm_set=_perms):

            result.append(user_group.get_api_data())

        return result

    @HasPermissionAnyDecorator('hg.admin', 'hg.usergroup.create.true')

    def create_user_group(self, group_name, description=Optional(''),

                          owner=Optional(OAttr('apiuser')), active=Optional(True)):

        """

        Creates new user group. This command can be executed only using api_key

        belonging to user with admin rights or an user who has create user group

        permission

                        "enable_downloads":  "<bool>",

                        "enable_locking":    "<bool>",

                        "enable_statistics": "<bool>",

                      },

                      …

                    ]

            error:  null

        """

        result = []

        if not HasPermissionAny('hg.admin')():

            repos = RepoModel().get_all_user_repos(user=self.authuser.user_id)

        else:

        for repo in repos:

            result.append(repo.get_api_data())

        return result

    # permission check inside

    def get_repo_nodes(self, repoid, revision, root_path,

                       ret_type=Optional('all')):

        """

        returns a list of nodes and it's children in a flat list for a given path

        at given revision. It's possible to specify ret_type to show only `files` or

        `dirs`.  This command can be executed only using api_key belonging to

        data = repo_group.get_api_data()

        data["members"] = members

        return data

    @HasPermissionAnyDecorator('hg.admin')

    def get_repo_groups(self):

        """

        Returns all repository groups

        """

        result = []

            result.append(repo_group.get_api_data())

        return result

    @HasPermissionAnyDecorator('hg.admin')

    def create_repo_group(self, group_name, description=Optional(''),

                          owner=Optional(OAttr('apiuser')),

                          parent=Optional(None),

                          copy_permissions=Optional(False)):

        """

        Creates a repository group. This command can be executed only using

        api_key belonging to user with admin rights.

    usage = "CONFIG_FILE"

    group_name = "Kallithea"

    takes_config_file = -1

    parser = BasePasterCommand.standard_parser(verbose=True)

    summary = "Updates repositories caches for last changeset"

    def command(self):

        #get SqlAlchemy session

        self._init_session()

        if self.options.repo_update_list is None:

        else:

            repo_names = [safe_unicode(n.strip())

                          for n in self.options.repo_update_list.split(',')]

            repo_list = list(Repository.query()

                .filter(Repository.repo_name.in_(repo_names)))

        RepoModel.update_repoinfo(repositories=repo_list)

        Session().commit()

        if self.options.invalidate_cache:

            for r in repo_list:

                r.set_invalidate()

            print 'Updated repo info and invalidated cache for %s repositories' % (len(repo_list))

    def get_or_404(cls, id_):

        try:

            id_ = int(id_)

        except (TypeError, ValueError):

            raise HTTPNotFound

        res = cls.query().get(id_)

        if res is None:

            raise HTTPNotFound

        return res

    @classmethod

    def delete(cls, id_):

        obj = cls.query().get(id_)

        Session().delete(obj)

    def __repr__(self):

        if hasattr(self, '__unicode__'):

            # python repr needs to return str

            try:

                return safe_str(self.__unicode__())

            except UnicodeDecodeError:

                pass

        return '<DB:%s>' % (self.__class__.__name__)

        from pylons.i18n.translation import _

        _tmpl_lookup = kallithea.CONFIG['pylons.app_globals'].mako_lookup

        template = _tmpl_lookup.get_template('data_table/_dt_elements.html')

        tmpl = template.get_def(tmpl)

        kwargs.update(dict(_=_, h=h, c=c))

        return tmpl.render(*args, **kwargs)

    @classmethod

    def update_repoinfo(cls, repositories=None):

        if repositories is None:

        for repo in repositories:

            repo.update_changeset_cache()

    def get_repos_as_dict(self, repos_list=None, admin=False, perm_check=True,

                          super_user_actions=False, short_name=False):

        _render = self._render_datatable

        from pylons import tmpl_context as c

        def quick_menu(repo_name):

            return _render('quick_menu', repo_name)

        def repo_lnk(name, rtype, rstate, private, fork_of):

                                  repoid='no-such-repo')

        response = api_call(self, params)

        ret = 'repository `%s` does not exist' % 'no-such-repo'

        expected = ret

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

    def test_api_get_repos(self):

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

        response = api_call(self, params)

        result = []

            result.append(repo.get_api_data())

        ret = jsonify(result)

        expected = ret

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

    def test_api_get_repos_non_admin(self):

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

        response = api_call(self, params)

        result = []

        for repo in RepoModel().get_all_user_repos(self.TEST_USER_LOGIN):

        }

        form_data.update(kwargs)

        gist = GistModel().create(

            description=form_data['description'],owner=form_data['owner'],

            gist_mapping=form_data['gist_mapping'], gist_type=form_data['gist_type'],

            lifetime=form_data['lifetime']

        )

        Session().commit()

        return gist

    def destroy_gists(self, gistid=None):

            if gistid:

                if gistid == g.gist_access_id:

                    GistModel().delete(g)

            else:

                GistModel().delete(g)

        Session().commit()

    def load_resource(self, resource_name, strip=True):

        with open(os.path.join(FIXTURES, resource_name), 'rb') as f:

            source = f.read()

            if strip:

                source = source.strip()

    }

    user = User.get_by_username(owner)

    gist = GistModel().create(description, owner=user,

                       gist_mapping=gist_mapping, gist_type=gist_type,

                       lifetime=lifetime)

    Session().commit()

    return gist

class TestGistsController(TestController):

    def teardown_method(self, method):

            GistModel().delete(g)

        Session().commit()

    def test_index(self):

        self.log_user()

        response = self.app.get(url('gists'))

        # Test response...

        response.mustcontain('There are no gists yet')

        g1 = _create_gist('gist1').gist_access_id

        g2 = _create_gist('gist2', lifetime=1400).gist_access_id

        g3 = _create_gist('gist3', description=u'gist3-desc').gist_access_id

fixture = Fixture()

class TestHomeController(TestController):

    def test_index(self):

        self.log_user()

        response = self.app.get(url(controller='home', action='index'))

        #if global permission is set

        response.mustcontain('Add Repository')

        # html in javascript variable:

        response.mustcontain(r'href=\"/%s\"' % HG_REPO)

        response.mustcontain(r'<span class="repotag">git')

        response.mustcontain(r'<i class=\"icon-globe\"')

        response.mustcontain("""fixes issue with having custom format for git-log""")

        response.mustcontain("""/%s/changeset/5f2c6ee195929b0be80749243c18121c9864a3b3""" % GIT_REPO)

        response.mustcontain("""disable security checks on hg clone for travis""")

        response.mustcontain("""/%s/changeset/96507bd11ecc815ebc6270fdf6db110928c09c1e""" % HG_REPO)

    def test_repo_summary_with_anonymous_access_disabled(self):

from kallithea.model.user_group import UserGroupModel

from kallithea.model.meta import Session

fixture = Fixture()

class TestUserGroups(TestController):

    def teardown_method(self, method):

        # delete all groups

            fixture.destroy_user_group(gr)

        Session().commit()

    @parametrize('pre_existing,regular_should_be,external_should_be,groups,expected', [

        ([], [], [], [], []),

        ([], [u'regular'], [], [], [u'regular']),  # no changes of regular

        ([u'some_other'], [], [], [u'some_other'], []),   # not added to regular group

        ([], [u'regular'], [u'container'], [u'container'], [u'regular', u'container']),

        ([], [u'regular'], [], [u'container', u'container2'], [u'regular', u'container', u'container2']),

        ([], [u'regular'], [u'other'], [], [u'regular']),  # remove not used

        ([u'some_other'], [u'regular'], [u'other', u'container'], [u'container', u'container2'], [u'regular', u'container', u'container2']),

    ])

    def test_enforce_groups(self, pre_existing, regular_should_be,

                            external_should_be, groups, expected):

        # delete all groups

            fixture.destroy_user_group(gr)

        Session().commit()

        user = User.get_by_username(TEST_USER_REGULAR_LOGIN)

        for gr in pre_existing:

            gr = fixture.create_user_group(gr)

        Session().commit()

        # make sure use is just in those groups

        for gr in regular_should_be:

            gr = fixture.create_user_group(gr)

            Session().commit()

        assert 'Last revision is now' in stdout

    def test_push_new_file_git(self):

        DEST = _get_tmp_dir()

        clone_url = _construct_url(GIT_REPO, dest=DEST)

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

        # 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)

        _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)

        key.cache_active = True

        Session().add(key)

        Session().commit()

        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)

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

        finally:

            #release IP restrictions

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to invalidate this cache explicitly

        invalidate_all_caches()

        clone_url = _construct_url(HG_REPO)

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

        assert 'requesting all changes' in stdout

        assert 'adding changesets' in stdout

        assert 'adding manifests' in stdout

    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)

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

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

        finally:

            #release IP restrictions

                UserIpMap.delete(ip.ip_id)

            Session().commit()

        # IP permissions are cached, need to invalidate this cache explicitly

        invalidate_all_caches()

        clone_url = _construct_url(GIT_REPO)

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

        assert 'Cloning into' in stdout + stderr

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