Files @ ae5ac36cdf83
Branch filter:

Location: kallithea/rhodecode/tests/functional/test_forks.py

Mads Kiilerich
pull request: use unionrepo instead of outgoing

This makes it possible to look the 'moving target' symbols up in the right repo.

Using a revset with the right revisions also removes the need for pruning
changesets that are outside the requested range.

It will also not be confused by changesets that for some reason has been pulled
to the repo but haven't been merged yet. They are going to be 'merged' by the
'pull' request and should thus be a part of what is reviewed.
from rhodecode.tests import *

from rhodecode.model.db import Repository
from rhodecode.model.repo import RepoModel
from rhodecode.model.user import UserModel
from rhodecode.model.meta import Session


class TestForksController(TestController):

    def setUp(self):
        self.username = u'forkuser'
        self.password = u'qweqwe'
        self.u1 = UserModel().create_or_update(
            username=self.username, password=self.password,
            email=u'fork_king@rhodecode.org', firstname=u'u1', lastname=u'u1'
        )
        Session().commit()

    def tearDown(self):
        Session().delete(self.u1)
        Session().commit()

    def test_index(self):
        self.log_user()
        repo_name = HG_REPO
        response = self.app.get(url(controller='forks', action='forks',
                                    repo_name=repo_name))

        self.assertTrue("""There are no forks yet""" in response.body)

    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 = HG_REPO
        self.app.post(url(controller='forks', action='fork_create',
                          repo_name=repo_name), {}, status=403)

    def test_index_with_fork_hg(self):
        self.log_user()

        # create a fork
        fork_name = HG_FORK
        description = 'fork of vcs test'
        repo_name = HG_REPO
        org_repo = Repository.get_by_repo_name(repo_name)
        response = self.app.post(url(controller='forks',
                                     action='fork_create',
                                    repo_name=repo_name),
                                    {'repo_name': fork_name,
                                     'repo_group': '',
                                     'fork_parent_id': org_repo.repo_id,
                                     'repo_type': 'hg',
                                     'description': description,
                                     'private': 'False',
                                     'landing_rev': 'tip'})

        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
        response = self.app.delete(url('repo', repo_name=fork_name))

    def test_index_with_fork_git(self):
        self.log_user()

        # create a fork
        fork_name = GIT_FORK
        description = 'fork of vcs test'
        repo_name = GIT_REPO
        org_repo = Repository.get_by_repo_name(repo_name)
        response = self.app.post(url(controller='forks',
                                     action='fork_create',
                                    repo_name=repo_name),
                                    {'repo_name': fork_name,
                                     'repo_group': '',
                                     'fork_parent_id': org_repo.repo_id,
                                     'repo_type': 'git',
                                     'description': description,
                                     'private': 'False',
                                     'landing_rev': 'tip'})

        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
        response = self.app.delete(url('repo', repo_name=fork_name))

    def test_z_fork_create(self):
        self.log_user()
        fork_name = HG_FORK
        description = 'fork of vcs test'
        repo_name = HG_REPO
        org_repo = Repository.get_by_repo_name(repo_name)
        response = self.app.post(url(controller='forks', action='fork_create',
                                    repo_name=repo_name),
                                    {'repo_name':fork_name,
                                     'repo_group':'',
                                     'fork_parent_id':org_repo.repo_id,
                                     'repo_type':'hg',
                                     'description':description,
                                     'private':'False',
                                     'landing_rev': 'tip'})

        #test if we have a message that fork is ok
        self.checkSessionFlash(response,
                'forked %s repository as %s' % (repo_name, fork_name))

        #test if the fork was created in the database
        fork_repo = Session().query(Repository)\
            .filter(Repository.repo_name == fork_name).one()

        self.assertEqual(fork_repo.repo_name, fork_name)
        self.assertEqual(fork_repo.fork.repo_name, repo_name)

        #test if fork is visible in the list ?
        response = response.follow()

        response = self.app.get(url(controller='summary', action='index',
                                    repo_name=fork_name))

        self.assertTrue('Fork of %s' % repo_name in response.body)

    def test_zz_fork_permission_page(self):
        usr = self.log_user(self.username, self.password)['user_id']
        repo_name = HG_REPO

        forks = Session().query(Repository)\
            .filter(Repository.fork_id != None)\
            .all()
        self.assertEqual(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('<div style="padding:5px 3px 3px 42px;">fork of vcs test</div>')

    def test_zzz_fork_permission_page(self):
        usr = self.log_user(self.username, self.password)['user_id']
        repo_name = HG_REPO

        forks = Session().query(Repository)\
            .filter(Repository.fork_id != None)\
            .all()
        self.assertEqual(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')