495 lines
16 KiB
Python
495 lines
16 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
import os
|
|
|
|
import pytest
|
|
from quixote import get_publisher
|
|
|
|
from wcs import fields
|
|
from wcs.api_access import ApiAccess
|
|
from wcs.formdef import FormDef
|
|
from wcs.qommon.http_request import HTTPRequest
|
|
from wcs.qommon.ident.password_accounts import PasswordAccount
|
|
from wcs.wf.jump import JumpWorkflowStatusItem
|
|
from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem
|
|
from wcs.workflows import Workflow
|
|
|
|
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app
|
|
from .utils import sign_uri
|
|
|
|
|
|
def pytest_generate_tests(metafunc):
|
|
if 'pub' in metafunc.fixturenames:
|
|
metafunc.parametrize('pub', ['pickle', 'sql'], indirect=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def pub(request, emails):
|
|
pub = create_temporary_pub(sql_mode=(request.param == 'sql'))
|
|
|
|
req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
|
|
pub.set_app_dir(req)
|
|
pub.cfg['identification'] = {'methods': ['password']}
|
|
pub.cfg['language'] = {'language': 'en'}
|
|
pub.write_cfg()
|
|
|
|
open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w').write(
|
|
'''\
|
|
[api-secrets]
|
|
coucou = 1234
|
|
'''
|
|
)
|
|
|
|
return pub
|
|
|
|
|
|
def teardown_module(module):
|
|
clean_temporary_pub()
|
|
|
|
|
|
@pytest.fixture
|
|
def local_user():
|
|
get_publisher().user_class.wipe()
|
|
user = get_publisher().user_class()
|
|
user.name = 'Jean Darmette'
|
|
user.email = 'jean.darmette@triffouilis.fr'
|
|
user.name_identifiers = ['0123456789']
|
|
user.store()
|
|
return user
|
|
|
|
|
|
@pytest.fixture
|
|
def admin_user():
|
|
get_publisher().user_class.wipe()
|
|
user = get_publisher().user_class()
|
|
user.name = 'John Doe Admin'
|
|
user.email = 'john.doe@example.com'
|
|
user.name_identifiers = ['0123456789']
|
|
user.is_admin = True
|
|
user.store()
|
|
|
|
account = PasswordAccount(id='admin')
|
|
account.set_password('admin')
|
|
account.user_id = user.id
|
|
account.store()
|
|
|
|
return user
|
|
|
|
|
|
def test_workflow_trigger(pub, local_user):
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
workflow.add_status('Status2', 'st2')
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=200)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].who is None
|
|
|
|
# check with trailing slash
|
|
formdata.store() # reset
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX/'), status=200)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
|
|
pub.role_class.wipe()
|
|
role = pub.role_class(name='xxx')
|
|
role.store()
|
|
|
|
jump.by = [role.id]
|
|
workflow.store()
|
|
|
|
formdata.store() # (will get back to wf-st1)
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=403)
|
|
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX', user=local_user), status=403)
|
|
|
|
local_user.roles = [role.id]
|
|
local_user.store()
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX', user=local_user), status=200)
|
|
|
|
|
|
def test_workflow_trigger_with_data(pub, local_user):
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
workflow.add_status('Status2', 'st2')
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
|
|
get_app(pub).post_json(
|
|
sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=200, params={'test': 'data'}
|
|
)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert formdef.data_class().get(formdata.id).workflow_data == {'test': 'data'}
|
|
|
|
# post with empty dictionary
|
|
formdata.store() # reset
|
|
get_app(pub).post_json(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=200, params={})
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert not formdef.data_class().get(formdata.id).workflow_data
|
|
|
|
# post with empty data
|
|
formdata.store() # reset
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=200)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert not formdef.data_class().get(formdata.id).workflow_data
|
|
|
|
# post with empty data, but declare json content-type
|
|
formdata.store() # reset
|
|
get_app(pub).post(
|
|
sign_uri(formdata.get_url() + 'jump/trigger/XXX'),
|
|
status=200,
|
|
headers={'content-type': 'application/json'},
|
|
)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert not formdef.data_class().get(formdata.id).workflow_data
|
|
|
|
# post with invalid JSON data
|
|
formdata.store() # reset
|
|
get_app(pub).post(
|
|
sign_uri(formdata.get_url() + 'jump/trigger/XXX'),
|
|
status=400,
|
|
headers={'content-type': 'application/json'},
|
|
params='ERROR',
|
|
)
|
|
|
|
|
|
def test_workflow_trigger_with_condition(pub, local_user):
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.condition = {'type': 'django', 'value': 'form_var_foo == "bar"'}
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
workflow.add_status('Status2', 'st2')
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = [fields.StringField(id='0', label='foo', varname='foo')]
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.data = {'0': 'foo'}
|
|
formdata.just_created()
|
|
formdata.store()
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
|
|
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'), status=403)
|
|
assert resp.json == {'err_desc': 'unmet condition', 'err': 1}
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
# check without json
|
|
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX', format=None), status=403)
|
|
assert resp.content_type == 'text/html'
|
|
|
|
formdata.data['0'] = 'bar'
|
|
formdata.store()
|
|
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'))
|
|
assert resp.json == {'err': 0, 'url': None}
|
|
|
|
|
|
def test_workflow_trigger_jump_once(pub, local_user):
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
st2 = workflow.add_status('Status2', 'st2')
|
|
workflow.add_status('Status3', 'st3')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st3'
|
|
st2.items.append(jump)
|
|
jump.parent = st2
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
|
|
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'))
|
|
assert resp.json == {'err': 0, 'url': None}
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
|
|
resp = get_app(pub).post(sign_uri(formdata.get_url() + 'jump/trigger/XXX'))
|
|
assert resp.json == {'err': 0, 'url': None}
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st3'
|
|
|
|
|
|
def test_workflow_trigger_api_access(pub, local_user):
|
|
pub.role_class.wipe()
|
|
role = pub.role_class(name='xxx')
|
|
role.store()
|
|
role2 = pub.role_class(name='xxx2')
|
|
role2.store()
|
|
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
workflow.add_status('Status2', 'st2')
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
|
|
jump.by = [role.id]
|
|
workflow.store()
|
|
|
|
access = ApiAccess()
|
|
access.name = 'test'
|
|
access.access_identifier = 'test'
|
|
access.access_key = '12345'
|
|
access.roles = [role2]
|
|
access.store()
|
|
|
|
get_app(pub).post(
|
|
sign_uri(formdata.get_url() + 'jump/trigger/XXX/', orig='test', key='12345'), status=403
|
|
)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1' # no change
|
|
|
|
access.roles = [role]
|
|
access.store()
|
|
|
|
get_app(pub).post(
|
|
sign_uri(formdata.get_url() + 'jump/trigger/XXX/', orig='test', key='12345'), status=200
|
|
)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].who is None
|
|
|
|
|
|
def test_workflow_trigger_http_auth_access(pub, local_user):
|
|
pub.role_class.wipe()
|
|
role = pub.role_class(name='xxx')
|
|
role.store()
|
|
role2 = pub.role_class(name='xxx2')
|
|
role2.store()
|
|
|
|
workflow = Workflow(name='test')
|
|
st1 = workflow.add_status('Status1', 'st1')
|
|
jump = JumpWorkflowStatusItem()
|
|
jump.trigger = 'XXX'
|
|
jump.status = 'st2'
|
|
st1.items.append(jump)
|
|
jump.parent = st1
|
|
workflow.add_status('Status2', 'st2')
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
|
|
jump.by = [role.id]
|
|
workflow.store()
|
|
|
|
access = ApiAccess()
|
|
access.name = 'test'
|
|
access.access_identifier = 'test'
|
|
access.access_key = '12345'
|
|
access.roles = [role2]
|
|
access.store()
|
|
|
|
app = get_app(pub)
|
|
app.set_authorization(('Basic', ('test', '12345')))
|
|
app.post(formdata.get_url() + 'jump/trigger/XXX/', status=403)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1' # no change
|
|
|
|
access.roles = [role]
|
|
access.store()
|
|
|
|
app.post(formdata.get_url() + 'jump/trigger/XXX/', headers={'accept': 'application/json'}, status=200)
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st2'
|
|
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')
|
|
|
|
ac1 = workflow.add_global_action('Action', 'ac1')
|
|
trigger = ac1.append_trigger('webservice')
|
|
trigger.identifier = 'plop'
|
|
|
|
add_to_journal = RegisterCommenterWorkflowStatusItem()
|
|
add_to_journal.id = '_add_to_journal'
|
|
add_to_journal.comment = 'HELLO WORLD'
|
|
ac1.items.append(add_to_journal)
|
|
add_to_journal.parent = ac1
|
|
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
|
|
# call to undefined hook
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'hooks/XXX/'), status=404)
|
|
get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/XXX/'), status=404)
|
|
|
|
# anonymous call
|
|
get_app(pub).post(formdata.get_url() + 'hooks/plop/', status=200)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD'
|
|
|
|
add_to_journal.comment = 'HELLO WORLD 2'
|
|
workflow.store()
|
|
get_app(pub).post(formdata.get_api_url() + 'hooks/plop/', status=200)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 2'
|
|
|
|
# call requiring user
|
|
add_to_journal.comment = 'HELLO WORLD 3'
|
|
trigger.roles = ['logged-users']
|
|
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)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 3'
|
|
|
|
# call requiring roles
|
|
add_to_journal.comment = 'HELLO WORLD 4'
|
|
trigger.roles = ['logged-users']
|
|
workflow.store()
|
|
pub.role_class.wipe()
|
|
role = pub.role_class(name='xxx')
|
|
role.store()
|
|
trigger.roles = [role.id]
|
|
workflow.store()
|
|
get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/'), status=403)
|
|
get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), status=403)
|
|
|
|
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)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD 4'
|
|
|
|
# call adding data
|
|
add_to_journal.comment = 'HELLO {{plop_test}}'
|
|
workflow.store()
|
|
get_app(pub).post_json(
|
|
sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), {'test': 'foobar'}, status=200
|
|
)
|
|
# (django templating make it turn into HTML)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == '<div>HELLO foobar</div>'
|
|
|
|
# call adding data but with no actions
|
|
ac1.items = []
|
|
workflow.store()
|
|
get_app(pub).post_json(
|
|
sign_uri(formdata.get_api_url() + 'hooks/plop/', user=local_user), {'test': 'BAR'}, status=200
|
|
)
|
|
assert formdef.data_class().get(formdata.id).workflow_data == {'plop': {'test': 'BAR'}}
|
|
|
|
|
|
def test_workflow_global_webservice_trigger_no_trailing_slash(pub, local_user):
|
|
workflow = Workflow(name='test')
|
|
workflow.add_status('Status1', 'st1')
|
|
|
|
ac1 = workflow.add_global_action('Action', 'ac1')
|
|
trigger = ac1.append_trigger('webservice')
|
|
trigger.identifier = 'plop'
|
|
|
|
add_to_journal = RegisterCommenterWorkflowStatusItem()
|
|
add_to_journal.id = '_add_to_journal'
|
|
add_to_journal.comment = 'HELLO WORLD'
|
|
ac1.items.append(add_to_journal)
|
|
add_to_journal.parent = ac1
|
|
|
|
workflow.store()
|
|
|
|
FormDef.wipe()
|
|
formdef = FormDef()
|
|
formdef.name = 'test'
|
|
formdef.fields = []
|
|
formdef.workflow_id = workflow.id
|
|
formdef.store()
|
|
|
|
formdef.data_class().wipe()
|
|
formdata = formdef.data_class()()
|
|
formdata.just_created()
|
|
formdata.store()
|
|
assert formdef.data_class().get(formdata.id).status == 'wf-st1'
|
|
|
|
# call to undefined hook
|
|
get_app(pub).post(sign_uri(formdata.get_url() + 'hooks/XXX'), status=404)
|
|
get_app(pub).post(sign_uri(formdata.get_api_url() + 'hooks/XXX'), status=404)
|
|
|
|
# anonymous call
|
|
get_app(pub).post(formdata.get_url() + 'hooks/plop', status=200)
|
|
assert formdef.data_class().get(formdata.id).evolution[-1].parts[-1].content == 'HELLO WORLD'
|