~launchpad-pqm/launchpad/devel

# Copyright 2009-2011 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Browser views for CodeImports."""

__metaclass__ = type

__all__ = [
    'CodeImportEditView',
    'CodeImportMachineView',
    'CodeImportNewView',
    'CodeImportSetBreadcrumb',
    'CodeImportSetNavigation',
    'CodeImportSetView',
    'CodeImportView',
    ]


from BeautifulSoup import BeautifulSoup
from lazr.restful.interface import (
    copy_field,
    use_template,
    )
from zope.app.form import CustomWidgetFactory
from zope.app.form.interfaces import IInputWidget
from zope.app.form.utility import setUpWidget
from zope.component import getUtility
from zope.formlib import form
from zope.interface import Interface
from zope.schema import Choice

from canonical.launchpad import _
from canonical.launchpad.webapp import (
    canonical_url,
    LaunchpadView,
    Navigation,
    stepto,
    )
from canonical.launchpad.webapp.batching import BatchNavigator
from canonical.launchpad.webapp.breadcrumb import Breadcrumb
from canonical.launchpad.webapp.menu import structured
from lp.app.browser.launchpadform import (
    action,
    custom_widget,
    LaunchpadFormView,
    )
from lp.app.errors import NotFoundError
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.app.widgets.itemswidgets import (
    LaunchpadDropdownWidget,
    LaunchpadRadioWidget,
    )
from lp.app.widgets.textwidgets import (
    StrippedTextWidget,
    URIWidget,
    )
from lp.code.enums import (
    BranchSubscriptionDiffSize,
    BranchSubscriptionNotificationLevel,
    CodeImportReviewStatus,
    CodeReviewNotificationLevel,
    RevisionControlSystems,
    )
from lp.code.errors import BranchExists
from lp.code.interfaces.branch import (
    IBranch,
    user_has_special_branch_access,
    )
from lp.code.interfaces.branchnamespace import (
    get_branch_namespace,
    IBranchNamespacePolicy,
    )
from lp.code.interfaces.branchtarget import IBranchTarget
from lp.code.interfaces.codeimport import (
    ICodeImport,
    ICodeImportSet,
    )
from lp.code.interfaces.codeimportmachine import ICodeImportMachineSet
from lp.registry.interfaces.product import IProduct
from lp.services.fields import URIField
from lp.services.propertycache import cachedproperty


class CodeImportSetNavigation(Navigation):
    """Navigation methods for IBuilder."""
    usedfor = ICodeImportSet

    @stepto('+machines')
    def bugs(self):
        return getUtility(ICodeImportMachineSet)


class CodeImportSetBreadcrumb(Breadcrumb):
    """Builds a breadcrumb for an `ICodeImportSet`."""
    text = u'Code Import System'


class DropdownWidgetWithAny(LaunchpadDropdownWidget):
    """A <select> widget with a more appropriate 'no value' message.

    By default `LaunchpadDropdownWidget` displays 'no value' when the
    associated value is None or not supplied, which is not what we want on
    this page.
    """
    _messageNoValue = _('Any')


class CodeImportSetView(LaunchpadView):
    """The default view for `ICodeImportSet`.

    We present the CodeImportSet as a list of all imports.
    """

    def initialize(self):
        """See `LaunchpadView.initialize`."""
        review_status_field = copy_field(
            ICodeImport['review_status'], required=False, default=None)
        self.review_status_widget = CustomWidgetFactory(DropdownWidgetWithAny)
        setUpWidget(self, 'review_status',  review_status_field, IInputWidget)

        rcs_type_field = copy_field(
            ICodeImport['rcs_type'], required=False, default=None)
        self.rcs_type_widget = CustomWidgetFactory(DropdownWidgetWithAny)
        setUpWidget(self, 'rcs_type',  rcs_type_field, IInputWidget)

        # status should be None if either (a) there were no query arguments
        # supplied, i.e. the user browsed directly to this page (this is when
        # hasValidInput returns False) or (b) the user chose 'Any' in the
        # status widget (this is when hasValidInput returns True but
        # getInputValue returns None).
        review_status = None
        if self.review_status_widget.hasValidInput():
            review_status = self.review_status_widget.getInputValue()
        # Similar for 'type'
        rcs_type = None
        if self.rcs_type_widget.hasValidInput():
            rcs_type = self.rcs_type_widget.getInputValue()

        imports = self.context.search(
            review_status=review_status, rcs_type=rcs_type)

        self.batchnav = BatchNavigator(imports, self.request)


