forks: moved pylons code into pyramid.
This commit is contained in:
Marcin Kuzminski
parent
81eb5a6b72
commit
05386904e7
15 changed files with 683 additions and 609 deletions
|
|
@ -1,288 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright (C) 2010-2017 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 pytest
|
||||
|
||||
from rhodecode.tests import *
|
||||
from rhodecode.tests.fixture import Fixture
|
||||
from rhodecode.lib import helpers as h
|
||||
|
||||
from rhodecode.model.db import Repository
|
||||
from rhodecode.model.repo import RepoModel
|
||||
from rhodecode.model.user import UserModel
|
||||
from rhodecode.model.meta import Session
|
||||
|
||||
fixture = Fixture()
|
||||
|
||||
|
||||
def route_path(name, params=None, **kwargs):
|
||||
import urllib
|
||||
|
||||
base_url = {
|
||||
'repo_summary': '/{repo_name}',
|
||||
'repo_creating_check': '/{repo_name}/repo_creating_check',
|
||||
}[name].format(**kwargs)
|
||||
|
||||
if params:
|
||||
base_url = '{}?{}'.format(base_url, urllib.urlencode(params))
|
||||
return base_url
|
||||
|
||||
|
||||
class _BaseTest(TestController):
|
||||
|
||||
REPO = None
|
||||
REPO_TYPE = None
|
||||
NEW_REPO = None
|
||||
REPO_FORK = None
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def prepare(self, request, pylonsapp):
|
||||
self.username = u'forkuser'
|
||||
self.password = u'qweqwe'
|
||||
self.u1 = fixture.create_user(self.username, password=self.password,
|
||||
email=u'fork_king@rhodecode.org')
|
||||
Session().commit()
|
||||
self.u1id = self.u1.user_id
|
||||
request.addfinalizer(self.cleanup)
|
||||
|
||||
def cleanup(self):
|
||||
u1 = UserModel().get(self.u1id)
|
||||
Session().delete(u1)
|
||||
Session().commit()
|
||||
|
||||
def test_index(self):
|
||||
self.log_user()
|
||||
repo_name = self.REPO
|
||||
response = self.app.get(url(controller='forks', action='forks',
|
||||
repo_name=repo_name))
|
||||
|
||||
response.mustcontain("""There are no forks yet""")
|
||||
|
||||
def test_no_permissions_to_fork(self):
|
||||
usr = self.log_user(TEST_USER_REGULAR_LOGIN,
|
||||
TEST_USER_REGULAR_PASS)['user_id']
|
||||
user_model = UserModel()
|
||||
user_model.revoke_perm(usr, 'hg.fork.repository')
|
||||
user_model.grant_perm(usr, 'hg.fork.none')
|
||||
u = UserModel().get(usr)
|
||||
u.inherit_default_permissions = False
|
||||
Session().commit()
|
||||
# try create a fork
|
||||
repo_name = self.REPO
|
||||
self.app.post(
|
||||
url(controller='forks', action='fork_create', repo_name=repo_name),
|
||||
{'csrf_token': self.csrf_token}, status=404)
|
||||
|
||||
def test_index_with_fork(self):
|
||||
self.log_user()
|
||||
|
||||
# create a fork
|
||||
fork_name = self.REPO_FORK
|
||||
description = 'fork of vcs test'
|
||||
repo_name = self.REPO
|
||||
source_repo = Repository.get_by_repo_name(repo_name)
|
||||
creation_args = {
|
||||
'repo_name': fork_name,
|
||||
'repo_group': '',
|
||||
'fork_parent_id': source_repo.repo_id,
|
||||
'repo_type': self.REPO_TYPE,
|
||||
'description': description,
|
||||
'private': 'False',
|
||||
'landing_rev': 'rev:tip',
|
||||
'csrf_token': self.csrf_token,
|
||||
}
|
||||
|
||||
self.app.post(url(controller='forks', action='fork_create',
|
||||
repo_name=repo_name), creation_args)
|
||||
|
||||
response = self.app.get(url(controller='forks', action='forks',
|
||||
repo_name=repo_name))
|
||||
|
||||
response.mustcontain(
|
||||
"""<a href="/%s">%s</a>""" % (fork_name, fork_name)
|
||||
)
|
||||
|
||||
# remove this fork
|
||||
fixture.destroy_repo(fork_name)
|
||||
|
||||
def test_fork_create_into_group(self):
|
||||
self.log_user()
|
||||
group = fixture.create_repo_group('vc')
|
||||
group_id = group.group_id
|
||||
fork_name = self.REPO_FORK
|
||||
fork_name_full = 'vc/%s' % fork_name
|
||||
description = 'fork of vcs test'
|
||||
repo_name = self.REPO
|
||||
source_repo = Repository.get_by_repo_name(repo_name)
|
||||
creation_args = {
|
||||
'repo_name': fork_name,
|
||||
'repo_group': group_id,
|
||||
'fork_parent_id': source_repo.repo_id,
|
||||
'repo_type': self.REPO_TYPE,
|
||||
'description': description,
|
||||
'private': 'False',
|
||||
'landing_rev': 'rev:tip',
|
||||
'csrf_token': self.csrf_token,
|
||||
}
|
||||
self.app.post(url(controller='forks', action='fork_create',
|
||||
repo_name=repo_name), creation_args)
|
||||
repo = Repository.get_by_repo_name(fork_name_full)
|
||||
assert repo.fork.repo_name == self.REPO
|
||||
|
||||
# run the check page that triggers the flash message
|
||||
response = self.app.get(
|
||||
route_path('repo_creating_check', repo_name=fork_name_full))
|
||||
# test if we have a message that fork is ok
|
||||
assert_session_flash(response,
|
||||
'Forked repository %s as <a href="/%s">%s</a>'
|
||||
% (repo_name, fork_name_full, fork_name_full))
|
||||
|
||||
# test if the fork was created in the database
|
||||
fork_repo = Session().query(Repository)\
|
||||
.filter(Repository.repo_name == fork_name_full).one()
|
||||
|
||||
assert fork_repo.repo_name == fork_name_full
|
||||
assert fork_repo.fork.repo_name == repo_name
|
||||
|
||||
# test if the repository is visible in the list ?
|
||||
response = self.app.get(h.route_path('repo_summary', repo_name=fork_name_full))
|
||||
response.mustcontain(fork_name_full)
|
||||
response.mustcontain(self.REPO_TYPE)
|
||||
|
||||
response.mustcontain('Fork of')
|
||||
response.mustcontain('<a href="/%s">%s</a>' % (repo_name, repo_name))
|
||||
|
||||
fixture.destroy_repo(fork_name_full)
|
||||
fixture.destroy_repo_group(group_id)
|
||||
|
||||
def test_z_fork_create(self):
|
||||
self.log_user()
|
||||
fork_name = self.REPO_FORK
|
||||
description = 'fork of vcs test'
|
||||
repo_name = self.REPO
|
||||
source_repo = Repository.get_by_repo_name(repo_name)
|
||||
creation_args = {
|
||||
'repo_name': fork_name,
|
||||
'repo_group': '',
|
||||
'fork_parent_id': source_repo.repo_id,
|
||||
'repo_type': self.REPO_TYPE,
|
||||
'description': description,
|
||||
'private': 'False',
|
||||
'landing_rev': 'rev:tip',
|
||||
'csrf_token': self.csrf_token,
|
||||
}
|
||||
self.app.post(url(controller='forks', action='fork_create',
|
||||
repo_name=repo_name), creation_args)
|
||||
repo = Repository.get_by_repo_name(self.REPO_FORK)
|
||||
assert repo.fork.repo_name == self.REPO
|
||||
|
||||
# run the check page that triggers the flash message
|
||||
response = self.app.get(
|
||||
route_path('repo_creating_check', repo_name=fork_name))
|
||||
# test if we have a message that fork is ok
|
||||
assert_session_flash(response,
|
||||
'Forked repository %s as <a href="/%s">%s</a>'
|
||||
% (repo_name, fork_name, fork_name))
|
||||
|
||||
# test if the fork was created in the database
|
||||
fork_repo = Session().query(Repository)\
|
||||
.filter(Repository.repo_name == fork_name).one()
|
||||
|
||||
assert fork_repo.repo_name == fork_name
|
||||
assert fork_repo.fork.repo_name == repo_name
|
||||
|
||||
# test if the repository is visible in the list ?
|
||||
response = self.app.get(h.route_path('repo_summary', repo_name=fork_name))
|
||||
response.mustcontain(fork_name)
|
||||
response.mustcontain(self.REPO_TYPE)
|
||||
response.mustcontain('Fork of')
|
||||
response.mustcontain('<a href="/%s">%s</a>' % (repo_name, repo_name))
|
||||
|
||||
def test_zz_fork_permission_page(self):
|
||||
usr = self.log_user(self.username, self.password)['user_id']
|
||||
repo_name = self.REPO
|
||||
|
||||
forks = Repository.query()\
|
||||
.filter(Repository.repo_type == self.REPO_TYPE)\
|
||||
.filter(Repository.fork_id != None).all()
|
||||
assert 1 == len(forks)
|
||||
|
||||
# set read permissions for this
|
||||
RepoModel().grant_user_permission(repo=forks[0],
|
||||
user=usr,
|
||||
perm='repository.read')
|
||||
Session().commit()
|
||||
|
||||
response = self.app.get(url(controller='forks', action='forks',
|
||||
repo_name=repo_name))
|
||||
|
||||
response.mustcontain('fork of vcs test')
|
||||
|
||||
def test_zzz_fork_permission_page(self):
|
||||
usr = self.log_user(self.username, self.password)['user_id']
|
||||
repo_name = self.REPO
|
||||
|
||||
forks = Repository.query()\
|
||||
.filter(Repository.repo_type == self.REPO_TYPE)\
|
||||
.filter(Repository.fork_id != None).all()
|
||||
assert 1 == len(forks)
|
||||
|
||||
# set none
|
||||
RepoModel().grant_user_permission(repo=forks[0],
|
||||
user=usr, perm='repository.none')
|
||||
Session().commit()
|
||||
# fork shouldn't be there
|
||||
response = self.app.get(url(controller='forks', action='forks',
|
||||
repo_name=repo_name))
|
||||
response.mustcontain('There are no forks yet')
|
||||
|
||||
|
||||
class TestGIT(_BaseTest):
|
||||
REPO = GIT_REPO
|
||||
NEW_REPO = NEW_GIT_REPO
|
||||
REPO_TYPE = 'git'
|
||||
REPO_FORK = GIT_FORK
|
||||
|
||||
|
||||
class TestHG(_BaseTest):
|
||||
REPO = HG_REPO
|
||||
NEW_REPO = NEW_HG_REPO
|
||||
REPO_TYPE = 'hg'
|
||||
REPO_FORK = HG_FORK
|
||||
|
||||
|
||||
@pytest.mark.usefixtures('app', 'autologin_user')
|
||||
@pytest.mark.skip_backends('git','hg')
|
||||
class TestSVNFork(object):
|
||||
|
||||
def test_fork_redirects(self, backend):
|
||||
denied_actions = ['fork','fork_create']
|
||||
for action in denied_actions:
|
||||
response = self.app.get(url(
|
||||
controller='forks', action=action,
|
||||
repo_name=backend.repo_name))
|
||||
assert response.status_int == 302
|
||||
|
||||
# Not allowed, redirect to the summary
|
||||
redirected = response.follow()
|
||||
summary_url = h.route_path('repo_summary', repo_name=backend.repo_name)
|
||||
|
||||
# URL adds leading slash and path doesn't have it
|
||||
assert redirected.request.path == summary_url
|
||||
|
|
@ -18,15 +18,10 @@
|
|||
# RhodeCode Enterprise Edition, including its added features, Support services,
|
||||
# and proprietary license terms, please see https://rhodecode.com/licenses/
|
||||
|
||||
import mock
|
||||
import pytest
|
||||
from webob.exc import HTTPNotFound
|
||||
|
||||
import rhodecode
|
||||
from rhodecode.model.db import Integration
|
||||
from rhodecode.model.meta import Session
|
||||
from rhodecode.tests import assert_session_flash, url, TEST_USER_ADMIN_LOGIN
|
||||
from rhodecode.tests.utils import AssertResponse
|
||||
from rhodecode.integrations import integration_type_registry
|
||||
from rhodecode.config.routing import ADMIN_PREFIX
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue