Files @ 60335b702a00
Branch filter:

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

Mads Kiilerich
invalidation: don't create CacheInvalidation records on startup

Creating the records early gave an advantage before lightweight was introduced.
With lightweight it is no longer necessary.

The records will be created on demand anyway and there is no reason to create and
maintain them before they are used.
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))

        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 = 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 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()

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

        response.mustcontain('Fork of %s' % repo_name)

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