Produits Entr'ouvert » w.c.s.

from wcs.qommon.http_request import HTTPRequest

from wcs.qommon.ident.password_accounts import PasswordAccount

from wcs.wf.jump import JumpWorkflowStatusItem

from wcs.workflows import Workflow

from ..utilities import clean_temporary_pub, create_temporary_pub, get_app

    assert formdef.data_class().get(formdata.id).evolution[-1].who is None

def test_workflow_global_webservice_trigger(pub, local_user):

    workflow = Workflow(name='test')

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

    # anonymous call

    get_app(pub).post(formdata.get_url() + 'hooks/plop/', status=200)

    add_to_journal.comment = 'HELLO WORLD 2'

    workflow.store()

    get_app(pub).post(formdata.get_api_url() + 'hooks/plop/', status=200)

    # call requiring user

    add_to_journal.comment = 'HELLO WORLD 3'

    workflow.store()

    get_app(pub).post(formdata.get_api_url() + 'hooks/plop/', status=403)

    get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/'), status=200)

    # call requiring roles

    add_to_journal.comment = 'HELLO WORLD 4'

    local_user.roles = [role.id]

    local_user.store()

    get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), status=200)

    # call adding data

    add_to_journal.comment = 'HELLO {{plop_test}}'

        sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), {'test': 'foobar'}, status=200

    )

    # (django templating make it turn into HTML)

    # call adding data but with no actions

    ac1.items = []

    # anonymous call

    get_app(pub).post(formdata.get_url() + 'hooks/plop', status=200)

            if item.status:

                self.formdata.jump_status(item.status)

            return json.dumps({'err': 0, 'data': {'id': str(self.formdata.id)}})

        formdata.store()

        formdata.just_created()

        formdata.store()

        formdata.store()

        return json.dumps(

            {

        else:

            formdata.just_created()

            formdata.store()

            formdata.store()

        return json.dumps(

            {

            data_instance.data = item

            data_instance.just_created()

            data_instance.store()

    def done_action_url(self):

        carddef = self.kwargs['carddef_class'].get(self.kwargs['carddef_id'])

        return self.redirect_after_submitted(form, filled)

    def redirect_after_submitted(self, form, filled):

        if url:

            pass  # always redirect to an URL the workflow returned

        elif not self.formdef.is_of_concern_for_user(self.user, filled):

            current_level = current_level - 100

        return levels[current_level]

        url = None

        get_publisher().substitutions.feed(self)

        wf_status = self.get_status()

        from wcs.workflows import perform_items

        return url

    def perform_global_action(self, action_id, user):

        for action in self.formdef.workflow.get_global_actions_for_user(formdata=self, user=user):

            if action.id != action_id:

                continue

            break

    def get_workflow_messages(self, position='top'):

            return None

    def jump_status(self, status_id, user_id=None):

        if status_id == '_previous':

            previous_status = self.pop_previous_marked_status()

            if not previous_status:

            self.status == status

            and self.evolution[-1].status == status

            and not self.evolution[-1].comment

        ):

            # if status do not change and last evolution is empty,

            # just update last jump time on last evolution, do not add one

        url = None

        if existing_formdata is None:

            self.clean_submission_context()

        if not filled.user_id:

            get_session().mark_anonymous_formdata(filled)

                wf_status = item.get_target_status(self.edited_data)

                if wf_status:

                    self.edited_data.jump_status(wf_status[0].id, user_id=user_id)

                else:

                    # add history entry

                    evo = Evolution()

    def remove_self(self):

        assert not self.is_readonly()

        self.remove_object(self.id)

    def get_last_modification_info(self):

        if not get_publisher().snapshot_class:

                new_formdata.store()

                if formdef.enable_tracking_codes:

                    code.formdata = new_formdata  # this will .store() the code

                new_formdata.store()

            # update local object as it may have been modified by new_formdata

JUMP_TIMEOUT_INTERVAL = max((60 // int(os.environ.get('WCS_JUMP_TIMEOUT_CHECKS', '3')), 1))

    action.handle_markers_stack(formdata)

    if workflow_data:

        formdata.update_workflow_data(workflow_data)

        formdata.store()

    formdata.jump_status(action.status)

    return url

                    workflow_data = None

                    if hasattr(get_request(), '_json'):

                        workflow_data = get_request().json

                else:

                    if get_request().is_json():

                        get_response().status_code = 403

                    get_publisher().substitutions.feed(formdef)

                    get_publisher().substitutions.feed(formdata)

                    if jump_action.must_jump(formdata):

                        break

        return -1

    if depth == 0:  # prevents infinite loops

        return

    url = None

    old_status = formdata.status

    for item in items:

        if not item.check_condition(formdata):

            continue

        try:

            url = item.perform(formdata) or url

        except AbortActionException as e:

            break

        if formdata.status != old_status:

            break

    if formdata.status != old_status:

        if not formdata.evolution:

            formdata.evolution = []

        formdata.store()

        # performs the items of the new status

        wf_status = formdata.get_status()

    if url:

        # hack around webtest as it checks type(url) is str and

        # this won't work on django safe strings (isinstance would work);

        )

class DuplicateGlobalActionNameError(Exception):

    pass

                            continue

                        formdata.evolution[-1].add_part(WorkflowGlobalActionTimeoutTriggerMarker(trigger.id))

                        formdata.store()

                        break

        # check for global actions

        for action in filled.formdef.workflow.get_global_actions_for_user(filled, user):

            if 'button-action-%s' % action.id in get_request().form:

                if url:

                    return url

                return

        if evo.status:

            filled.status = evo.status

        filled.store()

        if url:

            return url

        return '<%s %s %r>' % (self.__class__.__name__, self.id, self.name)

class WorkflowStatusItem(XmlSerialisable):

    node_name = 'item'

    description = 'XX'

    def get_add_role_label(self):

        return self.parent.parent.get_add_role_label()

    def perform(self, formdata):

        pass