class CodeImportView(LaunchpadView):
    """The default view for `ICodeImport`.

    We present the CodeImport as a simple page listing all the details of the
    import such as target and branch, who requested the import,
    and so on.
    """

    def initialize(self):
        """See `LaunchpadView.initialize`."""
        self.title = "Code Import for %s" % (self.context.branch.target.name,)


class CodeImportBaseView(LaunchpadFormView):
    """A base view for both new and edit code import views."""

    schema = ICodeImport

    custom_widget('cvs_root', StrippedTextWidget, displayWidth=50)
    custom_widget('cvs_module', StrippedTextWidget, displayWidth=20)
    custom_widget('url', URIWidget, displayWidth=50)

    @cachedproperty
    def _super_user(self):
        """Is the user an admin or member of vcs-imports?"""
        celebs = getUtility(ILaunchpadCelebrities)
        return (self.user.inTeam(celebs.admin) or
                self.user.inTeam(celebs.vcs_imports))

    def showOptionalMarker(self, field_name):
        """Don't show the optional marker for rcs locations."""
        # No field in either the new or edit view needs an optional marker,
        # so we can be simple here.
        return False

    def setSecondaryFieldError(self, field, error):
        """Set the field error only if there isn't an error already."""
        if self.getFieldError(field):
            # Leave this one as it is often required or a validator error.
            pass
        else:
            self.setFieldError(field, error)

    def _validateCVS(self, cvs_root, cvs_module, existing_import=None):
        """If the user has specified cvs, then we need to make
        sure that there isn't already an import with those values."""
        if cvs_root is None:
            self.setSecondaryFieldError(
                'cvs_root', 'Enter a CVS root.')
        if cvs_module is None:
            self.setSecondaryFieldError(
                'cvs_module', 'Enter a CVS module.')

        if cvs_root and cvs_module:
            code_import = getUtility(ICodeImportSet).getByCVSDetails(
                cvs_root, cvs_module)
            if (code_import is not None and
                code_import != existing_import):
                self.addError(structured("""
                    Those CVS details are already specified for
                    the imported branch <a href="%s">%s</a>.""",
                    canonical_url(code_import.branch),
                    code_import.branch.unique_name))

    def _validateURL(self, url, existing_import=None, field_name='url'):
        """If the user has specified a url, we need to make sure that there
        isn't already an import with that url."""
        if url is None:
            self.setSecondaryFieldError(
                field_name, 'Enter the URL of a foreign VCS branch.')
        else:
            code_import = getUtility(ICodeImportSet).getByURL(url)
            if (code_import is not None and
                code_import != existing_import):
                self.setFieldError(
                    field_name,
                    structured("""
                    This foreign branch URL is already specified for
                    the imported branch <a href="%s">%s</a>.""",
                    canonical_url(code_import.branch),
                    code_import.branch.unique_name))



class NewCodeImportForm(Interface):
    """The fields presented on the form for editing a code import."""

    use_template(IBranch, ['owner'])
    use_template(
        ICodeImport,
        ['rcs_type', 'cvs_root', 'cvs_module'])

    svn_branch_url = URIField(
        title=_("Branch URL"), required=False,
        description=_(
            "The URL of a Subversion branch, starting with svn:// or "
            "http(s)://.   You can include a username and password as part "
            "of the url, but this will be displayed on the branch page."),
        allowed_schemes=["http", "https", "svn"],
        allow_userinfo=True,
        allow_port=True,
        allow_query=False,
        allow_fragment=False,
        trailing_slash=False)

    git_repo_url = URIField(
        title=_("Repo URL"), required=False,
        description=_(
            "The URL of the git repository.  The HEAD branch will be "
            "imported."),
        allowed_schemes=["git", "http", "https"],
        allow_userinfo=False, # Only anonymous access is supported.
        allow_port=True,
        allow_query=False,
        allow_fragment=False,
        trailing_slash=False)

    hg_repo_url = URIField(
        title=_("Repo URL"), required=False,
        description=_(
            "The URL of the Mercurial repository.  The tip branch will be "
            "imported."),
        allowed_schemes=["http", "https"],
        allow_userinfo=False, # Only anonymous access is supported.
        allow_port=True,
        allow_query=False,    # Query makes no sense in Mercurial
        allow_fragment=False, # Fragment makes no sense in Mercurial
        trailing_slash=False) # See http://launchpad.net/bugs/56357.

    branch_name = copy_field(
        IBranch['name'],
        __name__='branch_name',
        title=_('Branch Name'),
        description=_(
            "This will be used in the branch URL to identify the "
            "imported branch.  Examples: main, trunk."),
        )

    product = Choice(
        title=_('Project'),
        description=_("The Project to associate the code import with."),
        vocabulary="Product",
        )


