ForgeSVN/forgesvn/svn_main.py

#       Licensed to the Apache Software Foundation (ASF) under one
#       or more contributor license agreements.  See the NOTICE file
#       distributed with this work for additional information
#       regarding copyright ownership.  The ASF licenses this file
#       to you under the Apache License, Version 2.0 (the
#       "License"); you may not use this file except in compliance
#       with the License.  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#       Unless required by applicable law or agreed to in writing,
#       software distributed under the License is distributed on an
#       "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#       KIND, either express or implied.  See the License for the
#       specific language governing permissions and limitations
#       under the License.

import logging

import six
from tg import tmpl_context as c, request

# Non-stdlib imports
from ming.utils import LazyProperty
from ming.odm.odmsession import ThreadLocalODMSession
from tg import expose, redirect, validate, flash
from tg.decorators import with_trailing_slash, without_trailing_slash
from timermiddleware import Timer

# Pyforge-specific imports
import allura.tasks.repo_tasks
from allura.controllers import BaseController
from allura.controllers.repository import RepoRootController
from allura.controllers.repository import RepoRestController
from allura.lib.decorators import require_post
from allura.lib.repository import RepositoryApp, RepoAdminController
from allura.app import SitemapEntry, ConfigOption, AdminControllerMixin
from allura.lib import helpers as h
from allura.lib import validators as v
from allura import model as M

# Local imports
from . import model as SM
from . import version
from . import widgets
from .controllers import BranchBrowser
from .model.svn import svn_path_exists

log = logging.getLogger(__name__)


class ForgeSVNApp(RepositoryApp):

    '''This is the SVN app for PyForge'''
    __version__ = version.__version__
    config_options = RepositoryApp.config_options + [
        ConfigOption('checkout_url', str, '')
    ]
    permissions_desc = dict(RepositoryApp.permissions_desc, **{
        'write': 'Repo commit access.',
        'admin': 'Set permissions, checkout url, and viewable files. Import a remote repo.',
    })
    tool_label = 'SVN'
    tool_description = """
        Subversion ("svn") is a centralized version control system.  In general, SVN is simpler to use
        but not as powerful as distributed systems like Git and Mercurial.
    """
    ordinal = 4
    forkable = False
    default_branch_name = 'HEAD'

    def __init__(self, project, config):
        super().__init__(project, config)
        self.root = BranchBrowser()
        default_root = RepoRootController()
        self.api_root = RepoRestController()
        self.root.refresh = default_root.refresh
        self.root.commit_browser = default_root.commit_browser
        self.root.commit_browser_data = SVNCommitBrowserController().commit_browser_data
        self.root.status = default_root.status
        self.admin = SVNRepoAdminController(self)

    @LazyProperty
    def repo(self):
        return SM.Repository.query.get(app_config_id=self.config._id)

    def install(self, project):
        '''Create repo object for this tool'''
        super().install(project)
        SM.Repository(
            name=self.config.options.mount_point,
            tool='svn',
            status='initializing',
            fs_path=self.config.options.get('fs_path'))
        ThreadLocalODMSession.flush_all()
        init_from_url = self.config.options.get('init_from_url')
        init_from_path = self.config.options.get('init_from_path')
        if init_from_url or init_from_path:
            allura.tasks.repo_tasks.clone.post(
                cloned_from_path=init_from_path,
                cloned_from_name=None,
                cloned_from_url=init_from_url)
        else:
            allura.tasks.repo_tasks.init.post()

    def admin_menu(self):
        links = super().admin_menu()
        links.insert(1, SitemapEntry(
            'Import Repo',
            c.project.url() + 'admin/' + self.config.options.mount_point + '/' + 'importer/'))
        return links


