Files @ fc38df41b9b7
Branch filter:

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

Marcin Kuzminski
Reverted delete zombie repos on server start, this can be misleading, and somehow dangerous option.
It's better that server restarts as fast as possible without any magical functions going on. Cleanup can be still made from rescan option in admin
from rhodecode.tests import *

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


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'
        )
        self.Session.commit()

    def tearDown(self):
        self.Session.delete(self.u1)
        self.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_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/summary">%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/summary">%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 = self.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 = self.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')
        self.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 = self.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')
        self.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')