integrations: fix re-submit of form on repo integrations.

rhodecode/integrations/views.py

@@ -1,393 +1,391 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2012-2017 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 pylons
 import deform
 import logging
 import colander
 import peppercorn
 import webhelpers.paginate
 
 from pyramid.httpexceptions import HTTPFound, HTTPForbidden, HTTPBadRequest
 from pyramid.renderers import render
 from pyramid.response import Response
 
 from rhodecode.lib import auth
 from rhodecode.lib.auth import LoginRequired, HasPermissionAllDecorator
 from rhodecode.lib.utils2 import safe_int
 from rhodecode.lib.helpers import Page
 from rhodecode.model.db import Repository, RepoGroup, Session, Integration
 from rhodecode.model.scm import ScmModel
 from rhodecode.model.integration import IntegrationModel
 from rhodecode.admin.navigation import navigation_list
 from rhodecode.translation import _
 from rhodecode.integrations import integration_type_registry
 from rhodecode.model.validation_schema.schemas.integration_schema import (
     make_integration_schema, IntegrationScopeType)
 
 log = logging.getLogger(__name__)
 
 
 class IntegrationSettingsViewBase(object):
     """ Base Integration settings view used by both repo / global settings """
 
     def __init__(self, context, request):
         self.context = context
         self.request = request
         self._load_general_context()
 
         if not self.perm_check(request.user):
             raise HTTPForbidden()
 
     def _load_general_context(self):
         """
         This avoids boilerplate for repo/global+list/edit+views/templates
         by doing all possible contexts at the same time however it should
         be split up into separate functions once more "contexts" exist
         """
 
         self.IntegrationType = None
         self.repo = None
         self.repo_group = None
         self.integration = None
         self.integrations = {}
 
         request = self.request
 
         if 'repo_name' in request.matchdict:  # in repo settings context
             repo_name = request.matchdict['repo_name']
             self.repo = Repository.get_by_repo_name(repo_name)
 
         if 'repo_group_name' in request.matchdict: # in group settings context
             repo_group_name = request.matchdict['repo_group_name']
             self.repo_group = RepoGroup.get_by_group_name(repo_group_name)
 
 
         if 'integration' in request.matchdict:  # integration type context
             integration_type = request.matchdict['integration']
             self.IntegrationType = integration_type_registry[integration_type]
 
         if 'integration_id' in request.matchdict:  # single integration context
             integration_id = request.matchdict['integration_id']
             self.integration = Integration.get(integration_id)
 
             # extra perms check just in case
             if not self._has_perms_for_integration(self.integration):
                 raise HTTPForbidden()
 
         self.settings = self.integration and self.integration.settings or {}
         self.admin_view = not (self.repo or self.repo_group)
 
     def _has_perms_for_integration(self, integration):
         perms = self.request.user.permissions
 
         if 'hg.admin' in perms['global']:
             return True
 
         if integration.repo:
             return perms['repositories'].get(
                 integration.repo.repo_name) == 'repository.admin'
 
         if integration.repo_group:
             return perms['repositories_groups'].get(
                 integration.repo_group.group_name) == 'group.admin'
 
         return False
 
     def _template_c_context(self):
         # TODO: dan: this is a stopgap in order to inherit from current pylons
         # based admin/repo settings templates - this should be removed entirely
         # after port to pyramid
 
         c = pylons.tmpl_context
         c.active = 'integrations'
         c.rhodecode_user = self.request.user
         c.repo = self.repo
         c.repo_group = self.repo_group
         c.repo_name = self.repo and self.repo.repo_name or None
         c.repo_group_name = self.repo_group and self.repo_group.group_name or None
 
         if self.repo:
             c.repo_info = self.repo
             c.rhodecode_db_repo = self.repo
             c.repository_pull_requests = ScmModel().get_pull_requests(self.repo)
         else:
             c.navlist = navigation_list(self.request)
 
         return c
 
     def _form_schema(self):
         schema = make_integration_schema(IntegrationType=self.IntegrationType,
                                          settings=self.settings)
 
         # returns a clone, important if mutating the schema later
         return schema.bind(
             permissions=self.request.user.permissions,
             no_scope=not self.admin_view)
 