class SVNRepoAdminController(RepoAdminController):

    def __init__(self, app):
        super().__init__(app)
        self.importer = SVNImportController(self.app)

    @without_trailing_slash
    @expose('jinja:forgesvn:templates/svn/checkout_url.html')
    def checkout_url(self, **kw):
        return dict(app=self.app, allow_config=True)

    @without_trailing_slash
    @expose()
    @require_post()
    @validate({'external_checkout_url': v.NonHttpUrl})
    def set_checkout_url(self, **post_data):
        checkout_url = (post_data.get('checkout_url') or '').strip()
        external_checkout_url = (post_data.get('external_checkout_url') or '').strip()
        if not checkout_url or svn_path_exists("file://%s%s/%s" %
                                               (self.app.repo.fs_path,
                                                self.app.repo.name,
                                                checkout_url)):
            if (self.app.config.options.get('checkout_url') or '') != checkout_url:
                M.AuditLog.log('{}: set "{}" {} => {}'.format(
                    self.app.config.options['mount_point'], "checkout_url",
                    self.app.config.options.get('checkout_url'), checkout_url))
                self.app.config.options.checkout_url = checkout_url
                flash("Checkout URL successfully changed")
        else:
            flash("%s is not a valid path for this repository" %
                  checkout_url, "error")
        if 'external_checkout_url' not in request.validation.errors:
            if (self.app.config.options.get('external_checkout_url') or '') != external_checkout_url:
                M.AuditLog.log('{}: set "{}" {} => {}'.format(
                    self.app.config.options['mount_point'], "external_checkout_url",
                    self.app.config.options.get('external_checkout_url'), external_checkout_url))
                self.app.config.options.external_checkout_url = external_checkout_url
                flash("External checkout URL successfully changed")
        else:
            flash(f"Invalid external checkout URL: {request.validation.errors['external_checkout_url']}", "error")


class SVNImportController(BaseController, AdminControllerMixin):
    import_form = widgets.ImportForm()

    def __init__(self, app):
        self.app = app

    @with_trailing_slash
    @expose('jinja:forgesvn:templates/svn/import.html')
    def index(self, **kw):
        c.is_empty = self.app.repo.is_empty()
        c.form = self.import_form
        return dict()

    @without_trailing_slash
    @expose()
    @require_post()
    @validate(import_form, error_handler=index)
    def do_import(self, checkout_url=None, **kwargs):
        if self.app.repo.is_empty():
            with h.push_context(
                    self.app.config.project_id,
                    app_config_id=self.app.config._id):
                allura.tasks.repo_tasks.reclone.post(
                    cloned_from_path=None,
                    cloned_from_name=None,
                    cloned_from_url=checkout_url)
                M.AuditLog.log('{}: import initiated from "{}"'.format(
                    self.app.config.options['mount_point'], checkout_url))

            M.Notification.post_user(
                c.user, self.app.repo, 'importing',
                text='''Repository import scheduled,
                        an email notification will be sent when complete.''')
        else:
            M.Notification.post_user(
                c.user, self.app.repo, 'error',
                text="Can't import into non empty repository.")
        redirect(six.ensure_text(request.referer or '/'))


class SVNCommitBrowserController(BaseController):

    @without_trailing_slash
    @expose('json:')
    def commit_browser_data(self, start=None, limit=None, **kw):
        data = {
            'commits': [],
            'next_column': 1,
            'max_row': 0,
            'built_tree': {},
            'next_commit': None,
        }
        limit, _ = h.paging_sanitizer(limit or 100, 0, 0)
        for i, commit in enumerate(c.app.repo.log(revs=start, id_only=False, limit=limit+1)):
            if i >= limit:
                data['next_commit'] = str(commit['id'])
                break
            data['commits'].append(str(commit['id']))
            data['built_tree'][commit['id']] = {
                'column': 0,
                'parents': list(map(str, commit['parents'])),
                'short_id': '[r%s]' % commit['id'],
                'message': commit['message'],
                'oid': str(commit['id']),
                'row': i,
                'url': c.app.repo.url_for_commit(commit['id']),
            }
        data['max_row'] = len(data['commits']) - 1
        return data


def svn_timers():
    return Timer(
        'svn_lib.{method_name}', SM.svn.SVNLibWrapper, 'checkout', 'add',
        'checkin', 'info2', 'log', 'cat', 'list')


def forgesvn_timers():
    return Timer('svn_tool.{method_name}', SM.svn.SVNImplementation, '*')