Produits Entr'ouvert » w.c.s.

import datetime

import json

import pytest

from quixote import cleanup

from wcs import sessions

from wcs.formdef import FormDef

from wcs.qommon.http_request import HTTPRequest

from wcs.wf.notification import SendNotificationWorkflowStatusItem

from ..utilities import clean_temporary_pub, create_temporary_pub

def setup_module(module):

    cleanup()

def teardown_module(module):

    clean_temporary_pub()

@pytest.fixture

def pub(request):

    pub = create_temporary_pub()

    pub.cfg['language'] = {'language': 'en'}

    pub.write_cfg()

    req = HTTPRequest(None, {'SERVER_NAME': 'example.net', 'SCRIPT_NAME': ''})

    req.response.filter = {}

    req._user = None

    pub._set_request(req)

    req.session = sessions.BasicSession(id=1)

    pub.set_config(req)

    return pub

def test_notifications(pub, http_requests):

    formdef = FormDef()

    formdef.name = 'baz'

    formdef.fields = []

    formdef.store()

    formdata = formdef.data_class()()

    formdata.just_created()

    formdata.store()

    assert not SendNotificationWorkflowStatusItem.is_available()

    if not pub.site_options.has_section('variables'):

        pub.site_options.add_section('variables')

    pub.site_options.set('variables', 'portal_url', 'https://portal/')

    assert SendNotificationWorkflowStatusItem.is_available()

    item = SendNotificationWorkflowStatusItem()

    assert item.to == ['_submitter']

    item.title = 'xxx'

    item.body = 'XXX'

    # no user

    http_requests.empty()

    item.perform(formdata)

    assert http_requests.count() == 0

    # user

    http_requests.empty()

    user = pub.user_class()

    user.name_identifiers = ['xxx']

    user.store()

    formdata.user_id = user.id

    formdata.store()

    item.perform(formdata)

    assert http_requests.count() == 1

    assert http_requests.get_last('url') == 'https://portal/api/notification/add/?NameID=xxx'

    assert json.loads(http_requests.get_last('body')) == {

        'body': 'XXX',

        'url': formdata.get_url(),

        'id': 'formdata:%s' % formdata.get_display_id(),

        'origin': '',

        'summary': 'xxx',

    }

    # deleted user

    http_requests.empty()

    user.deleted_timestamp = datetime.datetime.now()

    user.store()

    item.perform(formdata)

    assert http_requests.count() == 0

    # roles (not exposed in current UI)

    http_requests.empty()

    user.deleted_timestamp = datetime.datetime.now()

    user.store()

    role = pub.role_class(name='blah')

    role.store()

    user1 = pub.user_class()

    user1.roles = [role.id]

    user1.name_identifiers = ['xxy1']

    user1.store()

    user2 = pub.user_class()

    user2.roles = [role.id]

    user2.name_identifiers = ['xxy2']

    user2.store()

    formdef.workflow_roles = {'_receiver': role.id}

    item.to = ['_receiver']

    item.perform(formdata)

    assert http_requests.count() == 2

    assert {x['url'] for x in http_requests.requests} == {

        'https://portal/api/notification/add/?NameID=xxy1',

        'https://portal/api/notification/add/?NameID=xxy2',

    }

    # test inactive users are ignored

    user2.is_active = False

    user2.store()

    http_requests.empty()

    item.perform(formdata)

    assert http_requests.count() == 1

    assert {x['url'] for x in http_requests.requests} == {'https://portal/api/notification/add/?NameID=xxy1'}

    # check notifications are sent to interco portal if it exists

    pub.site_options.set('variables', '_interco_portal_url', 'https://interco-portal/')

    http_requests.empty()

    item.perform(formdata)

    assert http_requests.count() == 1

    assert {x['url'] for x in http_requests.requests} == {

        'https://interco-portal/api/notification/add/?NameID=xxy1'

    }