class CodeImportNewView(CodeImportBaseView):
    """The view to request a new code import."""

    schema = NewCodeImportForm
    for_input = True

    custom_widget('rcs_type', LaunchpadRadioWidget)

    @property
    def initial_values(self):
        return {
            'owner': self.user,
            'rcs_type': RevisionControlSystems.BZR_SVN,
            'branch_name': 'trunk',
            }

    @property
    def context_is_product(self):
        return IProduct.providedBy(self.context)

    @property
    def label(self):
        if self.context_is_product:
            return 'Request a code import for %s' % self.context.displayname
        else:
            return 'Request a code import'

    @property
    def cancel_url(self):
        """Cancel should take the user back to the root site."""
        return '/'

    def setUpFields(self):
        CodeImportBaseView.setUpFields(self)
        if self.context_is_product:
            self.form_fields = self.form_fields.omit('product')

        # If the user can administer branches, then they should be able to
        # assign the ownership of the branch to any valid person or team.
        if user_has_special_branch_access(self.user):
            owner_field = self.schema['owner']
            any_owner_choice = Choice(
                __name__='owner', title=owner_field.title,
                description = _("As an administrator you are able to reassign"
                                " this branch to any person or team."),
                required=True, vocabulary='ValidPersonOrTeam')
            any_owner_field = form.Fields(
                any_owner_choice, render_context=self.render_context)
            # Replace the normal owner field with a more permissive vocab.
            self.form_fields = self.form_fields.omit('owner')
            self.form_fields = any_owner_field + self.form_fields

    def setUpWidgets(self):
        CodeImportBaseView.setUpWidgets(self)

        # Extract the radio buttons from the rcs_type widget, so we can
        # display them separately in the form.
        soup = BeautifulSoup(self.widgets['rcs_type']())
        fields = soup.findAll('input')
        [cvs_button, svn_button, git_button, hg_button, empty_marker] = [
            field for field in fields
            if field.get('value') in ['CVS', 'BZR_SVN', 'GIT', 'HG', '1']]
        cvs_button['onclick'] = 'updateWidgets()'
        svn_button['onclick'] = 'updateWidgets()'
        git_button['onclick'] = 'updateWidgets()'
        hg_button['onclick'] = 'updateWidgets()'
        # The following attributes are used only in the page template.
        self.rcs_type_cvs = str(cvs_button)
        self.rcs_type_svn = str(svn_button)
        self.rcs_type_git = str(git_button)
        self.rcs_type_hg = str(hg_button)
        self.rcs_type_emptymarker = str(empty_marker)

    def _getImportLocation(self, data):
        """Return the import location based on type."""
        rcs_type = data['rcs_type']
        if rcs_type == RevisionControlSystems.CVS:
            return data.get('cvs_root'), data.get('cvs_module'), None
        elif rcs_type == RevisionControlSystems.BZR_SVN:
            return None, None, data.get('svn_branch_url')
        elif rcs_type == RevisionControlSystems.GIT:
            return None, None, data.get('git_repo_url')
        elif rcs_type == RevisionControlSystems.HG:
            return None, None, data.get('hg_repo_url')
        else:
            raise AssertionError(
                'Unexpected revision control type %r.' % rcs_type)

    def _create_import(self, data, status):
        """Create the code import."""
        product = self.getProduct(data)
        cvs_root, cvs_module, url = self._getImportLocation(data)
        return getUtility(ICodeImportSet).new(
            registrant=self.user,
            owner=data['owner'],
            target=IBranchTarget(product),
            branch_name=data['branch_name'],
            rcs_type=data['rcs_type'],
            url=url,
            cvs_root=cvs_root,
            cvs_module=cvs_module,
            review_status=status)

    def _setBranchExists(self, existing_branch):
        """Set a field error indicating that the branch already exists."""
        self.setFieldError(
           'branch_name',
            structured("""
            There is already an existing import for
            <a href="%(product_url)s">%(product_name)s</a>
            with the name of
            <a href="%(branch_url)s">%(branch_name)s</a>.""",
                       product_url=canonical_url(existing_branch.target),
                       product_name=existing_branch.target.name,
                       branch_url=canonical_url(existing_branch),
                       branch_name=existing_branch.name))

    @action(_('Request Import'), name='request_import')
    def request_import_action(self, action, data):
        """Create the code_import, and subscribe the user to the branch."""
        try:
            code_import = self._create_import(data, None)
        except BranchExists, e:
            self._setBranchExists(e.existing_branch)
            return

        # Subscribe the user.
        code_import.branch.subscribe(
            self.user,
            BranchSubscriptionNotificationLevel.FULL,
            BranchSubscriptionDiffSize.NODIFF,
            CodeReviewNotificationLevel.NOEMAIL,
            self.user)

        self.next_url = canonical_url(code_import.branch)

        self.request.response.addNotification("""
            New code import created. The code import operators
            have been notified and the request will be reviewed shortly.""")

    def _showApprove(self, ignored):
        """Is the user an admin or member of vcs-imports?"""
        return self._super_user

    @action(_('Create Approved Import'), name='approve',
            condition=_showApprove)
    def approve_action(self, action, data):
        """Create the code_import, and subscribe the user to the branch."""
        try:
            code_import = self._create_import(
                data, CodeImportReviewStatus.REVIEWED)
        except BranchExists, e:
            self._setBranchExists(e.existing_branch)
            return

        # Don't subscribe the requester as they are an import operator.
        self.next_url = canonical_url(code_import.branch)

        self.request.response.addNotification(
            "New reviewed code import created.")

    def getProduct(self, data):
        """If the context is a product, use that, otherwise get from data."""
        if self.context_is_product:
            return self.context
        else:
            return data.get('product')

    def validate(self, data):
        """See `LaunchpadFormView`."""
        # Make sure that the user is able to create branches for the specified
        # namespace.
        product = self.getProduct(data)
        # 'owner' in data may be None if it failed validation.
        owner = data.get('owner')
        if product is not None and owner is not None:
            namespace = get_branch_namespace(owner, product)
            policy = IBranchNamespacePolicy(namespace)
            if not policy.canCreateBranches(self.user):
                self.setFieldError(
                    'product',
                    "You are not allowed to register imports for %s."
                    % product.displayname)

        rcs_type = data['rcs_type']
        # Make sure fields for unselected revision control systems
        # are blanked out:
        if rcs_type == RevisionControlSystems.CVS:
            self._validateCVS(data.get('cvs_root'), data.get('cvs_module'))
        elif rcs_type == RevisionControlSystems.BZR_SVN:
            self._validateURL(
                data.get('svn_branch_url'), field_name='svn_branch_url')
        elif rcs_type == RevisionControlSystems.GIT:
            self._validateURL(
                data.get('git_repo_url'), field_name='git_repo_url')
        elif rcs_type == RevisionControlSystems.HG:
            self._validateURL(
                data.get('hg_repo_url'), field_name='hg_repo_url')
        else:
            raise AssertionError(
                'Unexpected revision control type %r.' % rcs_type)


