rhodecode/apps/repository/views/repo_branches.py

# Copyright (C) 2011-2024 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 logging

from pyramid.httpexceptions import HTTPFound

from rhodecode.apps._base import BaseReferencesView
from rhodecode.lib import ext_json
from rhodecode.lib import helpers as h
from rhodecode.lib.auth import (LoginRequired, HasRepoPermissionAnyDecorator, CSRFRequired)
from rhodecode.model.scm import ScmModel
from rhodecode.model.meta import Session
from rhodecode.model.db import PullRequest

log = logging.getLogger(__name__)


class RepoBranchesView(BaseReferencesView):

    @LoginRequired()
    @HasRepoPermissionAnyDecorator(
        'repository.read', 'repository.write', 'repository.admin')
    def branches(self):
        partial_render = self.request.get_partial_renderer(
            'rhodecode:templates/data_table/_dt_elements.mako')
        repo_name = self.db_repo_name
        c = self.load_default_context()
        self._prepare_and_set_clone_url(c)
        c.rhodecode_repo = self.rhodecode_vcs_repo
        c.repository_forks = ScmModel().get_forks(self.db_repo)
        ref_items = self.rhodecode_vcs_repo.branches_all.items()
        data = self.load_refs_context(
            ref_items=ref_items, partials_template='branches/branches_data.mako')
        data_with_actions = []
        if self.db_repo.repo_type != 'svn':
            for branch in data:
                branch['action'] = partial_render(
                    f"branch_actions_{self.db_repo.repo_type}", branch['name_raw'], repo_name, closed=branch['closed']
                )
                data_with_actions.append(branch)
            data = data_with_actions
        c.has_references = bool(data)
        c.data = ext_json.str_json(data)
        return self._get_template_context(c)

    @LoginRequired()
    @HasRepoPermissionAnyDecorator('repository.write', 'repository.admin')
    @CSRFRequired()
    def remove_branch(self):
        _ = self.request.translate
        self.load_default_context()
        repo = self.db_repo
        repo_name = self.db_repo_name
        repo_type = repo.repo_type
        action = _('deleted') if repo_type == 'git' else _('closed')
        redirect = HTTPFound(location=self.request.route_path('branches_home', repo_name=repo_name))
        branch_name = self.request.matchdict.get('branch_name')
        if repo.landing_ref_name == branch_name:
            h.flash(
                _("This branch {} cannot be removed as it's currently set as landing branch").format(branch_name),
                category='error'
            )
            return redirect
        if prs_related_to := Session().query(PullRequest).filter(PullRequest.target_repo_id == repo.repo_id,
                                                                 PullRequest.status != PullRequest.STATUS_CLOSED).filter(
            (PullRequest.source_ref.like(f'branch:{branch_name}:%')) | (
                    PullRequest.target_ref.like(f'branch:{branch_name}:%'))
        ).all():
            h.flash(_("Branch cannot be {} - it's used in following open Pull Request ids: {}").format(action, ','.join(
                map(str, prs_related_to))), category='error')
            return redirect

        match repo_type:
            case 'git':
                self.rhodecode_vcs_repo.delete_branch(branch_name)
            case 'hg':
                from rhodecode.lib.vcs.backends.base import Reference
                self.rhodecode_vcs_repo._local_close(
                    source_ref=Reference(type='branch', name=branch_name,
                                         commit_id=self.rhodecode_vcs_repo.branches[branch_name]),
                    target_ref=Reference(type='branch', name='', commit_id=None),
                    user_name=self.request.user.name,
                    user_email=self.request.user.email)
            case _:
                raise NotImplementedError('Branch deleting functionality not yet implemented')
        ScmModel().mark_for_invalidation(repo_name)
        self.rhodecode_vcs_repo._invalidate_prop_cache('commit_ids')
        self.rhodecode_vcs_repo._invalidate_prop_cache('_refs')
        self.rhodecode_vcs_repo._invalidate_prop_cache('branches')
        h.flash(_("Branch {} has been successfully {}").format(branch_name, action), category='success')
        return redirect