Produits Entr'ouvert » w.c.s.

from wcs.carddef import CardDef

from wcs.categories import Category

from wcs.formdef import FormDef

from wcs import fields

from wcs.wscalls import NamedWsCall

@pytest.mark.parametrize('notify_on_errors', [True, False])

@pytest.mark.parametrize('record_on_errors', [True, False])

def test_backoffice_wscall_on_error(http_requests, pub, emails, notify_on_errors, record_on_errors):

    pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}

    pub.write_cfg()

    create_user(pub)

    create_environment(pub)

    formdef = FormDef.get_by_urlname('form-title')

    form_class = formdef.data_class()

    else:

        assert emails.count() == 0

    if record_on_errors:

    else:

def test_backoffice_wscall_attachment(http_requests, pub):

def test_backoffice_logged_errors(pub):

    Workflow.wipe()

    workflow = Workflow.get_default_workflow()

    workflow.id = '12'

    formdef.fields = []

    formdef.store()

    create_superuser(pub)

    app = login(get_app(pub))

    resp = app.get('/test/')

    resp = resp.form.submit('submit').follow()

    resp = resp.form.submit('submit')

    app = login(get_app(pub))

    resp = app.get('/backoffice/forms/%s/' % formdef.id)

    assert not 'Acked' in resp.text

    resp = resp.click('Ack').follow()

    assert 'Acked' in resp.text

    resp = resp.click('Delete').follow()

    pub.cfg.update({'debug': {'error_email': None}})

    pub.write_cfg()

    resp = app.get('/test/')

    resp = resp.form.submit('submit').follow()

    resp = resp.form.submit('submit')

    app = login(get_app(pub))

    resp = app.get('/backoffice/workflows/%s/' % workflow.id)

    resp = app.get('/test/')

    resp = resp.form.submit('submit').follow()

    resp = resp.form.submit('submit')

    app = login(get_app(pub))

    resp = app.get('/backoffice/workflows/%s/' % workflow.id)

    source_formdef.store()

    source_formdef.data_class().wipe()

    target_formdef.data_class().wipe()

    return locals()

    # create submitting user

    user = create_formdata['pub'].user_class()

    user.name = 'Jean Darmette'

    assert target_data_class.count() == 0

    # resubmit it through backoffice submission

    resp = resp.form.submit(name='button_resubmit')

    assert target_data_class.count() == 1

    target_formdata = target_data_class.select()[0]

    assert pq('.field-type-file .value').text() == 'bar'

    # create source formdata

    formdata = create_formdata['source_formdef'].data_class()()

    upload = PicklableUpload('/foo/bar', content_type='text/plain')

    formdata.perform_workflow()

    formdata.store()

    assert str(substvars['form_links_resubmitted_form_var_foo_string']) == 'coucou'

    assert 'form_links_resubmitted_form_var_foo_string' in substvars.get_flat_keys()

    assert 'form_links_resubmitted_form_var_foo_string' not in resp

    create_formdata['create_formdata'].map_fields_by_varname = True

    create_formdata['create_formdata'].mappings  = []

    create_formdata['wf'].store()

    assert target_data_class.count() == 0

    # resubmit it through backoffice submission

    resp = resp.form.submit(name='button_resubmit')

    assert target_data_class.count() == 1

    target_formdata = target_data_class.select()[0]

from wcs.carddef import CardDef

from wcs.categories import CardDefCategory

from wcs.formdef import FormDef

from wcs.qommon.http_request import HTTPRequest

from wcs.wf.wscall import WebserviceCallStatusItem

from wcs.workflows import ChoiceWorkflowStatusItem, Workflow

def test_backoffice_cards_wscall_failure_display(http_requests, pub, studio):

    user = create_user(pub)

    Workflow.wipe()

    resp = resp.follow()

    assert 'Error during webservice call' in resp.text

def test_block_card_item_link(pub, studio, blocks_feature):

from wcs.data_sources import NamedDataSource

from wcs.wscalls import NamedWsCall

