rhodecode/apps/repository/tests/test_repo_forks.py

# Copyright (C) 2010-2023 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 TestController, assert_session_flash, HG_FORK, GIT_FORK

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.request
    import urllib.parse
    import urllib.error

    base_url = {
        'repo_summary': '/{repo_name}',
        'repo_creating_check': '/{repo_name}/repo_creating_check',
        'repo_fork_new': '/{repo_name}/fork',
        'repo_fork_create': '/{repo_name}/fork/create',
        'repo_forks_show_all': '/{repo_name}/forks',
        'repo_forks_data': '/{repo_name}/forks/data',
    }[name].format(**kwargs)

    if params:
        base_url = '{}?{}'.format(base_url, urllib.parse.urlencode(params))
    return base_url


FORK_NAME = {
    'hg': HG_FORK,
    'git': GIT_FORK
}


@pytest.mark.skip_backends('svn')
class TestRepoForkViewTests(TestController):

    def test_show_forks(self, backend, xhr_header):
        self.log_user()
        response = self.app.get(
            route_path('repo_forks_data', repo_name=backend.repo_name),
            extra_environ=xhr_header)

        assert response.json == {u'data': [], u'draw': None,
                                 u'recordsFiltered': 0, u'recordsTotal': 0}

    def test_no_permissions_to_fork_page(self, backend, user_util):
        user = user_util.create_user(password='qweqwe')
        user_id = user.user_id
        self.log_user(user.username, 'qweqwe')

        user_model = UserModel()
        user_model.revoke_perm(user_id, 'hg.fork.repository')
        user_model.grant_perm(user_id, 'hg.fork.none')
        u = UserModel().get(user_id)
        u.inherit_default_permissions = False
        Session().commit()
        # try create a fork
        self.app.get(
            route_path('repo_fork_new', repo_name=backend.repo_name),
            status=404)

    def test_no_permissions_to_fork_submit(self, backend, csrf_token, user_util):
        user = user_util.create_user(password='qweqwe')
        user_id = user.user_id
        self.log_user(user.username, 'qweqwe')

        user_model = UserModel()
        user_model.revoke_perm(user_id, 'hg.fork.repository')
        user_model.grant_perm(user_id, 'hg.fork.none')
        u = UserModel().get(user_id)
        u.inherit_default_permissions = False
        Session().commit()
        # try create a fork
        self.app.post(
            route_path('repo_fork_create', repo_name=backend.repo_name),
            {'csrf_token': csrf_token},
            status=404)

    def test_fork_missing_data(self, autologin_user, backend, csrf_token):
        # try create a fork
        response = self.app.post(
            route_path('repo_fork_create', repo_name=backend.repo_name),
            {'csrf_token': csrf_token},
            status=200)
        # test if html fill works fine
        response.mustcontain('Missing value')

    def test_create_fork_page(self, autologin_user, backend):
        self.app.get(
            route_path('repo_fork_new', repo_name=backend.repo_name),
            status=200)

    def test_create_and_show_fork(
            self, autologin_user, backend, csrf_token, xhr_header):

        # create a fork
        fork_name = FORK_NAME[backend.alias]
        description = 'fork of vcs test'
        repo_name = backend.repo_name
        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': backend.alias,
            'description': description,
            'private': 'False',
            'csrf_token': csrf_token,
        }

        self.app.post(
            route_path('repo_fork_create', repo_name=repo_name), creation_args)

        response = self.app.get(
            route_path('repo_forks_data', repo_name=repo_name),
            extra_environ=xhr_header)

        assert response.json['data'][0]['fork_name'] == \
            """<a href="/%s">%s</a>""" % (fork_name, fork_name)

        # remove this fork
        fixture.destroy_repo(fork_name)

    def test_fork_create(self, autologin_user, backend, csrf_token):
        fork_name = FORK_NAME[backend.alias]
        description = 'fork of vcs test'
        repo_name = backend.repo_name
        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': backend.alias,
            'description': description,
            'private': 'False',
            'csrf_token': csrf_token,
        }
        self.app.post(
            route_path('repo_fork_create', repo_name=repo_name), creation_args)
        repo = Repository.get_by_repo_name(FORK_NAME[backend.alias])
        assert repo.fork.repo_name == backend.repo_name

        # 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(backend.alias)
        response.mustcontain('Fork of')
        response.mustcontain('<a href="/%s">%s</a>' % (repo_name, repo_name))

    def test_fork_create_into_group(self, autologin_user, backend, csrf_token):
        group = fixture.create_repo_group('vc')
        group_id = group.group_id
        fork_name = FORK_NAME[backend.alias]
        fork_name_full = 'vc/%s' % fork_name
        description = 'fork of vcs test'
        repo_name = backend.repo_name
        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': backend.alias,
            'description': description,
            'private': 'False',
            'csrf_token': csrf_token,
        }
        self.app.post(
            route_path('repo_fork_create', repo_name=repo_name), creation_args)
        repo = Repository.get_by_repo_name(fork_name_full)
        assert repo.fork.repo_name == backend.repo_name

        # 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(backend.alias)

        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_fork_read_permission(self, backend, xhr_header, user_util):
        user = user_util.create_user(password='qweqwe')
        user_id = user.user_id
        self.log_user(user.username, 'qweqwe')

        # create a fake fork
        fork = user_util.create_repo(repo_type=backend.alias)
        source = user_util.create_repo(repo_type=backend.alias)
        repo_name = source.repo_name

        fork.fork_id = source.repo_id
        fork_name = fork.repo_name
        Session().commit()

        forks = Repository.query()\
            .filter(Repository.repo_type == backend.alias)\
            .filter(Repository.fork_id == source.repo_id).all()
        assert 1 == len(forks)

        # set read permissions for this
        RepoModel().grant_user_permission(
            repo=forks[0], user=user_id, perm='repository.read')
        Session().commit()

        response = self.app.get(
            route_path('repo_forks_data', repo_name=repo_name),
            extra_environ=xhr_header)

        assert response.json['data'][0]['fork_name'] == \
            """<a href="/%s">%s</a>""" % (fork_name, fork_name)

    def test_fork_none_permission(self, backend, xhr_header, user_util):
        user = user_util.create_user(password='qweqwe')
        user_id = user.user_id
        self.log_user(user.username, 'qweqwe')

        # create a fake fork
        fork = user_util.create_repo(repo_type=backend.alias)
        source = user_util.create_repo(repo_type=backend.alias)
        repo_name = source.repo_name

        fork.fork_id = source.repo_id

        Session().commit()

        forks = Repository.query()\
            .filter(Repository.repo_type == backend.alias)\
            .filter(Repository.fork_id == source.repo_id).all()
        assert 1 == len(forks)

        # set none
        RepoModel().grant_user_permission(
            repo=forks[0], user=user_id, perm='repository.none')
        Session().commit()

        # fork shouldn't be there
        response = self.app.get(
            route_path('repo_forks_data', repo_name=repo_name),
            extra_environ=xhr_header)

        assert response.json == {u'data': [], u'draw': None,
                                 u'recordsFiltered': 0, u'recordsTotal': 0}

    @pytest.mark.parametrize('url_type', [
        'repo_fork_new',
        'repo_fork_create'
    ])
    def test_fork_is_forbidden_on_archived_repo(self, backend, xhr_header, user_util, url_type):
        user = user_util.create_user(password='qweqwe')
        self.log_user(user.username, 'qweqwe')

        # create a temporary repo
        source = user_util.create_repo(repo_type=backend.alias)
        repo_name = source.repo_name
        repo = Repository.get_by_repo_name(repo_name)
        repo.archived = True
        Session().commit()

        response = self.app.get(
            route_path(url_type, repo_name=repo_name), status=302)

        msg = 'Action not supported for archived repository.'
        assert_session_flash(response, msg)


class TestSVNFork(TestController):
    @pytest.mark.parametrize('route_name', [
        'repo_fork_create', 'repo_fork_new'
    ])
    def test_fork_redirects(self, autologin_user, backend_svn, route_name):

        self.app.get(route_path(
            route_name, repo_name=backend_svn.repo_name),
            status=404)