class EditCodeImportForm(Interface):
    """The fields presented on the form for editing a code import."""

    url = copy_field(ICodeImport['url'], readonly=False)
    cvs_root = copy_field(ICodeImport['cvs_root'], readonly=False)
    cvs_module = copy_field(ICodeImport['cvs_module'], readonly=False)
    whiteboard = copy_field(IBranch['whiteboard'])


def _makeEditAction(label, status, text):
    """Make an Action to call a particular code import method.

    :param label: The label for the action, which will end up as the
         button title.
    :param status: If the code import has this as its review_status, don't
        show the button (always show the button if it is None).
    :param text: The text to go after 'The code import has been' in a
        notifcation, if a change was made.
    """
    if status is not None:
        def condition(self, ignored):
            return self._showButtonForStatus(status)
    else:
        condition = None
    def success(self, action, data):
        """Make the requested status change."""
        if status is not None:
            data['review_status'] = status
        event = self.code_import.updateFromData(data, self.user)
        if event is not None:
            self.request.response.addNotification(
                'The code import has been ' + text + '.')
        else:
            self.request.response.addNotification('No changes made.')
    name = label.lower().replace(' ', '_')
    return form.Action(
        label, name=name, success=success, condition=condition)


class CodeImportEditView(CodeImportBaseView):
    """View for editing code imports.

    This view is registered against the branch, but mostly edits the code
    import for that branch -- the exception being that it also allows the
    editing of the branch whiteboard.  If the branch has no associated code
    import, then the result is a 404.  If the branch does have a code import,
    then the adapters property allows the form internals to do the associated
    mappings.
    """

    schema = EditCodeImportForm

    # Need this to render the context to prepopulate the form fields.
    # Added here as the base class isn't LaunchpadEditFormView.
    render_context = True
    page_title = 'Edit import details'
    label = page_title

    @property
    def initial_values(self):
        return {'whiteboard': self.context.whiteboard}

    def initialize(self):
        """Show a 404 if the branch has no code import."""
        self.code_import = self.context.code_import
        if self.code_import is None:
            raise NotFoundError
        # The next and cancel location is the branch details page.
        self.cancel_url = self.next_url = canonical_url(self.context)
        CodeImportBaseView.initialize(self)

    @property
    def adapters(self):
        """See `LaunchpadFormView`."""
        return {EditCodeImportForm: self.code_import}

    def setUpFields(self):
        CodeImportBaseView.setUpFields(self)

        # If the import is a Subversion import, then omit the CVS
        # fields, and vice versa.
        if self.code_import.rcs_type == RevisionControlSystems.CVS:
            self.form_fields = self.form_fields.omit('url')
        elif self.code_import.rcs_type in (RevisionControlSystems.SVN,
                                           RevisionControlSystems.BZR_SVN,
                                           RevisionControlSystems.GIT,
                                           RevisionControlSystems.HG):
            self.form_fields = self.form_fields.omit(
                'cvs_root', 'cvs_module')
        else:
            raise AssertionError('Unknown rcs_type for code import.')

    def _showButtonForStatus(self, status):
        """If the status is different, and the user is super, show button."""
        return self._super_user and self.code_import.review_status != status

    actions = form.Actions(
        _makeEditAction(_('Update'), None, 'updated'),
        _makeEditAction(
            _('Approve'), CodeImportReviewStatus.REVIEWED,
            'approved'),
        _makeEditAction(
            _('Mark Invalid'), CodeImportReviewStatus.INVALID,
            'set as invalid'),
        _makeEditAction(
            _('Suspend'), CodeImportReviewStatus.SUSPENDED,
            'suspended'),
        _makeEditAction(
            _('Mark Failing'), CodeImportReviewStatus.FAILING,
            'marked as failing'),
        )

    def validate(self, data):
        """See `LaunchpadFormView`."""
        if self.code_import.rcs_type == RevisionControlSystems.CVS:
            self._validateCVS(
                data.get('cvs_root'), data.get('cvs_module'),
                self.code_import)
        elif self.code_import.rcs_type in (RevisionControlSystems.SVN,
                                           RevisionControlSystems.BZR_SVN,
                                           RevisionControlSystems.GIT,
                                           RevisionControlSystems.HG):
            self._validateURL(data.get('url'), self.code_import)
        else:
            raise AssertionError('Unknown rcs_type for code import.')


class CodeImportMachineView(LaunchpadView):
    """The view for the page that shows all the import machines."""

    label = "Import machines for Launchpad"

    @property
    def machines(self):
        """Get the machines, sorted alphabetically by hostname."""
        return getUtility(ICodeImportMachineSet).getAll()