from wcs import fields

from wcs.forms.root import PublicFormStatusPage

from utilities import get_app, login, create_temporary_pub, clean_temporary_pub

    assert formdef.data_class().get(data_id).status == 'wf-%s' % st2.id

    # jump to a nonexistent status == do not jump, but add a LoggedError

    editable.status = 'deleted_status_id'

    workflow.store()

    # go back to st1

    resp = resp.forms[0].submit('submit')

    resp = resp.follow()

    assert formdef.data_class().get(data_id).status == 'wf-%s' % st1.id  # stay on st1

    # do it again: increment logged_error.occurences_count

    page = login(get_app(pub), username='foo', password='foo').get('/test/%s/' % data_id)

    resp = resp.forms[0].submit('submit')

    resp = resp.follow()

    assert formdef.data_class().get(data_id).status == 'wf-%s' % st1.id  # stay on st1

def test_form_edit_autocomplete_list(pub):

def test_logged_errors(pub):

    Workflow.wipe()

    workflow = Workflow.get_default_workflow()

    workflow.id = '12'

    formdef.fields = []

    formdef.store()

    app = get_app(pub)

    resp = app.get('/test/')

    resp = resp.form.submit('submit').follow()

    resp = resp.form.submit('submit')

    resp = app.get('/test/')

    resp = resp.form.submit('submit').follow()

    resp = resp.form.submit('submit')

    assert error.occurences_count == 2

def test_formdata_named_wscall(http_requests, pub):

    assert pq('.linked .foo_string').text() == 'zob'

    NamedDataSource.wipe()

    data_source = NamedDataSource(name='foobar')

    data_source.data_source = {

        resp = resp.form.submit('submit')  # -> submission

        resp = resp.follow()

        assert create_formdata['target_formdef'].data_class().count() == 0

        resp = resp.form.submit('button_resubmit')

def test_js_libraries(pub):

from quixote import cleanup

from wcs.formdef import FormDef

from wcs.fields import FileField

from wcs.mail_templates import MailTemplate

from wcs.workflows import Workflow, SendmailWorkflowStatusItem

from wcs.qommon.http_request import HTTPRequest

    assert workflow.possible_status[0].items[0].mail_template == 'test-mail-template'

    Workflow.wipe()

    MailTemplate.wipe()

    # check nothing is sent and an error is logged if the mail template is

    # missing

    emails.empty()

    MailTemplate.wipe()

    item.perform(formdata)

    pub.get_request().response.process_after_jobs()

    assert emails.count() == 0

    assert logged_error.summary == 'reference to invalid mail template test-mail-template in status Status1'

        ItemsField, CommentField, EmailField, PageField, TitleField,

        SubtitleField, TextField, BoolField, TableField)

from wcs.formdata import Evolution

from wcs.roles import Role

from wcs.workflows import (Workflow, WorkflowStatusItem,

        SendmailWorkflowStatusItem, SendSMSWorkflowStatusItem,

    assert item.must_jump(formdata) is False

    FormDef.wipe()

    formdef = FormDef()

    formdef.name = 'foobar'

    formdef.store()

    formdef.data_class().wipe()

    formdata = formdef.data_class()()

    item = JumpWorkflowStatusItem()

    item.condition = {'type': 'python', 'value': 'form_var_foobar == 0'}

    assert item.must_jump(formdata) is False

    assert logged_error.summary == 'Failed to evaluate condition'

    assert logged_error.exception_class == 'NameError'

    assert logged_error.exception_message == "name 'form_var_foobar' is not defined"

    assert logged_error.expression == 'form_var_foobar == 0'

    assert logged_error.expression_type == 'python'

    item.condition = {'type': 'python', 'value': '~ invalid ~'}

    assert item.must_jump(formdata) is False

    assert logged_error.summary == 'Failed to evaluate condition'

    assert logged_error.exception_class == 'SyntaxError'

    assert logged_error.exception_message == 'unexpected EOF while parsing (<string>, line 1)'

    assert logged_error.expression_type == 'python'

    FormDef.wipe()

    formdef = FormDef()

    formdef.name = 'foobar'

    formdef.store()

    formdata = formdef.data_class()()

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

    item = JumpWorkflowStatusItem()

    item.condition = {'type': 'django', 'value': '1 < 2'}

    assert item.must_jump(formdata) is True

    item.condition = {'type': 'django', 'value': 'form_var_foo|first|upper == "X"'}

    assert item.must_jump(formdata) is False

    item.condition = {'type': 'django', 'value': '~ invalid ~'}

    assert item.must_jump(formdata) is False

def test_check_auth(pub):

def test_jump_missing_previous_mark(two_pubs):

    workflow = Workflow(name='jump-mark')

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

    formdata.store()

    time.sleep(0.3)

    _apply_timeouts(two_pubs)

def test_sms(pub, sms_mocking):

def test_set_backoffice_field(http_requests, two_pubs):

    Workflow.wipe()

    FormDef.wipe()

    wf = Workflow(name='xxx')

    wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)

    wf.backoffice_fields_formdef.fields = [

    formdata = formdef.data_class().get(formdata.id)

    assert formdata.data['bo1'] == ''

    item.fields = [{'field_id': 'bo1', 'value': '= ~ invalid python ~'}]

    item.perform(formdata)

    formdata = formdef.data_class().get(formdata.id)

    item.fields = [{'field_id': 'bo1', 'value': '{% if bad django %}'}]

    item.perform(formdata)

    formdata = formdef.data_class().get(formdata.id)

def test_set_backoffice_field_file(http_requests, two_pubs):

        item.parent = st1

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

        item.perform(formdata)

        formdata = formdef.data_class().get(formdata.id)

        assert formdata.data['bo1'].base_filename == 'hello.txt'

        assert formdata.data['bo1'].get_content() == b'HELLO WORLD'

    # check wrong field

    item = SetBackofficeFieldsWorkflowStatusItem()

def test_set_backoffice_field_date(two_pubs):

    Workflow.wipe()

    FormDef.wipe()

    wf = Workflow(name='xxx')

    wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)

    st1 = wf.add_status('Status1')

    assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)

    # invalid values => do nothing

    for value in ('plop', '={}', '=[]'):

        item = SetBackofficeFieldsWorkflowStatusItem()

        item.parent = st1

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

        item.perform(formdata)

        formdata = formdef.data_class().get(formdata.id)

        assert datetime.date(*formdata.data['bo1'][:3]) == datetime.date(2017, 3, 23)

    # None : empty date

    item = SetBackofficeFieldsWorkflowStatusItem()

def test_set_backoffice_field_boolean(two_pubs):

    Workflow.wipe()

    FormDef.wipe()

    wf = Workflow(name='xxx')

    wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(wf)

    st1 = wf.add_status('Status1')

def test_workflow_action_condition(two_pubs):

    workflow = Workflow(name='jump condition migration')

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

    workflow.store()

    role = Role(name='bar1')

    role.store()

    user.roles = [role.id]

    user.store()

    choice.condition = {'type': 'python', 'value': 'form_var_foo == "foo"'}

    workflow.store()

        assert FormDef.get(formdef.id).data_class().get(formdata1.id).get_actions_roles() == set([role.id])

        assert FormDef.get(formdef.id).data_class().get(formdata2.id).get_actions_roles() == set()

    assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 1

    assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 2

    # bad condition

    choice.condition = {'type': 'python', 'value': 'foobar == barfoo'}

    workflow.store()

    assert len(FormDef.get(formdef.id).data_class().get_actionable_ids([role.id])) == 0

def test_notifications(pub, http_requests):

    formdef.fields = []

    formdef.workflow_id = workflow.id

    formdef.store()

    for i in range(5):

        formdata = formdef.data_class()()

    assert 'http://example.net/foobar/%s/status (New)' % formdata.id in emails.emails['New arrivals']['payload']

    FormDef.wipe()

    target_formdef = FormDef()

    target_formdef.name = 'target form'

    formdata.just_created()

    assert target_formdef.data_class().count() == 0

    # check unconfigure action do nothing

    formdata.perform_workflow()

    assert target_formdef.data_class().count() == 0

    formdata.perform_workflow()

    assert target_formdef.data_class().count() == 1

    # no tracking code has been created

    created_formdata = target_formdef.data_class().select()[0]

    assert created_formdata.tracking_code is None

    # now we want one

    target_formdef.enable_tracking_codes = True

    target_formdef.store()

    assert target_formdef.data_class().count() == 1

    created_formdata = target_formdef.data_class().select()[0]

    assert created_formdata.tracking_code is not None

    create.condition = {'type': 'python', 'value': '1 == 2'}

    wf.store()

    assert target_formdef.data_class().count() == 0

    CardDef.wipe()

    FormDef.wipe()

    carddef = CardDef()

    carddef.name = 'My card'

    assert carddef.data_class().count() == 1

    formdata = formdef.data_class()()

    today = datetime.date.today()

    formdata.data = {'1': 'item1',

                     '1_display': 'item1',

    formdata.just_created()

    formdata.perform_workflow()

    assert carddef.data_class().count() == 0

    FormDef.wipe()

    CardDef.wipe()

    external_wf = Workflow(name='External Workflow')

    st1 = external_wf.add_status(name='New')

    # remove external formdata

    perform_items([action], formdata)

    assert external_formdef.data_class().count() == 0

    assert external_carddef.data_class().count() == 1

    # formdata is already deleted: cannot find it again

    perform_items([action], formdata)

    # try remove an unexisting carddef: do nothing

    unused_carddef = CardDef()

    unused_carddef.fields = []

    unused_carddef.workflow = external_wf

    unused_carddef.store()

    action.slug = 'carddef:%s' % unused_carddef.url_name

    wf.store()

    perform_items([action], formdata)

    assert external_formdef.data_class().count() == 0

    assert external_carddef.data_class().count() == 1

    # remove the right carddef

    action.slug = 'carddef:%s' % external_carddef.url_name

    wf.store()

    perform_items([action], formdata)

    assert external_formdef.data_class().count() == 0

    assert external_carddef.data_class().count() == 0

    FormDef.wipe()

    CardDef.wipe()

    carddef_wf = Workflow(name='Carddef Workflow')

    carddef_wf.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(carddef_wf)

    formdef.workflow = wf

    formdef.store()

    assert carddef.data_class().count() == 1

    formdata = formdef.data_class()()

    formdata.perform_workflow()

    perform_items([update_action], formdata)

    assert carddef.data_class().count() == 1

    data = carddef.data_class().select()[0]

    assert data.data['bo0'] == '1'

    assert data.data['bo0'] == '2'

    perform_items([delete_action], formdata)

    assert carddef.data_class().count() == 0

    perform_items([delete_action], formdata)

def test_call_external_workflow_with_parent_object(pub):

    FormDef.wipe()

    CardDef.wipe()

    # carddef workflow, with global action to increment a counter in its

    # backoffice fields.

def test_call_external_workflow_use_caller_variable(pub):

    FormDef.wipe()

    CardDef.wipe()

    # carddef workflow, with global action to set a value in a backoffice field

    carddef_wf = Workflow(name='Carddef Workflow')

def test_edit_carddata_with_data_sourced_object(pub):

    FormDef.wipe()

    CardDef.wipe()

    datasource = {

        'type': 'formula',

def test_edit_carddata_with_linked_object(pub):

    FormDef.wipe()

    CardDef.wipe()

    carddef = CardDef()

    carddef.name = 'Parent'

from wcs import fields

from wcs.formdef import FormDef

from wcs.qommon.http_request import HTTPRequest

from wcs.qommon.template import Template

from wcs.wscalls import NamedWsCall

def test_named_wscall(pub):

    # create object

    wscall = NamedWsCall()

    wscall.name = 'Hello'

    wscall.request = {'url': 'http://example.net', 'qs_data': {'a': 'b'}}

@pytest.mark.parametrize('notify_on_errors', [True, False])

@pytest.mark.parametrize('record_on_errors', [True, False])

    pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}

    pub.write_cfg()

    NamedWsCall.wipe()

    FormDef.wipe()

    wscall = NamedWsCall()

        resp = get_app(pub).get('/foobar/')

        assert 'Foo Bar ' in resp.text

        assert emails.count() == 0

    for url_part in ['400', '400-json', '404', '404-json', '500', 'json-err1', 'json-errheader1']:

        status_code = 200

        else:

            assert emails.count() == 0

        if record_on_errors:

        else:

import tempfile

import random

import psycopg2

import shutil

import sys

from wcs import sql, sessions, custom_views

from wcs.qommon import force_str

import wcs

import wcs.wsgi

from wcs.users import User

from wcs.tracking_code import TrackingCode

import wcs.qommon.emails

        pub.session_class = sql.Session

        pub.custom_view_class = sql.CustomView

        pub.snapshot_class = sql.Snapshot

        pub.is_using_postgresql = lambda: True

    else:

        pub.user_class = User

        pub.tracking_code_class = TrackingCode

        pub.session_class = sessions.BasicSession

        pub.custom_view_class = custom_views.CustomView

        pub.is_using_postgresql = lambda: False

    pub.session_manager_class = sessions.StorageSessionManager

        sql.do_session_table()

        sql.do_custom_views_table()

        sql.do_snapshots_table()

        sql.do_meta_table()

        conn.close()

from wcs.qommon import errors, get_cfg

from wcs.qommon.misc import localstrftime

class LoggedErrorDirectory(Directory):

    _q_exports = ['', 'delete', 'ack']

        return redirect('.')

    def delete(self):

        return redirect('..')

    @classmethod

    def get_errors(cls, formdef_id=None, workflow_id=None):

        errors = []

        if formdef_id:

        elif workflow_id:

        return list(errors)

    @classmethod

    def _q_lookup(self, component):

        try:

        except KeyError:

            raise errors.TraversalError()

        get_response().breadcrumb.append(('logged-errors/', _('Logged Errors')))

            if self.log_errors:

                get_logger().warning('failed to evaluate %r (%r)', self, e)

            if self.record_errors:

                summary = _('Failed to evaluate condition')

            raise RuntimeError()

    def evaluate_python(self, local_variables):

        url += param_name + '=' + urllib.quote(param_value)

        def find_item(items, name, value):

            for item in items:

                if str(item.get(name)) == str(value):

                    return item

            # not found

            return None

        request = get_request()

            except TemplateError:

                return ('', explicit_lock)

            except AttributeError as e:

                return ('', explicit_lock)

        elif t == 'user' and user:

        if status_id == '_previous':

            previous_status = self.pop_previous_marked_status()

            if not previous_status:

                summary = _('Failed to compute previous status')

                return

            status_id = previous_status.id

        status = 'wf-%s' % status_id

    def iter_target_datas(self, objectdef=None, object_type=None, status_item=None):

        # objectdef, object_type and status_item are provided when called from a workflow action

        from wcs.wf.create_formdata import LinkedFormdataEvolutionPart

        from .carddef import CardDef

        from .formdef import FormDef

                    yield objectdef.data_class().get(target_id)

                except KeyError as e:

                    # use custom error message depending on target type

                        formdata=self, status_item=status_item, exception=e)

            else:

                # inspect page

import datetime

from .qommon.misc import simplify

from wcs.carddef import CardDef

from wcs.formdef import FormDef

from wcs.workflows import Workflow

    _names = 'logged-errors'

    summary = None

    formdef_class = 'FormDef'

    formdata_id = None

    latest_occurence_timestamp = None

    acked = False

    @classmethod

    def record(cls, error_summary, plain_error_msg=None, formdata=None,

               formdef=None, workflow=None, status=None, status_item=None,

            error.status_id = status.id

        error.first_occurence_timestamp = datetime.datetime.now()

        error.occurences_count += 1

        error.latest_occurence_timestamp = datetime.datetime.now()

        error.store()

    def record_exception(cls, error_summary, plain_error_msg, publisher):

        try:

            context = publisher.substitutions.get_context_variables()

            return

        formdata_id = context.get('form_number_raw')

        formdef_urlname = context.get('form_slug')

            klass = CardDef

        formdef = klass.get_by_urlname(formdef_urlname)

        formdata = formdef.data_class().get(formdata_id, ignore_errors=True)

        tech_id = '%s-%s-' % (self.formdef_id, self.workflow_id)

        if self.status_id:

            tech_id += '%s-' % self.status_id

from .users import User

from .tracking_code import TrackingCode

import pickle

            self.session_class = sql.Session

            self.custom_view_class = sql.CustomView

            self.snapshot_class = sql.Snapshot

            sql.get_connection(new=True)

        else:

            self.user_class = User

            self.session_class = sessions.BasicSession

            self.custom_view_class = custom_views.CustomView

            self.snapshot_class = None

        self.session_manager_class = sessions.StorageSessionManager

        self.set_session_manager(self.session_manager_class(session_class=self.session_class))

        sql.do_tracking_code_table()

        sql.do_custom_views_table()

        sql.do_snapshots_table()

        sql.do_meta_table()

        from .formdef import FormDef

        from .carddef import CardDef

    def log_internal_error(self, error_summary, plain_error_msg, record=False, notify=True):

        tech_id = None

                    error_summary, plain_error_msg, publisher=self)

            if logged_exception:

                tech_id = logged_exception.tech_id

            # this could happen on file descriptor exhaustion

            pass

    def apply_global_action_timeouts(self):

        from wcs.workflows import Workflow, WorkflowGlobalActionTimeoutTrigger

        for workflow in Workflow.select():