-
     def _form_defaults(self):
         defaults = {}
 
         if self.integration:
             defaults['settings'] = self.integration.settings or {}
             defaults['options'] = {
                 'name': self.integration.name,
                 'enabled': self.integration.enabled,
                 'scope': {
                     'repo': self.integration.repo,
                     'repo_group': self.integration.repo_group,
                     'child_repos_only': self.integration.child_repos_only,
                 },
             }
         else:
             if self.repo:
                 scope = _('{repo_name} repository').format(
                     repo_name=self.repo.repo_name)
             elif self.repo_group:
                 scope = _('{repo_group_name} repo group').format(
                     repo_group_name=self.repo_group.group_name)
             else:
                 scope = _('Global')
 
             defaults['options'] = {
                 'enabled': True,
                 'name': _('{name} integration').format(
                     name=self.IntegrationType.display_name),
             }
             defaults['options']['scope'] = {
                 'repo': self.repo,
                 'repo_group': self.repo_group,
             }
 
         return defaults
 
     def _delete_integration(self, integration):
         Session().delete(self.integration)
         Session().commit()
         self.request.session.flash(
             _('Integration {integration_name} deleted successfully.').format(
                 integration_name=self.integration.name),
             queue='success')
 
         if self.repo:
             redirect_to = self.request.route_url(
                 'repo_integrations_home', repo_name=self.repo.repo_name)
         elif self.repo_group:
             redirect_to = self.request.route_url(
                 'repo_group_integrations_home',
                 repo_group_name=self.repo_group.group_name)
         else:
             redirect_to = self.request.route_url('global_integrations_home')
         raise HTTPFound(redirect_to)
 
     def settings_get(self, defaults=None, form=None):
         """
         View that displays the integration settings as a form.
         """
 
         defaults = defaults or self._form_defaults()
         schema = self._form_schema()
 
         if self.integration:
             buttons = ('submit', 'delete')
         else:
             buttons = ('submit',)
 
         form = form or deform.Form(schema, appstruct=defaults, buttons=buttons)
 
         template_context = {
             'form': form,
             'current_IntegrationType': self.IntegrationType,
             'integration': self.integration,
             'c': self._template_c_context(),
         }
 
         return template_context
 
     @auth.CSRFRequired()
     def settings_post(self):
         """
         View that validates and stores the integration settings.
         """
         controls = self.request.POST.items()
         pstruct = peppercorn.parse(controls)
 
         if self.integration and pstruct.get('delete'):
             return self._delete_integration(self.integration)
 
         schema = self._form_schema()
 
         skip_settings_validation = False
         if self.integration and 'enabled' not in pstruct.get('options', {}):
             skip_settings_validation = True
             schema['settings'].validator = None
             for field in schema['settings'].children:
                 field.validator = None
                 field.missing = ''
 
         if self.integration:
             buttons = ('submit', 'delete')
         else:
             buttons = ('submit',)
 
         form = deform.Form(schema, buttons=buttons)
 
         if not self.admin_view:
             # scope is read only field in these cases, and has to be added
             options = pstruct.setdefault('options', {})
             if 'scope' not in options:
                 options['scope'] = IntegrationScopeType().serialize(None, {
                     'repo': self.repo,
                     'repo_group': self.repo_group,
                 })
 
         try:
             valid_data = form.validate_pstruct(pstruct)
         except deform.ValidationFailure as e:
             self.request.session.flash(
                 _('Errors exist when saving integration settings. '
                   'Please check the form inputs.'),
                 queue='error')
             return self.settings_get(form=e)
 
         if not self.integration:
             self.integration = Integration()
             self.integration.integration_type = self.IntegrationType.key
             Session().add(self.integration)
 
         scope = valid_data['options']['scope']
 
         IntegrationModel().update_integration(self.integration,
             name=valid_data['options']['name'],
             enabled=valid_data['options']['enabled'],
             settings=valid_data['settings'],
             repo=scope['repo'],
             repo_group=scope['repo_group'],
             child_repos_only=scope['child_repos_only'],
         )
 
-
         self.integration.settings = valid_data['settings']
         Session().commit()
         # Display success message and redirect.
         self.request.session.flash(
             _('Integration {integration_name} updated successfully.').format(
                 integration_name=self.IntegrationType.display_name),
             queue='success')
 
