tests: Fix a helper method that checks if a repository is available on filesystem.

- Moved it into utils to remove duplication
- Refactored it to not hide all kinds of exceptions.

rhodecode/tests/functional/test_admin_repos.py

@@ -1,1260 +1,1250 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
-import os
 import urllib
 
 import mock
 import pytest
 
 from rhodecode.lib import auth
-from rhodecode.lib import vcs
 from rhodecode.lib.utils2 import safe_str, str2bool
 from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
 from rhodecode.model.db import Repository, RepoGroup, UserRepoToPerm, User,\
     Permission
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.repo_group import RepoGroupModel
 from rhodecode.model.settings import SettingsModel, VcsSettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.tests import (
     login_user_session, url, assert_session_flash, TEST_USER_ADMIN_LOGIN,
     TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS, HG_REPO, GIT_REPO,
-    TESTS_TMP_PATH, logout_user_session)
+    logout_user_session)
 from rhodecode.tests.fixture import Fixture, error_function
-from rhodecode.tests.utils import AssertResponse
+from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
 
 fixture = Fixture()
 
 
 @pytest.mark.usefixtures("app")
 class TestAdminRepos:
 
     def test_index(self):
         self.app.get(url('repos'))
 
     def test_create_page_restricted(self, autologin_user, backend):
         with mock.patch('rhodecode.BACKENDS', {'git': 'git'}):
             response = self.app.get(url('new_repo'), status=200)
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#repo_type')
         assert element.text_content() == '\ngit\n'
 
     def test_create_page_non_restricted(self, autologin_user, backend):
         response = self.app.get(url('new_repo'), status=200)
         assert_response = AssertResponse(response)
         assert_response.element_contains('#repo_type', 'git')
         assert_response.element_contains('#repo_type', 'svn')
         assert_response.element_contains('#repo_type', 'hg')
 
     @pytest.mark.parametrize("suffix", [u'', u''], ids=['', 'non-ascii'])
     def test_create(self, autologin_user, backend, suffix, csrf_token):
         repo_name_unicode = backend.new_repo_name(suffix=suffix)
         repo_name = repo_name_unicode.encode('utf8')
         description_unicode = u'description for newly created repo' + suffix
         description = description_unicode.encode('utf8')
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token),
             status=302
             )
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     def test_create_numeric(self, autologin_user, backend, csrf_token):
         numeric_repo = '1234'
         repo_name = numeric_repo
         description = 'description for newly created repo' + numeric_repo
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         self.assert_repository_is_created_correctly(
             repo_name, description, backend)
 
     @pytest.mark.parametrize("suffix", [u'', u'ąćę'], ids=['', 'non-ascii'])
     def test_create_in_group(
             self, autologin_user, backend, suffix, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = u'ingroup' + suffix
         repo_name_full = RepoGroup.url_sep().join(
             [group_name, repo_name])
         description = u'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=safe_str(repo_name),
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_numeric(
             self, autologin_user, backend, csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         repo_name = '12345'
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup work to fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query()\
                 .filter(UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             Session().commit()
 
     def test_create_in_group_without_needed_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         # create GROUP
         group_name = 'reg_sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         Session().commit()
 
         group_name_allowed = 'reg_sometest_allowed_%s' % backend.alias
         gr_allowed = RepoGroupModel().create(
             group_name=group_name_allowed,
             group_description='test',
             owner=TEST_USER_REGULAR_LOGIN)
         Session().commit()
 
         repo_name = 'ingroup'
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 csrf_token=csrf_token))
 
         response.mustcontain('Invalid value')
 
         # user is allowed to create in this group
         repo_name = 'ingroup'
         repo_name_full = RepoGroup.url_sep().join(
             [group_name_allowed, repo_name])
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr_allowed.group_id,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup in pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
 
             new_repo = RepoModel().get_by_repo_name(repo_name_full)
             inherited_perms = UserRepoToPerm.query().filter(
                 UserRepoToPerm.repository_id == new_repo.repo_id).all()
             assert len(inherited_perms) == 1
 
             assert repo_on_filesystem(repo_name_full)
         finally:
             RepoModel().delete(repo_name_full)
             RepoGroupModel().delete(group_name)
             RepoGroupModel().delete(group_name_allowed)
             Session().commit()
 
     def test_create_in_group_inherit_permissions(self, autologin_user, backend,
                                                  csrf_token):
         # create GROUP
         group_name = 'sometest_%s' % backend.alias
         gr = RepoGroupModel().create(group_name=group_name,
                                      group_description='test',
                                      owner=TEST_USER_ADMIN_LOGIN)
         perm = Permission.get_by_key('repository.write')
         RepoGroupModel().grant_user_permission(
             gr, TEST_USER_REGULAR_LOGIN, perm)
 
         # add repo permissions
         Session().commit()
 
         repo_name = 'ingroup_inherited_%s' % backend.alias
         repo_name_full = RepoGroup.url_sep().join([group_name, repo_name])
         description = 'description for newly created repo'
         self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 repo_group=gr.group_id,
                 repo_copy_permissions=True,
                 csrf_token=csrf_token))
 
         # TODO: johbo: Cleanup to pytest fixture
         try:
             self.assert_repository_is_created_correctly(
                 repo_name_full, description, backend)
         except Exception:
             RepoGroupModel().delete(group_name)
             Session().commit()
             raise
 
         # check if inherited permissions are applied
         new_repo = RepoModel().get_by_repo_name(repo_name_full)
         inherited_perms = UserRepoToPerm.query().filter(
             UserRepoToPerm.repository_id == new_repo.repo_id).all()
         assert len(inherited_perms) == 2
 
         assert TEST_USER_REGULAR_LOGIN in [
             x.user.username for x in inherited_perms]
         assert 'repository.write' in [
             x.permission.permission_name for x in inherited_perms]
 
         RepoModel().delete(repo_name_full)
         RepoGroupModel().delete(group_name)
         Session().commit()
 
     @pytest.mark.xfail_backends(
         "git", "hg", reason="Missing reposerver support")
     def test_create_with_clone_uri(self, autologin_user, backend, reposerver,
                                    csrf_token):
         source_repo = backend.create_repo(number_of_commits=2)
         source_repo_name = source_repo.repo_name
         reposerver.serve(source_repo.scm_instance())
 
         repo_name = backend.new_repo_name()
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description='',
                 clone_uri=reposerver.url,
                 csrf_token=csrf_token),
             status=302)
 
         # Should be redirected to the creating page
         response.mustcontain('repo_creating')
 
         # Expecting that both repositories have same history
         source_repo = RepoModel().get_by_repo_name(source_repo_name)
         source_vcs = source_repo.scm_instance()
         repo = RepoModel().get_by_repo_name(repo_name)
         repo_vcs = repo.scm_instance()
         assert source_vcs[0].message == repo_vcs[0].message
         assert source_vcs.count() == repo_vcs.count()
         assert source_vcs.commit_ids == repo_vcs.commit_ids
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri(self, autologin_user, backend,
                                                 csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='http://repo.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     @pytest.mark.xfail_backends("svn", reason="Depends on import support")
     def test_create_remote_repo_wrong_clone_uri_hg_svn(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 clone_uri='svn+http://svn.invalid/repo',
                 csrf_token=csrf_token))
         response.mustcontain('invalid clone url')
 
     @pytest.mark.parametrize("suffix", [u'', u'ąęł'], ids=['', 'non-ascii'])
     def test_delete(self, autologin_user, backend, suffix, csrf_token):
         repo = backend.create_repo(name_suffix=suffix)
         repo_name = repo.repo_name
 
         response = self.app.post(url('repo', repo_name=repo_name),
                                  params={'_method': 'delete',
                                          'csrf_token': csrf_token})
         assert_session_flash(response, 'Deleted repository %s' % (repo_name))
         response.follow()
 
         # check if repo was deleted from db
         assert RepoModel().get_by_repo_name(repo_name) is None
         assert not repo_on_filesystem(repo_name)
 
     def test_show(self, autologin_user, backend):
         self.app.get(url('repo', repo_name=backend.repo_name))
 
     def test_edit(self, backend, autologin_user):
         self.app.get(url('edit_repo', repo_name=backend.repo_name))
 
     def test_edit_accessible_when_missing_requirements(
             self, backend_hg, autologin_user):
         scm_patcher = mock.patch.object(
             Repository, 'scm_instance', side_effect=RepositoryRequirementError)
         with scm_patcher:
             self.app.get(url('edit_repo', repo_name=backend_hg.repo_name))
 
     def test_set_private_flag_sets_default_to_none(
             self, autologin_user, backend, csrf_token):
         # initially repository perm should be read
         perm = _get_permission_for_user(user='default', repo=backend.repo_name)
         assert len(perm) == 1
         assert perm[0].permission.permission_name == 'repository.read'
         assert not backend.repo.private
 
         response = self.app.post(
             url('repo', repo_name=backend.repo_name),
             fixture._get_repo_create_params(
                 repo_private=1,
                 repo_name=backend.repo_name,
                 repo_type=backend.alias,
                 user=TEST_USER_ADMIN_LOGIN,
                 _method='put',
                 csrf_token=csrf_token))
         assert_session_flash(
             response,
             msg='Repository %s updated successfully' % (backend.repo_name))
         assert backend.repo.private
 
         # now the repo default permission should be None
         perm = _get_permission_for_user(user='default', repo=backend.repo_name)
         assert len(perm) == 1
         assert perm[0].permission.permission_name == 'repository.none'
 
         response = self.app.post(
             url('repo', repo_name=backend.repo_name),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=backend.repo_name,
                 repo_type=backend.alias,
                 user=TEST_USER_ADMIN_LOGIN,
                 _method='put',
                 csrf_token=csrf_token))
         assert_session_flash(
             response,
             msg='Repository %s updated successfully' % (backend.repo_name))
         assert not backend.repo.private
 
         # we turn off private now the repo default permission should stay None
         perm = _get_permission_for_user(user='default', repo=backend.repo_name)
         assert len(perm) == 1
         assert perm[0].permission.permission_name == 'repository.none'
 
         # update this permission back
         perm[0].permission = Permission.get_by_key('repository.read')
         Session().add(perm[0])
         Session().commit()
 
     def test_default_user_cannot_access_private_repo_in_a_group(
             self, autologin_user, user_util, backend, csrf_token):
 
         group = user_util.create_repo_group()
 
         repo = backend.create_repo(
             repo_private=True, repo_group=group, repo_copy_permissions=True)
 
         permissions = _get_permission_for_user(
             user='default', repo=repo.repo_name)
         assert len(permissions) == 1
         assert permissions[0].permission.permission_name == 'repository.none'
         assert permissions[0].repository.private is True
 
     def test_set_repo_fork_has_no_self_id(self, autologin_user, backend):
         repo = backend.repo
         response = self.app.get(
             url('edit_repo_advanced', repo_name=backend.repo_name))
         opt = """<option value="%s">vcs_test_git</option>""" % repo.repo_id
         response.mustcontain(no=[opt])
 
     def test_set_fork_of_target_repo(
             self, autologin_user, backend, csrf_token):
         target_repo = 'target_%s' % backend.alias
         fixture.create_repo(target_repo, repo_type=backend.alias)
         repo2 = Repository.get_by_repo_name(target_repo)
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo2.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         repo = Repository.get_by_repo_name(backend.repo_name)
         repo2 = Repository.get_by_repo_name(target_repo)
         assert_session_flash(
             response,
             'Marked repo %s as fork of %s' % (repo.repo_name, repo2.repo_name))
 
         assert repo.fork == repo2
         response = response.follow()
         # check if given repo is selected
 
         opt = 'This repository is a fork of <a href="%s">%s</a>' % (
             url('summary_home', repo_name=repo2.repo_name), repo2.repo_name)
 
         response.mustcontain(opt)
 
         fixture.destroy_repo(target_repo, forks='detach')
 
     @pytest.mark.backends("hg", "git")
     def test_set_fork_of_other_type_repo(self, autologin_user, backend,
                                          csrf_token):
         TARGET_REPO_MAP = {
             'git': {
                 'type': 'hg',
                 'repo_name': HG_REPO},
             'hg': {
                 'type': 'git',
                 'repo_name': GIT_REPO},
         }
         target_repo = TARGET_REPO_MAP[backend.alias]
 
         repo2 = Repository.get_by_repo_name(target_repo['repo_name'])
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo2.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response,
             'Cannot set repository as fork of repository with other type')
 
     def test_set_fork_of_none(self, autologin_user, backend, csrf_token):
         # mark it as None
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': None, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response,
             'Marked repo %s as fork of %s'
             % (backend.repo_name, "Nothing"))
         assert backend.repo.fork is None
 
     def test_set_fork_of_same_repo(self, autologin_user, backend, csrf_token):
         repo = Repository.get_by_repo_name(backend.repo_name)
         response = self.app.post(
             url('edit_repo_advanced_fork', repo_name=backend.repo_name),
             params={'id_fork_of': repo.repo_id, '_method': 'put',
                     'csrf_token': csrf_token})
         assert_session_flash(
             response, 'An error occurred during this operation')
 
     def test_create_on_top_level_without_permissions(self, backend):
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
 
         # revoke
         user_model = UserModel()
         # disable fork and create on default user
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.create.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.create.none')
         user_model.revoke_perm(User.DEFAULT_USER, 'hg.fork.repository')
         user_model.grant_perm(User.DEFAULT_USER, 'hg.fork.none')
 
         # disable on regular user
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.create.none')
         user_model.revoke_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.repository')
         user_model.grant_perm(TEST_USER_REGULAR_LOGIN, 'hg.fork.none')
         Session().commit()
 
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         response.mustcontain(
             u"You do not have the permission to store repositories in "
             u"the root location.")
 
     @mock.patch.object(RepoModel, '_create_filesystem_repo', error_function)
     def test_create_repo_when_filesystem_op_fails(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.new_repo_name()
         description = 'description for newly created repo'
 
         response = self.app.post(
             url('repos'),
             fixture._get_repo_create_params(
                 repo_private=False,
                 repo_name=repo_name,
                 repo_type=backend.alias,
                 repo_description=description,
                 csrf_token=csrf_token))
 
         assert_session_flash(
             response, 'Error creating repository %s' % repo_name)
         # repo must not be in db
         assert backend.repo is None
         # repo must not be in filesystem !
         assert not repo_on_filesystem(repo_name)
 
     def assert_repository_is_created_correctly(
             self, repo_name, description, backend):
         repo_name_utf8 = repo_name.encode('utf-8')
 
         # run the check page that triggers the flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert response.json == {u'result': True}
         assert_session_flash(
             response,
             u'Created repository <a href="/%s">%s</a>'
             % (urllib.quote(repo_name_utf8), repo_name))
 
         # test if the repo was created in the database
         new_repo = RepoModel().get_by_repo_name(repo_name)
 
         assert new_repo.repo_name == repo_name
         assert new_repo.description == description
 
         # test if the repository is visible in the list ?
         response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(repo_name)
         response.mustcontain(backend.alias)
 
         assert repo_on_filesystem(repo_name)
 
 
 @pytest.mark.usefixtures("app")
 class TestVcsSettings(object):
     FORM_DATA = {
         'inherit_global_settings': False,
         'hooks_changegroup_repo_size': False,
         'hooks_changegroup_push_logger': False,
         'hooks_outgoing_pull_logger': False,
         'extensions_largefiles': False,
         'phases_publish': 'false',
         'rhodecode_pr_merge_enabled': False,
         'rhodecode_use_outdated_comments': False,
         'new_svn_branch': '',
         'new_svn_tag': ''
     }
 
     @pytest.mark.skip_backends('svn')
     def test_global_settings_initial_values(self, autologin_user, backend):
         repo_name = backend.repo_name
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         expected_settings = (
             'rhodecode_use_outdated_comments', 'rhodecode_pr_merge_enabled',
             'hooks_changegroup_repo_size', 'hooks_changegroup_push_logger',
             'hooks_outgoing_pull_logger'
         )
         for setting in expected_settings:
             self.assert_repo_value_equals_global_value(response, setting)
 
     def test_show_settings_requires_repo_admin_permission(
             self, backend, user_util, settings_util):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         self.app.get(url('repo_vcs_settings', repo_name=repo_name), status=200)
 
     def test_inherit_global_settings_flag_is_true_by_default(
             self, autologin_user, backend):
         repo_name = backend.repo_name
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#inherit_global_settings')
         assert element.checked
 
     @pytest.mark.parametrize('checked_value', [True, False])
     def test_inherit_global_settings_value(
             self, autologin_user, backend, checked_value, settings_util):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         settings_util.create_repo_rhodecode_setting(
             repo, 'inherit_vcs_settings', checked_value, 'bool')
         response = self.app.get(url('repo_vcs_settings', repo_name=repo_name))
 
         assert_response = AssertResponse(response)
         element = assert_response.get_element('#inherit_global_settings')
         assert element.checked == checked_value
 
     @pytest.mark.skip_backends('svn')
     def test_hooks_settings_are_created(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hooks_settings_are_not_created_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui is None
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_hooks_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings.create_ui_section_value(section, '', key=key, active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hooks_settings_are_not_updated_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         settings = SettingsModel(repo=repo_name)
         for section, key in VcsSettingsModel.HOOKS_SETTINGS:
             settings.create_ui_section_value(section, '', key=key, active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for section, key in VcsSettingsModel.HOOKS_SETTINGS:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 assert ui.ui_active is True
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_pr_settings_are_created(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_are_not_created_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting is None
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_creation_requires_repo_admin_permission(
             self, backend, user_util, settings_util, csrf_token):
         repo = backend.create_repo()
         repo_name = repo.repo_name
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
 
         logout_user_session(self.app, csrf_token)
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         new_csrf_token = auth.get_csrf_token(session)
 
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         data = self.FORM_DATA.copy()
         data['csrf_token'] = new_csrf_token
         settings = SettingsModel(repo=repo_name)
 
         try:
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 status=302)
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_pr_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             settings.create_or_update_setting(name, True, 'bool')
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_pr_settings_are_not_updated_for_svn(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         settings = SettingsModel(repo=repo_name)
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             settings.create_or_update_setting(name, True, 'bool')
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 assert setting.app_settings_value is True
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_created(
             self, autologin_user, backend_svn, csrf_token, settings_util):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'svn-tag'
         data['new_svn_branch'] = 'svn-branch'
         data['csrf_token'] = csrf_token
 
         # Create few global settings to make sure that uniqueness validators
         # are not triggered
         settings_util.create_rhodecode_ui(
             VcsSettingsModel.SVN_BRANCH_SECTION, 'svn-branch')
         settings_util.create_rhodecode_ui(
             VcsSettingsModel.SVN_TAG_SECTION, 'svn-tag')
 
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_branch_names = [b.ui_value for b in svn_branches]
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             svn_tag_names = [b.ui_value for b in svn_tags]
             assert 'svn-branch' in svn_branch_names
             assert 'svn-tag' in svn_tag_names
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_unique(
             self, autologin_user, backend_svn, csrf_token, settings_util):
         repo = backend_svn.repo
         repo_name = repo.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'test_tag'
         data['new_svn_branch'] = 'test_branch'
         data['csrf_token'] = csrf_token
         settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch')
         settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag')
 
         response = self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=200)
         response.mustcontain('Pattern already exists')
 
     def test_svn_settings_with_empty_values_are_not_created(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             assert len(svn_branches) == 0
             assert len(svn_tags) == 0
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_svn_settings_are_shown_for_svn_repository(
             self, autologin_user, backend_svn, csrf_token):
         repo_name = backend_svn.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain('Subversion Settings')
 
     @pytest.mark.skip_backends('svn')
     def test_svn_settings_are_not_created_for_not_svn_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             svn_branches = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION)
             svn_tags = settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION)
             assert len(svn_branches) == 0
             assert len(svn_tags) == 0
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('svn')
     def test_svn_settings_are_shown_only_for_svn_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='Subversion Settings')
 
     def test_hg_settings_are_created(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         data = self.FORM_DATA.copy()
         data['new_svn_tag'] = 'svn-tag'
         data['new_svn_branch'] = 'svn-branch'
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is False
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert str2bool(phases_ui.ui_value) is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hg_settings_are_updated(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         settings = SettingsModel(repo=repo_name)
         settings.create_ui_section_value(
             'extensions', '', key='largefiles', active=True)
         settings.create_ui_section_value(
             'phases', '1', key='publish', active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is False
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert str2bool(phases_ui.ui_value) is False
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_hg_settings_are_shown_for_hg_repository(
             self, autologin_user, backend_hg, csrf_token):
         repo_name = backend_hg.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain('Mercurial Settings')
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_created_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         settings = SettingsModel(repo=repo_name)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui is None
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert phases_ui is None
         finally:
             self._cleanup_repo_settings(settings)
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_shown_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='Mercurial Settings')
 
     @pytest.mark.skip_backends('hg')
     def test_hg_settings_are_updated_only_for_hg_repository(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         settings = SettingsModel(repo=repo_name)
         settings.create_ui_section_value(
             'extensions', '', key='largefiles', active=True)
         settings.create_ui_section_value(
             'phases', '1', key='publish', active=True)
 
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
         try:
             largefiles_ui = settings.get_ui_by_section_and_key(
                 'extensions', 'largefiles')
             assert largefiles_ui.ui_active is True
             phases_ui = settings.get_ui_by_section_and_key(
                 'phases', 'publish')
             assert phases_ui.ui_value == '1'
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_per_repo_svn_settings_are_displayed(
             self, autologin_user, backend_svn, settings_util):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         branches = [
             settings_util.create_repo_rhodecode_ui(
                 repo, VcsSettingsModel.SVN_BRANCH_SECTION,
                 'branch_{}'.format(i))
             for i in range(10)]
         tags = [
             settings_util.create_repo_rhodecode_ui(
                 repo, VcsSettingsModel.SVN_TAG_SECTION, 'tag_{}'.format(i))
             for i in range(10)]
 
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         assert_response = AssertResponse(response)
         for branch in branches:
             css_selector = '[name=branch_value_{}]'.format(branch.ui_id)
             element = assert_response.get_element(css_selector)
             assert element.value == branch.ui_value
         for tag in tags:
             css_selector = '[name=tag_ui_value_new_{}]'.format(tag.ui_id)
             element = assert_response.get_element(css_selector)
             assert element.value == tag.ui_value
 
     def test_per_repo_hg_and_pr_settings_are_not_displayed_for_svn(
             self, autologin_user, backend_svn, settings_util):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         response = self.app.get(
             url('repo_vcs_settings', repo_name=repo_name), status=200)
         response.mustcontain(no='<label>Hooks:</label>')
         response.mustcontain(no='<label>Pull Request Settings:</label>')
 
     def test_inherit_global_settings_value_is_saved(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
 
         settings = SettingsModel(repo=repo_name)
         vcs_settings = VcsSettingsModel(repo=repo_name)
         try:
             assert vcs_settings.inherit_global_settings is True
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_repo_cache_is_invalidated_when_settings_are_updated(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         settings = SettingsModel(repo=repo_name)
 
         invalidation_patcher = mock.patch(
             'rhodecode.controllers.admin.repos.ScmModel.mark_for_invalidation')
         with invalidation_patcher as invalidation_mock:
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 status=302)
         try:
             invalidation_mock.assert_called_once_with(repo_name, delete=True)
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_other_settings_not_saved_inherit_global_settings_is_true(
             self, autologin_user, backend, csrf_token):
         repo_name = backend.repo_name
         data = self.FORM_DATA.copy()
         data['csrf_token'] = csrf_token
         data['inherit_global_settings'] = True
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data, status=302)
 
         settings = SettingsModel(repo=repo_name)
         ui_settings = (
             VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
 
         vcs_settings = []
         try:
             for section, key in ui_settings:
                 ui = settings.get_ui_by_section_and_key(section, key)
                 if ui:
                     vcs_settings.append(ui)
             vcs_settings.extend(settings.get_ui_by_section(
                 VcsSettingsModel.SVN_BRANCH_SECTION))
             vcs_settings.extend(settings.get_ui_by_section(
                 VcsSettingsModel.SVN_TAG_SECTION))
             for name in VcsSettingsModel.GENERAL_SETTINGS:
                 setting = settings.get_setting_by_name(name)
                 if setting:
                     vcs_settings.append(setting)
             assert vcs_settings == []
         finally:
             self._cleanup_repo_settings(settings)
 
     def test_delete_svn_branch_and_tag_patterns(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         branch = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
             cleanup=False)
         tag = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_TAG_SECTION, 'test_tag', cleanup=False)
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token
         }
         for id_ in (branch.ui_id, tag.ui_id):
             data['delete_svn_pattern'] = id_,
             self.app.post(
                 url('repo_vcs_settings', repo_name=repo_name), data,
                 headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
         settings = VcsSettingsModel(repo=repo_name)
         assert settings.get_repo_svn_branch_patterns() == []
 
     def test_delete_svn_branch_requires_repo_admin_permission(
             self, backend_svn, user_util, settings_util, csrf_token):
         repo = backend_svn.create_repo()
         repo_name = repo.repo_name
         user = UserModel().get_by_username(TEST_USER_REGULAR_LOGIN)
         logout_user_session(self.app, csrf_token)
         session = login_user_session(
             self.app, TEST_USER_REGULAR_LOGIN, TEST_USER_REGULAR_PASS)
         csrf_token = auth.get_csrf_token(session)
         user_util.grant_user_permission_to_repo(repo, user, 'repository.admin')
         branch = settings_util.create_repo_rhodecode_ui(
             repo, VcsSettingsModel.SVN_BRANCH_SECTION, 'test_branch',
             cleanup=False)
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token,
             'delete_svn_pattern': branch.ui_id
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=200)
 
     def test_delete_svn_branch_raises_400_when_not_found(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo_name = backend_svn.repo_name
         data = {
             '_method': 'delete',
             'delete_svn_pattern': 123,
             'csrf_token': csrf_token
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
 
     def test_delete_svn_branch_raises_400_when_no_id_specified(
             self, autologin_user, backend_svn, settings_util, csrf_token):
         repo_name = backend_svn.repo_name
         data = {
             '_method': 'delete',
             'csrf_token': csrf_token
         }
         self.app.post(
             url('repo_vcs_settings', repo_name=repo_name), data,
             headers={'X-REQUESTED-WITH': 'XMLHttpRequest', }, status=400)
 
     def _cleanup_repo_settings(self, settings_model):
         cleanup = []
         ui_settings = (
             VcsSettingsModel.HOOKS_SETTINGS + VcsSettingsModel.HG_SETTINGS)
 
         for section, key in ui_settings:
             ui = settings_model.get_ui_by_section_and_key(section, key)
             if ui:
                 cleanup.append(ui)
 
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.INHERIT_SETTINGS))
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.SVN_BRANCH_SECTION))
         cleanup.extend(settings_model.get_ui_by_section(
             VcsSettingsModel.SVN_TAG_SECTION))
 
         for name in VcsSettingsModel.GENERAL_SETTINGS:
             setting = settings_model.get_setting_by_name(name)
             if setting:
                 cleanup.append(setting)
 
         for object_ in cleanup:
             Session().delete(object_)
         Session().commit()
 
     def assert_repo_value_equals_global_value(self, response, setting):
         assert_response = AssertResponse(response)
         global_css_selector = '[name={}_inherited]'.format(setting)
         repo_css_selector = '[name={}]'.format(setting)
         repo_element = assert_response.get_element(repo_css_selector)
         global_element = assert_response.get_element(global_css_selector)
         assert repo_element.value == global_element.value
 
 
-def repo_on_filesystem(repo_name):
-    try:
-        vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
-        return True
-    except Exception:
-        return False
-
-
 def _get_permission_for_user(user, repo):
     perm = UserRepoToPerm.query()\
         .filter(UserRepoToPerm.repository ==
                 Repository.get_by_repo_name(repo))\
         .filter(UserRepoToPerm.user == User.get_by_username(user))\
         .all()
     return perm

rhodecode/tests/functional/test_summary.py

@@ -1,525 +1,515 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import re
-import os
 
 import mock
 import pytest
 
 from rhodecode.controllers import summary
-from rhodecode.lib import vcs
 from rhodecode.lib import helpers as h
 from rhodecode.lib.compat import OrderedDict
 from rhodecode.lib.vcs.exceptions import RepositoryRequirementError
 from rhodecode.model.db import Repository
 from rhodecode.model.meta import Session
 from rhodecode.model.repo import RepoModel
 from rhodecode.model.scm import ScmModel
 from rhodecode.tests import (
-    TestController, url, HG_REPO, assert_session_flash, TESTS_TMP_PATH)
+    TestController, url, HG_REPO, assert_session_flash)
 from rhodecode.tests.fixture import Fixture
-from rhodecode.tests.utils import AssertResponse
+from rhodecode.tests.utils import AssertResponse, repo_on_filesystem
 
 
 fixture = Fixture()
 
 
 class TestSummaryController(TestController):
     def test_index(self, backend):
         self.log_user()
         repo_id = backend.repo.repo_id
         repo_name = backend.repo_name
         with mock.patch('rhodecode.lib.helpers.is_svn_without_proxy',
                         return_value=False):
             response = self.app.get(url('summary_home', repo_name=repo_name))
 
         # repo type
         response.mustcontain(
             '<i class="icon-%s">' % (backend.alias, )
         )
         # public/private
         response.mustcontain(
             """<i class="icon-unlock-alt">"""
         )
 
         # clone url...
         response.mustcontain(
             'id="clone_url" readonly="readonly"'
             ' value="http://[e-mail]:80/%s"' % (repo_name, ))
         response.mustcontain(
             'id="clone_url_id" readonly="readonly"'
             ' value="http://[e-mail]:80/_%s"' % (repo_id, ))
 
     def test_index_svn_without_proxy(self, backend_svn):
         self.log_user()
         repo_id = backend_svn.repo.repo_id
         repo_name = backend_svn.repo_name
         response = self.app.get(url('summary_home', repo_name=repo_name))
         # clone url...
         response.mustcontain(
             'id="clone_url" disabled'
             ' value="http://[e-mail]:80/%s"' % (repo_name, ))
         response.mustcontain(
             'id="clone_url_id" disabled'
             ' value="http://[e-mail]:80/_%s"' % (repo_id, ))
 
     def test_index_with_trailing_slash(self, autologin_user, backend):
         repo_id = backend.repo.repo_id
         repo_name = backend.repo_name
         with mock.patch('rhodecode.lib.helpers.is_svn_without_proxy',
                         return_value=False):
             response = self.app.get(
                 url('summary_home', repo_name=repo_name) + '/',
                 status=200)
 
         # clone url...
         response.mustcontain(
             'id="clone_url" readonly="readonly"'
             ' value="http://[e-mail]:80/%s"' % (repo_name, ))
         response.mustcontain(
             'id="clone_url_id" readonly="readonly"'
             ' value="http://[e-mail]:80/_%s"' % (repo_id, ))
 
     def test_index_by_id(self, backend):
         self.log_user()
         repo_id = backend.repo.repo_id
         response = self.app.get(url(
             'summary_home', repo_name='_%s' % (repo_id,)))
 
         # repo type
         response.mustcontain(
             '<i class="icon-%s">' % (backend.alias, )
         )
         # public/private
         response.mustcontain(
             """<i class="icon-unlock-alt">"""
         )
 
     def test_index_by_repo_having_id_path_in_name_hg(self):
         self.log_user()
         fixture.create_repo(name='repo_1')
         response = self.app.get(url('summary_home', repo_name='repo_1'))
 
         try:
             response.mustcontain("repo_1")
         finally:
             RepoModel().delete(Repository.get_by_repo_name('repo_1'))
             Session().commit()
 
     def test_index_with_anonymous_access_disabled(self):
         with fixture.anon_access(False):
             response = self.app.get(url('summary_home', repo_name=HG_REPO),
                                     status=302)
             assert 'login' in response.location
 
     def _enable_stats(self, repo):
         r = Repository.get_by_repo_name(repo)
         r.enable_statistics = True
         Session().add(r)
         Session().commit()
 
     expected_trending = {
         'hg': {
             "py": {"count": 68, "desc": ["Python"]},
             "rst": {"count": 16, "desc": ["Rst"]},
             "css": {"count": 2, "desc": ["Css"]},
             "sh": {"count": 2, "desc": ["Bash"]},
             "bat": {"count": 1, "desc": ["Batch"]},
             "cfg": {"count": 1, "desc": ["Ini"]},
             "html": {"count": 1, "desc": ["EvoqueHtml", "Html"]},
             "ini": {"count": 1, "desc": ["Ini"]},
             "js": {"count": 1, "desc": ["Javascript"]},
             "makefile": {"count": 1, "desc": ["Makefile", "Makefile"]}
         },
         'git': {
             "py": {"count": 68, "desc": ["Python"]},
             "rst": {"count": 16, "desc": ["Rst"]},
             "css": {"count": 2, "desc": ["Css"]},
             "sh": {"count": 2, "desc": ["Bash"]},
             "bat": {"count": 1, "desc": ["Batch"]},
             "cfg": {"count": 1, "desc": ["Ini"]},
             "html": {"count": 1, "desc": ["EvoqueHtml", "Html"]},
             "ini": {"count": 1, "desc": ["Ini"]},
             "js": {"count": 1, "desc": ["Javascript"]},
             "makefile": {"count": 1, "desc": ["Makefile", "Makefile"]}
         },
         'svn': {
             "py": {"count": 75, "desc": ["Python"]},
             "rst": {"count": 16, "desc": ["Rst"]},
             "html": {"count": 11, "desc": ["EvoqueHtml", "Html"]},
             "css": {"count": 2, "desc": ["Css"]},
             "bat": {"count": 1, "desc": ["Batch"]},
             "cfg": {"count": 1, "desc": ["Ini"]},
             "ini": {"count": 1, "desc": ["Ini"]},
             "js": {"count": 1, "desc": ["Javascript"]},
             "makefile": {"count": 1, "desc": ["Makefile", "Makefile"]},
             "sh": {"count": 1, "desc": ["Bash"]}
         },
     }
 
     def test_repo_stats(self, backend, xhr_header):
         self.log_user()
         response = self.app.get(
             url('repo_stats',
                 repo_name=backend.repo_name, commit_id='tip'),
             extra_environ=xhr_header,
             status=200)
         assert re.match(r'6[\d\.]+ KiB', response.json['size'])
 
     def test_repo_stats_code_stats_enabled(self, backend, xhr_header):
         self.log_user()
         repo_name = backend.repo_name
 
         # codes stats
         self._enable_stats(repo_name)
         ScmModel().mark_for_invalidation(repo_name)
 
         response = self.app.get(
             url('repo_stats',
                 repo_name=backend.repo_name, commit_id='tip'),
             extra_environ=xhr_header,
             status=200)
 
         expected_data = self.expected_trending[backend.alias]
         returned_stats = response.json['code_stats']
         for k, v in expected_data.items():
             assert v == returned_stats[k]
 
     def test_repo_refs_data(self, backend):
         response = self.app.get(
             url('repo_refs_data', repo_name=backend.repo_name),
             status=200)
 
         # Ensure that there is the correct amount of items in the result
         repo = backend.repo.scm_instance()
         data = response.json['results']
         items = sum(len(section['children']) for section in data)
         repo_refs = len(repo.branches) + len(repo.tags) + len(repo.bookmarks)
         assert items == repo_refs
 
     def test_index_shows_missing_requirements_message(
             self, backend, autologin_user):
         repo_name = backend.repo_name
         scm_patcher = mock.patch.object(
             Repository, 'scm_instance', side_effect=RepositoryRequirementError)
 
         with scm_patcher:
             response = self.app.get(url('summary_home', repo_name=repo_name))
         assert_response = AssertResponse(response)
         assert_response.element_contains(
             '.main .alert-warning strong', 'Missing requirements')
         assert_response.element_contains(
             '.main .alert-warning',
             'These commits cannot be displayed, because this repository'
             ' uses the Mercurial largefiles extension, which was not enabled.')
 
     def test_missing_requirements_page_does_not_contains_switch_to(
             self, backend):
         self.log_user()
         repo_name = backend.repo_name
         scm_patcher = mock.patch.object(
             Repository, 'scm_instance', side_effect=RepositoryRequirementError)
 
         with scm_patcher:
             response = self.app.get(url('summary_home', repo_name=repo_name))
         response.mustcontain(no='Switch To')
 
 
 @pytest.mark.usefixtures('pylonsapp')
 class TestSwitcherReferenceData:
 
     def test_creates_reference_urls_based_on_name(self):
         references = {
             'name': 'commit_id',
         }
         controller = summary.SummaryController()
         is_svn = False
         result = controller._switcher_reference_data(
             'repo_name', references, is_svn)
         expected_url = h.url(
             'files_home', repo_name='repo_name', revision='name',
             at='name')
         assert result[0]['files_url'] == expected_url
 
     def test_urls_contain_commit_id_if_slash_in_name(self):
         references = {
             'name/with/slash': 'commit_id',
         }
         controller = summary.SummaryController()
         is_svn = False
         result = controller._switcher_reference_data(
             'repo_name', references, is_svn)
         expected_url = h.url(
             'files_home', repo_name='repo_name', revision='commit_id',
             at='name/with/slash')
         assert result[0]['files_url'] == expected_url
 
     def test_adds_reference_to_path_for_svn(self):
         references = {
             'name/with/slash': 'commit_id',
         }
         controller = summary.SummaryController()
         is_svn = True
         result = controller._switcher_reference_data(
             'repo_name', references, is_svn)
         expected_url = h.url(
             'files_home', repo_name='repo_name', f_path='name/with/slash',
             revision='commit_id', at='name/with/slash')
         assert result[0]['files_url'] == expected_url
 
 
 @pytest.mark.usefixtures('pylonsapp')
 class TestCreateReferenceData:
 
     @pytest.fixture
     def example_refs(self):
         section_1_refs = OrderedDict((('a', 'a_id'), ('b', 'b_id')))
         example_refs = [
             ('section_1', section_1_refs, 't1'),
             ('section_2', {'c': 'c_id'}, 't2'),
         ]
         return example_refs
 
     def test_generates_refs_based_on_commit_ids(self, example_refs):
         repo = mock.Mock()
         repo.name = 'test-repo'
         repo.alias = 'git'
         full_repo_name = 'pytest-repo-group/' + repo.name
         controller = summary.SummaryController()
 
         result = controller._create_reference_data(
             repo, full_repo_name, example_refs)
 
         expected_files_url = '/{}/files/'.format(full_repo_name)
         expected_result = [
             {
                 'children': [
                     {
                         'id': 'a', 'raw_id': 'a_id', 'text': 'a', 'type': 't1',
                         'files_url': expected_files_url + 'a/?at=a',
                     },
                     {
                         'id': 'b', 'raw_id': 'b_id', 'text': 'b', 'type': 't1',
                         'files_url': expected_files_url + 'b/?at=b',
                     }
                 ],
                 'text': 'section_1'
             },
             {
                 'children': [
                     {
                         'id': 'c', 'raw_id': 'c_id', 'text': 'c', 'type': 't2',
                         'files_url': expected_files_url + 'c/?at=c',
                     }
                 ],
                 'text': 'section_2'
             }]
         assert result == expected_result
 
     def test_generates_refs_with_path_for_svn(self, example_refs):
         repo = mock.Mock()
         repo.name = 'test-repo'
         repo.alias = 'svn'
         full_repo_name = 'pytest-repo-group/' + repo.name
         controller = summary.SummaryController()
         result = controller._create_reference_data(
             repo, full_repo_name, example_refs)
 
         expected_files_url = '/{}/files/'.format(full_repo_name)
         expected_result = [
             {
                 'children': [
                     {
                         'id': 'a@a_id', 'raw_id': 'a_id',
                         'text': 'a', 'type': 't1',
                         'files_url': expected_files_url + 'a_id/a?at=a',
                     },
                     {
                         'id': 'b@b_id', 'raw_id': 'b_id',
                         'text': 'b', 'type': 't1',
                         'files_url': expected_files_url + 'b_id/b?at=b',
                     }
                 ],
                 'text': 'section_1'
             },
             {
                 'children': [
                     {
                         'id': 'c@c_id', 'raw_id': 'c_id',
                         'text': 'c', 'type': 't2',
                         'files_url': expected_files_url + 'c_id/c?at=c',
                     }
                 ],
                 'text': 'section_2'
             }
         ]
         assert result == expected_result
 
 
 @pytest.mark.usefixtures("app")
 class TestRepoLocation:
 
     @pytest.mark.parametrize("suffix", [u'', u'ąęł'], ids=['', 'non-ascii'])
     def test_manual_delete(self, autologin_user, backend, suffix, csrf_token):
         repo = backend.create_repo(name_suffix=suffix)
         repo_name = repo.repo_name
 
         # delete from file system
         RepoModel()._delete_filesystem_repo(repo)
 
         # test if the repo is still in the database
         new_repo = RepoModel().get_by_repo_name(repo_name)
         assert new_repo.repo_name == repo_name
 
         # check if repo is not in the filesystem
         assert not repo_on_filesystem(repo_name)
         self.assert_repo_not_found_redirect(repo_name)
 
     def assert_repo_not_found_redirect(self, repo_name):
         # run the check page that triggers the other flash message
         response = self.app.get(url('repo_check_home', repo_name=repo_name))
         assert_session_flash(
             response, 'The repository at %s cannot be located.' % repo_name)
 
 
-def repo_on_filesystem(repo_name):
-    try:
-        vcs.get_repo(os.path.join(TESTS_TMP_PATH, repo_name))
-        return True
-    except Exception:
-        return False
-
-
 class TestCreateFilesUrl(object):
     def test_creates_non_svn_url(self):
         controller = summary.SummaryController()
         repo = mock.Mock()
         repo.name = 'abcde'
         full_repo_name = 'test-repo-group/' + repo.name
         ref_name = 'branch1'
         raw_id = 'deadbeef0123456789'
         is_svn = False
 
         with mock.patch.object(summary.h, 'url') as url_mock:
             result = controller._create_files_url(
                 repo, full_repo_name, ref_name, raw_id, is_svn)
         url_mock.assert_called_once_with(
             'files_home', repo_name=full_repo_name, f_path='',
             revision=ref_name, at=ref_name)
         assert result == url_mock.return_value
 
     def test_creates_svn_url(self):
         controller = summary.SummaryController()
         repo = mock.Mock()
         repo.name = 'abcde'
         full_repo_name = 'test-repo-group/' + repo.name
         ref_name = 'branch1'
         raw_id = 'deadbeef0123456789'
         is_svn = True
 
         with mock.patch.object(summary.h, 'url') as url_mock:
             result = controller._create_files_url(
                 repo, full_repo_name, ref_name, raw_id, is_svn)
         url_mock.assert_called_once_with(
             'files_home', repo_name=full_repo_name, f_path=ref_name,
             revision=raw_id, at=ref_name)
         assert result == url_mock.return_value
 
     def test_name_has_slashes(self):
         controller = summary.SummaryController()
         repo = mock.Mock()
         repo.name = 'abcde'
         full_repo_name = 'test-repo-group/' + repo.name
         ref_name = 'branch1/branch2'
         raw_id = 'deadbeef0123456789'
         is_svn = False
 
         with mock.patch.object(summary.h, 'url') as url_mock:
             result = controller._create_files_url(
                 repo, full_repo_name, ref_name, raw_id, is_svn)
         url_mock.assert_called_once_with(
             'files_home', repo_name=full_repo_name, f_path='', revision=raw_id,
             at=ref_name)
         assert result == url_mock.return_value
 
 
 class TestReferenceItems(object):
     repo = mock.Mock()
     repo.name = 'pytest-repo'
     repo_full_name = 'pytest-repo-group/' + repo.name
     ref_type = 'branch'
     fake_url = '/abcde/'
 
     @staticmethod
     def _format_function(name, id_):
         return 'format_function_{}_{}'.format(name, id_)
 
     def test_creates_required_amount_of_items(self):
         amount = 100
         refs = {
             'ref{}'.format(i): '{0:040d}'.format(i)
             for i in range(amount)
         }
 
         controller = summary.SummaryController()
 
         url_patcher = mock.patch.object(
             controller, '_create_files_url')
         svn_patcher = mock.patch.object(
             summary.h, 'is_svn', return_value=False)
 
         with url_patcher as url_mock, svn_patcher:
             result = controller._create_reference_items(
                 self.repo, self.repo_full_name, refs, self.ref_type,
                 self._format_function)
         assert len(result) == amount
         assert url_mock.call_count == amount
 
     def test_single_item_details(self):
         ref_name = 'ref1'
         ref_id = 'deadbeef'
         refs = {
             ref_name: ref_id
         }
 
         controller = summary.SummaryController()
         url_patcher = mock.patch.object(
             controller, '_create_files_url', return_value=self.fake_url)
         svn_patcher = mock.patch.object(
             summary.h, 'is_svn', return_value=False)
 
         with url_patcher as url_mock, svn_patcher:
             result = controller._create_reference_items(
                 self.repo, self.repo_full_name, refs, self.ref_type,
                 self._format_function)
 
         url_mock.assert_called_once_with(
             self.repo, self.repo_full_name, ref_name, ref_id, False)
         expected_result = [
             {
                 'text': ref_name,
                 'id': self._format_function(ref_name, ref_id),
                 'raw_id': ref_id,
                 'type': self.ref_type,
                 'files_url': self.fake_url
             }
         ]
         assert result == expected_result

rhodecode/tests/utils.py

@@ -1,285 +1,293 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2010-2016  RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # 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 Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import threading
 import time
 import logging
 import os.path
 import subprocess
 import urllib2
 from urlparse import urlparse, parse_qsl
 from urllib import unquote_plus
 
 import pytest
 import rc_testdata
 from lxml.html import fromstring, tostring
 from lxml.cssselect import CSSSelector
 
 from rhodecode.model.db import User
 from rhodecode.model.meta import Session
 from rhodecode.model.scm import ScmModel
 from rhodecode.lib.vcs.backends.svn.repository import SubversionRepository
 
 
 log = logging.getLogger(__name__)
 
 
 def set_anonymous_access(enabled):
     """(Dis)allows anonymous access depending on parameter `enabled`"""
     user = User.get_default_user()
     user.active = enabled
     Session().add(user)
     Session().commit()
     log.info('anonymous access is now: %s', enabled)
     assert enabled == User.get_default_user().active, (
         'Cannot set anonymous access')
 
 
 def check_xfail_backends(node, backend_alias):
     # Using "xfail_backends" here intentionally, since this marks work
     # which is "to be done" soon.
     skip_marker = node.get_marker('xfail_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Support for backend %s to be developed." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.xfail(msg)
 
 
 def check_skip_backends(node, backend_alias):
     # Using "skip_backends" here intentionally, since this marks work which is
     # not supported.
     skip_marker = node.get_marker('skip_backends')
     if skip_marker and backend_alias in skip_marker.args:
         msg = "Feature not supported for backend %s." % (backend_alias, )
         msg = skip_marker.kwargs.get('reason', msg)
         pytest.skip(msg)
 
 
 def extract_git_repo_from_dump(dump_name, repo_name):
     """Create git repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_git_dump(dump_name, target_path)
     return target_path
 
 
 def extract_hg_repo_from_dump(dump_name, repo_name):
     """Create hg repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     rc_testdata.extract_hg_dump(dump_name, target_path)
     return target_path
 
 
 def extract_svn_repo_from_dump(dump_name, repo_name):
     """Create a svn repo `repo_name` from dump `dump_name`."""
     repos_path = ScmModel().repos_path
     target_path = os.path.join(repos_path, repo_name)
     SubversionRepository(target_path, create=True)
     _load_svn_dump_into_repo(dump_name, target_path)
     return target_path
 
 
 def assert_message_in_log(log_records, message, levelno, module):
     messages = [
         r.message for r in log_records
         if r.module == module and r.levelno == levelno
     ]
     assert message in messages
 
 
 def _load_svn_dump_into_repo(dump_name, repo_path):
     """
     Utility to populate a svn repository with a named dump
 
     Currently the dumps are in rc_testdata. They might later on be
     integrated with the main repository once they stabilize more.
     """
     dump = rc_testdata.load_svn_dump(dump_name)
     load_dump = subprocess.Popen(
         ['svnadmin', 'load', repo_path],
         stdin=subprocess.PIPE, stdout=subprocess.PIPE,
         stderr=subprocess.PIPE)
     out, err = load_dump.communicate(dump)
     if load_dump.returncode != 0:
         log.error("Output of load_dump command: %s", out)
         log.error("Error output of load_dump command: %s", err)
         raise Exception(
             'Failed to load dump "%s" into repository at path "%s".'
             % (dump_name, repo_path))
 
 
 class AssertResponse(object):
     """
     Utility that helps to assert things about a given HTML response.
     """
 
     def __init__(self, response):
         self.response = response
 
     def one_element_exists(self, css_selector):
         self.get_element(css_selector)
 
     def no_element_exists(self, css_selector):
         assert not self._get_elements(css_selector)
 
     def element_equals_to(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         element_text = self._element_to_string(element)
         assert expected_content in element_text
 
     def element_contains(self, css_selector, expected_content):
         element = self.get_element(css_selector)
         assert expected_content in element.text_content()
 
     def contains_one_link(self, link_text, href):
         doc = fromstring(self.response.body)
         sel = CSSSelector('a[href]')
         elements = [
             e for e in sel(doc) if e.text_content().strip() == link_text]
         assert len(elements) == 1, "Did not find link or found multiple links"
         self._ensure_url_equal(elements[0].attrib.get('href'), href)
 
     def contains_one_anchor(self, anchor_id):
         doc = fromstring(self.response.body)
         sel = CSSSelector('#' + anchor_id)
         elements = sel(doc)
         assert len(elements) == 1
 
     def _ensure_url_equal(self, found, expected):
         assert _Url(found) == _Url(expected)
 
     def get_element(self, css_selector):
         elements = self._get_elements(css_selector)
         assert len(elements) == 1
         return elements[0]
 
     def get_elements(self, css_selector):
         return self._get_elements(css_selector)
 
     def _get_elements(self, css_selector):
         doc = fromstring(self.response.body)
         sel = CSSSelector(css_selector)
         elements = sel(doc)
         return elements
 
     def _element_to_string(self, element):
         return tostring(element)
 
 
 class _Url(object):
     """
     A url object that can be compared with other url orbjects
     without regard to the vagaries of encoding, escaping, and ordering
     of parameters in query strings.
 
     Inspired by
     http://stackoverflow.com/questions/5371992/comparing-two-urls-in-python
     """
 
     def __init__(self, url):
         parts = urlparse(url)
         _query = frozenset(parse_qsl(parts.query))
         _path = unquote_plus(parts.path)
         parts = parts._replace(query=_query, path=_path)
         self.parts = parts
 
     def __eq__(self, other):
         return self.parts == other.parts
 
     def __hash__(self):
         return hash(self.parts)
 
 
 def run_test_concurrently(times, raise_catched_exc=True):
     """
     Add this decorator to small pieces of code that you want to test
     concurrently
 
     ex:
 
     @test_concurrently(25)
     def my_test_function():
         ...
     """
     def test_concurrently_decorator(test_func):
         def wrapper(*args, **kwargs):
             exceptions = []
 
             def call_test_func():
                 try:
                     test_func(*args, **kwargs)
                 except Exception, e:
                     exceptions.append(e)
                     if raise_catched_exc:
                         raise
             threads = []
             for i in range(times):
                 threads.append(threading.Thread(target=call_test_func))
             for t in threads:
                 t.start()
             for t in threads:
                 t.join()
             if exceptions:
                 raise Exception(
                     'test_concurrently intercepted %s exceptions: %s' % (
                         len(exceptions), exceptions))
         return wrapper
     return test_concurrently_decorator
 
 
 def wait_for_url(url, timeout=10):
     """
     Wait until URL becomes reachable.
 
     It polls the URL until the timeout is reached or it became reachable.
     If will call to `py.test.fail` in case the URL is not reachable.
     """
     timeout = time.time() + timeout
     last = 0
     wait = 0.1
 
     while (timeout > last):
         last = time.time()
         if is_url_reachable(url):
             break
         elif ((last + wait) > time.time()):
             # Go to sleep because not enough time has passed since last check.
             time.sleep(wait)
     else:
         pytest.fail("Timeout while waiting for URL {}".format(url))
 
 
 def is_url_reachable(url):
     try:
         urllib2.urlopen(url)
     except urllib2.URLError:
         return False
     return True
 
 
 def get_session_from_response(response):
     """
     This returns the session from a response object. Pylons has some magic
     to make the session available as `response.session`. But pyramid
     doesn't expose it.
     """
     # TODO: Try to look up the session key also.
     return response.request.environ['beaker.session']
+
+
+def repo_on_filesystem(repo_name):
+    from rhodecode.lib import vcs
+    from rhodecode.tests import TESTS_TMP_PATH
+    repo = vcs.get_vcs_instance(
+        os.path.join(TESTS_TMP_PATH, repo_name), create=False)
+    return repo is not None