import wcs.carddata

import wcs.custom_views

import wcs.formdata

import wcs.snapshots

import wcs.tracking_code

import wcs.users

    cur.close()

@guard_postgres

def do_meta_table(conn=None, cur=None, insert_current_sql_level=True):

    own_conn = False

        return cls.get(row[0])

class classproperty(object):

    def __init__(self, f):

        self.f = f

# latest migration, number + description (description is not used

# programmaticaly but will make sure git conflicts if two migrations are

# separately added with the same number)

def migrate_global_views(conn, cur):

    if sql_level < 42:

        # 42: create snapshots table

        do_snapshots_table()

    cur.execute('''UPDATE wcs_meta SET value = %s WHERE key = %s''', (

        str(SQL_LEVEL[0]), 'sql_level'))

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

import xml.etree.ElementTree as ET

from quixote.html import htmltext

from ..qommon import _, N_

                try:

                    new_value = formdef_field.convert_value_from_anything(new_value)

                except ValueError as e:

                    summary = _('Failed to convert %(class)s value to %(kind)s field (%(id)s)') % {

                                    'class': type(new_value),

                                    'kind': _(getattr(formdef_field, 'description', 'unknown')),

                                    'id': field['field_id'],

                                }

                    expression_dict = self.get_expression(field['value'])

                    continue

            formdata.data['%s' % field['field_id']] = new_value

                             SingleSelectWidget, ComputedExpressionWidget,

                             CheckboxWidget, VarnameWidget, HtmlWidget)

from wcs.workflows import WorkflowStatusItem, register_item_class

from wcs.formdef import FormDef

                        field=dest_field,

                        value=src.data.get(field.id))

                except Exception as e:

        # field.id can be serialized to xml, so we must always convert them to

        # str when matching

                    value=value)

            except Exception as e:

                expression = self.get_expression(mapping.expression)

        if missing_fields:

            summary = _('Missing field %r') % missing_fields

    def _set_value(self, formdata, field, value):

        if field.convert_value_from_anything:

from wcs.qommon import _

from wcs.qommon.form import SingleSelectWidget