-
         # if integration scope changes, we must redirect to the right place
         # keeping in mind if the original view was for /repo/ or /_admin/
         admin_view = not (self.repo or self.repo_group)
 
         if self.integration.repo and not admin_view:
             redirect_to = self.request.route_path(
                 'repo_integrations_edit',
                 repo_name=self.integration.repo.repo_name,
                 integration=self.integration.integration_type,
                 integration_id=self.integration.integration_id)
         elif self.integration.repo_group and not admin_view:
             redirect_to = self.request.route_path(
                 'repo_group_integrations_edit',
                 repo_group_name=self.integration.repo_group.group_name,
                 integration=self.integration.integration_type,
                 integration_id=self.integration.integration_id)
         else:
             redirect_to = self.request.route_path(
                 'global_integrations_edit',
                 integration=self.integration.integration_type,
                 integration_id=self.integration.integration_id)
 
         return HTTPFound(redirect_to)
 
     def index(self):
         """ List integrations """
         if self.repo:
             scope = self.repo
         elif self.repo_group:
             scope = self.repo_group
         else:
             scope = 'all'
 
         integrations = []
 
         for IntType, integration in IntegrationModel().get_integrations(
                         scope=scope, IntegrationType=self.IntegrationType):
 
             # extra permissions check *just in case*
             if not self._has_perms_for_integration(integration):
                 continue
 
             integrations.append((IntType, integration))
 
         sort_arg = self.request.GET.get('sort', 'name:asc')
         if ':' in sort_arg:
             sort_field, sort_dir = sort_arg.split(':')
         else:
             sort_field = sort_arg, 'asc'
 
         assert sort_field in ('name', 'integration_type', 'enabled', 'scope')
 
         integrations.sort(
-            key=lambda x: getattr(x[1], sort_field), reverse=(sort_dir=='desc'))
-
+            key=lambda x: getattr(x[1], sort_field),
+            reverse=(sort_dir == 'desc'))
 
         page_url = webhelpers.paginate.PageURL(
             self.request.path, self.request.GET)
         page = safe_int(self.request.GET.get('page', 1), 1)
 
         integrations = Page(integrations, page=page, items_per_page=10,
                                url=page_url)
 
         template_context = {
             'sort_field': sort_field,
             'rev_sort_dir': sort_dir != 'desc' and 'desc' or 'asc',
             'current_IntegrationType': self.IntegrationType,
             'integrations_list': integrations,
             'available_integrations': integration_type_registry,
             'c': self._template_c_context(),
             'request': self.request,
         }
         return template_context
 
     def new_integration(self):
         template_context = {
             'available_integrations': integration_type_registry,
             'c': self._template_c_context(),
         }
         return template_context
 
+
 class GlobalIntegrationsView(IntegrationSettingsViewBase):
     def perm_check(self, user):
         return auth.HasPermissionAll('hg.admin').check_permissions(user=user)
 
 
 class RepoIntegrationsView(IntegrationSettingsViewBase):
     def perm_check(self, user):
         return auth.HasRepoPermissionAll('repository.admin'
             )(repo_name=self.repo.repo_name, user=user)
 
 
 class RepoGroupIntegrationsView(IntegrationSettingsViewBase):
     def perm_check(self, user):
         return auth.HasRepoGroupPermissionAll('group.admin'
             )(group_name=self.repo_group.group_name, user=user)
 

rhodecode/model/validation_schema/schemas/integration_schema.py

