Produits Entr'ouvert » w.c.s.

from wcs.wf.profile import UpdateUserProfileStatusItem

def test_profile_action():

    wf = Workflow(name='status')

    st1 = wf.add_status('Status1', 'st1')

    item = UpdateUserProfileStatusItem()

    item.id = '_item'

    item.fields = [{'field_id': '__email', 'value': '=form_var_foo'}]

    st1.items.append(item)

    item.parent = st1

    wf2 = assert_import_export_works(wf)

    item2 = wf2.possible_status[0].items[0]

    assert item2.fields == [{'field_id': '__email', 'value': '=form_var_foo'}]

from wcs.wf.profile import UpdateUserProfileStatusItem

def test_profile(pub):

    user = pub.user_class()

    user.store()

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = [

        StringField(id='1', label='Test', type='string', varname='foo'),

    ]

    formdef.store()

    formdata = formdef.data_class()()

    formdata.user_id = user.id

    formdata.data = {'1': 'bar@localhost'}

    item = UpdateUserProfileStatusItem()

    item.fields = [{'field_id': '__email', 'value': '=form_var_foo'}]

    item.perform(formdata)

    assert pub.user_class.get(user.id).email == 'bar@localhost'

    formdata.data = {'1': 'Plop'}

    item.fields = [{'field_id': '__name', 'value': '=form_var_foo'}]

    item.perform(formdata)

    assert pub.user_class.get(user.id).name == 'Plop'

    from wcs.admin.settings import UserFieldsFormDef

    formdef = UserFieldsFormDef(pub)

    formdef.fields = [StringField(id='3', label='test', type='string', varname='plop')]

    formdef.store()

    item.fields = [{'field_id': 'plop', 'value': '=form_var_foo'}]

    item.perform(formdata)

    assert pub.user_class.get(user.id).form_data == {'3': 'Plop'}

def http_put_request(url, body=None, headers={}, timeout=None):

    return _http_request(url, 'PUT', body, headers, timeout=timeout)

# w.c.s. - web application for online forms

# Copyright (C) 2005-2016  Entr'ouvert

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# 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 General Public License

# along with this program; if not, see <http://www.gnu.org/licenses/>.

import json

import urlparse

import xml.etree.ElementTree as ET

from quixote import get_publisher, get_response

from qommon.form import (CompositeWidget, SingleSelectWidget, StringWidget,

        WidgetListAsTable)

from qommon.ident.idp import is_idp_managing_user_attributes

from qommon.misc import http_put_request

from qommon.publisher import get_cfg, get_logger

from wcs.api_utils import sign_url, get_secret_and_orig, MissingSecret

from wcs.workflows import XmlSerialisable, WorkflowStatusItem, register_item_class

def user_ws_url(user_uuid):

    idps = get_cfg('idp', {})

    entity_id = idps.values()[0]['metadata_url']

    base_url = entity_id.split('idp/saml2/metadata')[0]

    url = urlparse.urljoin(base_url, '/api/users/%s/' % user_uuid)

    secret, orig = get_secret_and_orig(url)

    url += '?orig=%s' % orig

    return sign_url(url, secret)

class ProfileUpdateRowWidget(CompositeWidget):

    def __init__(self, name, value=None, **kwargs):

        CompositeWidget.__init__(self, name, value, **kwargs)

        if not value:

            value = {}

        fields = []

        users_cfg = get_cfg('users', {})

        user_formdef = get_publisher().user_class.get_formdef()

        if not user_formdef or not users_cfg.get('field_name'):

            fields.append(('__name', _('Name'), '__name'))

        if not user_formdef or not users_cfg.get('field_email'):

            fields.append(('__email', _('Email'), '__email'))

        if user_formdef and user_formdef.fields:

            for field in user_formdef.fields:

                if field.varname:

                    fields.append((field.varname, field.label, field.varname))

        self.add(SingleSelectWidget, name='field_id', title=_('Field'),

                value=value.get('field_id'),

                options=fields, **kwargs)

        self.add(StringWidget, name='value', title=_('Value'),

                value=value.get('value'))

    def _parse(self, request):

        if self.get('value') and self.get('field_id'):

            self.value = {

                'value': self.get('value'),

                'field_id': self.get('field_id')

            }

        else:

            self.value = None

class ProfileUpdateTableWidget(WidgetListAsTable):

    readonly = False

    def __init__(self, name, **kwargs):

        super(ProfileUpdateTableWidget, self).__init__(name,

                element_type=ProfileUpdateRowWidget, **kwargs)

class FieldNode(XmlSerialisable):

    node_name = 'field'

    def __init__(self, rule={}):

        self.field_id = rule.get('field_id')

        self.value = rule.get('value')

    def as_dict(self):

        return {

            'field_id': self.field_id,

            'value': self.value

        }

    def get_parameters(self):

        return ('field_id', 'value')

class UpdateUserProfileStatusItem(WorkflowStatusItem):

    description = N_('Update User Profile')

    key = 'update_user_profile'

    fields = None

    def get_parameters(self):

        return ('fields',)

    def add_parameters_widgets(self, form, parameters, prefix='', formdef=None):

        if 'fields' in parameters:

            form.add(ProfileUpdateTableWidget, '%sfields' % prefix,

                    title=_('Profile Update'),

                    value=self.fields)

    def fields_export_to_xml(self, item, charset, include_id=False):

        if not self.fields:

            return

        fields_node = ET.SubElement(item, 'fields')

        for field in self.fields:

            fields_node.append(FieldNode(field).export_to_xml(charset=charset,

                include_id=include_id))

        return fields_node

    def fields_init_with_xml(self, elem, charset, include_id=False):

        fields = []

        if elem is None:

            return

        for field_xml_node in elem.findall('field'):

            field_node = FieldNode()

            field_node.init_with_xml(field_xml_node, charset,

                    include_id=include_id)

            fields.append(field_node.as_dict())

        if fields:

            self.fields = fields

    def perform(self, formdata):

        if not self.fields:

            return

        user = formdata.get_user()

        if not user:

            return

        get_publisher().substitutions.feed(formdata)

        new_data = {}

        for field in self.fields:

            new_data[field.get('field_id')] = self.compute(field.get('value'))

        user_formdef = get_publisher().user_class.get_formdef()

        new_user_data = {}

        for field in user_formdef.fields:

            if field.varname in new_data:

                new_user_data[field.id] = new_data.get(field.varname)

        if '__name' in new_data:

            user.name = new_data.get('__name')

        if '__email' in new_data:

            user.email = new_data.get('__email')

        if not user.form_data and new_user_data:

            user.form_data = {}

        if new_user_data:

            user.form_data.update(new_user_data)

        if user.form_data:

            user.set_attributes_from_formdata(user.form_data)

        user.store()

        if user.name_identifiers and is_idp_managing_user_attributes():

            self.perform_idp(user, new_data)

    def perform_idp(self, user, new_data):

        user_uuid = user.name_identifiers[0]

        try:

            url = user_ws_url(user_uuid)

        except MissingSecret:

            get_publisher().notify_of_exception(sys.exc_info(), context='[PROFILE]')

            return

        payload = new_data.copy()

        if '__email' in new_data:

            payload['email'] = new_data.get('__email')

        payload = json.dumps(payload)

        def after_job(job):

            response, status, data, auth_header = http_put_request(url,

                    payload, headers={'Content-type': 'application/json'})

            if status != 200:

                get_logger().error('failed to update profile for user %r', user)

        get_response().add_after_job(str(N_('Updating user profile')), after_job)

register_item_class(UpdateUserProfileStatusItem)

    import wf.profile