@@ -1,226 +1,226 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 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 os
 
 import deform
 import colander
 
 from rhodecode.translation import _
 from rhodecode.model.db import Repository, RepoGroup
 from rhodecode.model.validation_schema import validators, preparers
 
 
 def integration_scope_choices(permissions):
     """
     Return list of (value, label) choices for integration scopes depending on
     the permissions
     """
     result = [('', _('Pick a scope:'))]
     if 'hg.admin' in permissions['global']:
         result.extend([
            ('global', _('Global (all repositories)')),
            ('root-repos', _('Top level repositories only')),
         ])
 
     repo_choices = [
         ('repo:%s' % repo_name, '/' + repo_name)
         for repo_name, repo_perm
         in permissions['repositories'].items()
         if repo_perm == 'repository.admin'
     ]
     repogroup_choices = [
         ('repogroup:%s' % repo_group_name, '/' + repo_group_name + '/ (child repos only)')
         for repo_group_name, repo_group_perm
         in permissions['repositories_groups'].items()
         if repo_group_perm == 'group.admin'
     ]
     repogroup_recursive_choices = [
         ('repogroup-recursive:%s' % repo_group_name, '/' + repo_group_name + '/ (recursive)')
         for repo_group_name, repo_group_perm
         in permissions['repositories_groups'].items()
         if repo_group_perm == 'group.admin'
     ]
     result.extend(
         sorted(repogroup_recursive_choices + repogroup_choices + repo_choices,
             key=lambda (choice, label): choice.split(':', 1)[1]
         )
     )
     return result
 
 
 @colander.deferred
 def deferred_integration_scopes_validator(node, kw):
     perms = kw.get('permissions')
     def _scope_validator(_node, scope):
         is_super_admin = 'hg.admin' in perms['global']
 
         if scope.get('repo'):
             if (is_super_admin or perms['repositories'].get(
                 scope['repo'].repo_name) == 'repository.admin'):
                 return True
             msg = _('Only repo admins can create integrations')
             raise colander.Invalid(_node, msg)
         elif scope.get('repo_group'):
             if (is_super_admin or perms['repositories_groups'].get(
                 scope['repo_group'].group_name) == 'group.admin'):
                 return True
 
             msg = _('Only repogroup admins can create integrations')
             raise colander.Invalid(_node, msg)
         else:
             if is_super_admin:
                 return True
             msg = _('Only superadmins can create global integrations')
             raise colander.Invalid(_node, msg)
 
     return _scope_validator
 
 
 @colander.deferred
 def deferred_integration_scopes_widget(node, kw):
     if kw.get('no_scope'):
         return deform.widget.TextInputWidget(readonly=True)
 
     choices = integration_scope_choices(kw.get('permissions'))
     widget = deform.widget.Select2Widget(values=choices)
     return widget
 
 
 class IntegrationScopeType(colander.SchemaType):
     def serialize(self, node, appstruct):
         if appstruct is colander.null:
             return colander.null
 
         if appstruct.get('repo'):
             return 'repo:%s' % appstruct['repo'].repo_name
         elif appstruct.get('repo_group'):
             if appstruct.get('child_repos_only'):
                 return 'repogroup:%s' % appstruct['repo_group'].group_name
             else:
                 return 'repogroup-recursive:%s' % (
                     appstruct['repo_group'].group_name)
         else:
             if appstruct.get('child_repos_only'):
                 return 'root-repos'
             else:
                 return 'global'
 
         raise colander.Invalid(node, '%r is not a valid scope' % appstruct)
 
     def deserialize(self, node, cstruct):
         if cstruct is colander.null:
             return colander.null
 
         if cstruct.startswith('repo:'):
             repo = Repository.get_by_repo_name(cstruct.split(':')[1])
             if repo:
                 return {
                     'repo': repo,
                     'repo_group': None,
-                    'child_repos_only': None,
+                    'child_repos_only': False,
                 }
         elif cstruct.startswith('repogroup-recursive:'):
             repo_group = RepoGroup.get_by_group_name(cstruct.split(':')[1])
             if repo_group:
                 return {
                     'repo': None,
                     'repo_group': repo_group,
                     'child_repos_only': False
                 }
         elif cstruct.startswith('repogroup:'):
             repo_group = RepoGroup.get_by_group_name(cstruct.split(':')[1])
             if repo_group:
                 return {
                     'repo': None,
                     'repo_group': repo_group,
                     'child_repos_only': True
                 }
         elif cstruct == 'global':
             return {
                 'repo': None,
                 'repo_group': None,
                 'child_repos_only': False
             }
         elif cstruct == 'root-repos':
             return {
                 'repo': None,
                 'repo_group': None,
                 'child_repos_only': True
             }
 
         raise colander.Invalid(node, '%r is not a valid scope' % cstruct)
 
 
 class IntegrationOptionsSchemaBase(colander.MappingSchema):
 
     name = colander.SchemaNode(
         colander.String(),
         description=_('Short name for this integration.'),
         missing=colander.required,
         title=_('Integration name'),
     )
 
     scope = colander.SchemaNode(
         IntegrationScopeType(),
         description=_(
             'Scope of the integration. Recursive means the integration '
             ' runs on all repos of that group and children recursively.'),
         title=_('Integration scope'),
         validator=deferred_integration_scopes_validator,
         widget=deferred_integration_scopes_widget,
         missing=colander.required,
     )
 
     enabled = colander.SchemaNode(
         colander.Bool(),
         default=True,
         description=_('Enable or disable this integration.'),
         missing=False,
         title=_('Enabled'),
     )
 
 
 
 def make_integration_schema(IntegrationType, settings=None):
     """
     Return a colander schema for an integration type
 
     :param IntegrationType: the integration type class
     :param settings: existing integration settings dict (optional)
     """
 
     settings = settings or {}
     settings_schema = IntegrationType(settings=settings).settings_schema()
 
     class IntegrationSchema(colander.Schema):
         options = IntegrationOptionsSchemaBase()
 
     schema = IntegrationSchema()
     schema['options'].title = _('General integration options')
 
     settings_schema.name = 'settings'
     settings_schema.title = _('{integration_type} settings').format(
         integration_type=IntegrationType.display_name)
     schema.add(settings_schema)
 
     return schema
 
 

rhodecode/tests/models/schemas/test_integration_schema.py

@@ -1,167 +1,167 b''
 # -*- coding: utf-8 -*-
 
 # Copyright (C) 2016-2017 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 colander
 import pytest
 
 from rhodecode.integrations import integration_type_registry
 from rhodecode.integrations.types.base import IntegrationTypeBase
 from rhodecode.model.validation_schema.schemas.integration_schema import (
     make_integration_schema
 )
 
 
 @pytest.mark.usefixtures('app', 'autologin_user')
 class TestIntegrationSchema(object):
 
     def test_deserialize_integration_schema_perms(
             self, backend_random, test_repo_group, StubIntegrationType):
 
         repo = backend_random.repo
         repo_group = test_repo_group
 
         empty_perms_dict = {
             'global': [],
             'repositories': {},
             'repositories_groups': {},
         }
 
         perms_tests = [
             (
                 'repo:%s' % repo.repo_name,
                 {
-                    'child_repos_only': None,
+                    'child_repos_only': False,
                     'repo_group': None,
                     'repo': repo,
                 },
                 [
                     ({},  False),
                     ({'global': ['hg.admin']},  True),
                     ({'global': []},  False),
                     ({'repositories': {repo.repo_name: 'repository.admin'}}, True),
                     ({'repositories': {repo.repo_name: 'repository.read'}}, False),
                     ({'repositories': {repo.repo_name: 'repository.write'}}, False),
                     ({'repositories': {repo.repo_name: 'repository.none'}}, False),
                 ]
             ),
             (
                 'repogroup:%s' % repo_group.group_name,
                 {
                     'repo': None,
                     'repo_group': repo_group,
                     'child_repos_only': True,
                 },
                 [
                     ({},  False),
                     ({'global': ['hg.admin']},  True),
                     ({'global': []},  False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.admin'}}, True),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.read'}}, False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.write'}}, False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.none'}}, False),
                 ]
             ),
             (
                 'repogroup-recursive:%s' % repo_group.group_name,
                 {
                     'repo': None,
                     'repo_group': repo_group,
                     'child_repos_only': False,
                 },
                 [
                     ({},  False),
                     ({'global': ['hg.admin']},  True),
                     ({'global': []},  False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.admin'}}, True),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.read'}}, False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.write'}}, False),
                     ({'repositories_groups':
                         {repo_group.group_name: 'group.none'}}, False),
                 ]
             ),
             (
                 'global',
                 {
                     'repo': None,
                     'repo_group': None,
                     'child_repos_only': False,
                 }, [
                     ({},  False),
                     ({'global': ['hg.admin']},  True),
                     ({'global': []},  False),
                 ]
             ),
             (
                 'root-repos',
                 {
                     'repo': None,
                     'repo_group': None,
                     'child_repos_only': True,
                 }, [
                     ({},  False),
                     ({'global': ['hg.admin']},  True),
                     ({'global': []},  False),
                 ]
             ),
         ]
 
         for scope_input, scope_output, perms_allowed in perms_tests:
             for perms_update, allowed in perms_allowed:
                 perms = dict(empty_perms_dict, **perms_update)
 
                 schema = make_integration_schema(
                     IntegrationType=StubIntegrationType
                 ).bind(permissions=perms)
 
                 input_data = {
                     'options': {
                         'enabled': 'true',
                         'scope': scope_input,
                         'name': 'test integration',
                     },
                     'settings': {
                         'test_string_field': 'stringy',
                         'test_int_field': '100',
                     }
                 }
 
                 if not allowed:
                     with pytest.raises(colander.Invalid):
                         schema.deserialize(input_data)
                 else:
                     assert schema.deserialize(input_data) == {
                         'options': {
                             'enabled': True,
                             'scope': scope_output,
                             'name': 'test integration',
                         },
                         'settings': {
                             'test_string_field': 'stringy',
                             'test_int_field': 100,
                         }
                     }