misc: split some tests
This commit is contained in:
Lauréline Guérin
parent
e8acd48cc0
commit
69e97b1c42
|
|
@ -0,0 +1,350 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import base64
|
||||
import datetime
|
||||
import hashlib
|
||||
import hmac
|
||||
import os
|
||||
import shutil
|
||||
|
||||
import pytest
|
||||
from django.utils.encoding import force_bytes
|
||||
from django.utils.six.moves.urllib import parse as urllib
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app, login
|
||||
|
||||
from wcs.api_utils import get_secret_and_orig, is_url_signed, sign_url
|
||||
from wcs.qommon.errors import AccessForbiddenError
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.ident.password_accounts import PasswordAccount
|
||||
from wcs.roles import Role
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
@pytest.fixture(params=['sql', 'pickle'])
|
||||
def no_request_pub(request):
|
||||
pub = create_temporary_pub(sql_mode=bool(request.param == 'sql'))
|
||||
pub.app_dir = os.path.join(pub.APP_DIR, 'example.net')
|
||||
pub.set_config()
|
||||
|
||||
open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w').write(
|
||||
'''
|
||||
[wscall-secrets]
|
||||
api.example.com = 1234
|
||||
'''
|
||||
)
|
||||
return pub
|
||||
|
||||
|
||||
def test_get_secret_and_orig(no_request_pub):
|
||||
secret, orig = get_secret_and_orig('https://api.example.com/endpoint/')
|
||||
assert secret == '1234'
|
||||
assert orig == 'example.net'
|
||||
|
||||
|
||||
def test_user_page_redirect(pub):
|
||||
output = get_app(pub).get('/user')
|
||||
assert output.headers.get('location') == 'http://example.net/myspace/'
|
||||
|
||||
|
||||
def test_user_page_error(pub):
|
||||
# check we get json as output for errors
|
||||
output = get_app(pub).get('/api/user/', status=403)
|
||||
assert output.json['err_desc'] == 'no user specified'
|
||||
|
||||
|
||||
def test_user_page_error_when_json_and_no_user(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json', status=403)
|
||||
assert output.json['err_desc'] == 'no user specified'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_missing_orig(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json&signature=xxx', status=403)
|
||||
assert output.json['err_desc'] == 'missing/multiple orig field'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_invalid_orig(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json&orig=coin&signature=xxx', status=403)
|
||||
assert output.json['err_desc'] == 'invalid orig'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_missing_algo(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json&orig=coucou&signature=xxx', status=403)
|
||||
assert output.json['err_desc'] == 'missing/multiple algo field'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_invalid_algo(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json&orig=coucou&signature=xxx&algo=coin', status=403)
|
||||
assert output.json['err_desc'] == 'invalid algo'
|
||||
output = get_app(pub).get(
|
||||
'/api/user/?format=json&orig=coucou&signature=xxx&algo=__getattribute__', status=403
|
||||
)
|
||||
assert output.json['err_desc'] == 'invalid algo'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_invalid_signature(pub):
|
||||
output = get_app(pub).get('/api/user/?format=json&orig=coucou&signature=xxx&algo=sha1', status=403)
|
||||
assert output.json['err_desc'] == 'invalid signature'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_missing_timestamp(pub):
|
||||
signature = urllib.quote(
|
||||
base64.b64encode(hmac.new(b'1234', b'format=json&orig=coucou&algo=sha1', hashlib.sha1).digest())
|
||||
)
|
||||
output = get_app(pub).get(
|
||||
'/api/user/?format=json&orig=coucou&algo=sha1&signature=%s' % signature, status=403
|
||||
)
|
||||
assert output.json['err_desc'] == 'missing/multiple timestamp field'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_missing_email(pub):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha1×tamp=' + timestamp
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)
|
||||
assert output.json['err_desc'] == 'no user specified'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_unknown_nameid(pub):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha1&NameID=xxx×tamp=' + timestamp
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)
|
||||
assert output.json['err_desc'] == 'unknown NameID'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_missing_email_valid_endpoint(pub):
|
||||
# check it's ok to sign an URL without specifiying an user if the endpoint
|
||||
# works fine without user.
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha1×tamp=' + timestamp
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/categories?%s&signature=%s' % (query, signature))
|
||||
assert output.json == {'data': []}
|
||||
output = get_app(pub).get('/json?%s&signature=%s' % (query, signature))
|
||||
assert output.json == {'err': 0, 'data': []}
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_unknown_nameid_valid_endpoint(pub):
|
||||
# check the categories and forms endpoints accept an unknown NameID
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&NameID=xxx&orig=coucou&algo=sha1×tamp=' + timestamp
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/categories?%s&signature=%s' % (query, signature))
|
||||
assert output.json == {'data': []}
|
||||
output = get_app(pub).get('/json?%s&signature=%s' % (query, signature))
|
||||
assert output.json == {'err': 0, 'data': []}
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_success_sha1(pub, local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = (
|
||||
'format=json&orig=coucou&algo=sha1&email='
|
||||
+ urllib.quote(local_user.email)
|
||||
+ '×tamp='
|
||||
+ timestamp
|
||||
)
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature))
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_invalid_signature_algo_mismatch(pub, local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = (
|
||||
'format=json&orig=coucou&algo=sha256&email='
|
||||
+ urllib.quote(local_user.email)
|
||||
+ '×tamp='
|
||||
+ timestamp
|
||||
)
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha1).digest()))
|
||||
output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature), status=403)
|
||||
assert output.json['err_desc'] == 'invalid signature'
|
||||
|
||||
|
||||
def test_get_user_from_api_query_string_error_success_sha256(pub, local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = (
|
||||
'format=json&orig=coucou&algo=sha256&email='
|
||||
+ urllib.quote(local_user.email)
|
||||
+ '×tamp='
|
||||
+ timestamp
|
||||
)
|
||||
signature = urllib.quote(base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha256).digest()))
|
||||
output = get_app(pub).get('/api/user/?%s&signature=%s' % (query, signature))
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
|
||||
def test_sign_url(pub, local_user):
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/user/?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url)
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
# try to add something after signed url
|
||||
get_app(pub).get('%s&foo=bar' % url, status=403)
|
||||
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/user/?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'12345',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url, status=403)
|
||||
|
||||
|
||||
def test_get_user(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/user/?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url)
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
assert [x['name'] for x in output.json['user_roles']] == ['Foo bar']
|
||||
assert [x['slug'] for x in output.json['user_roles']] == ['foo-bar']
|
||||
|
||||
|
||||
def test_api_access_from_xml_storable_object(pub, local_user, admin_user):
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/settings/api-access/new')
|
||||
resp.form['name'] = 'Salut API access key'
|
||||
resp.form['access_identifier'] = 'salut'
|
||||
resp.form['access_key'] = '5678'
|
||||
resp = resp.form.submit('submit')
|
||||
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/user/?format=json&orig=UNKNOWN_ACCESS&email=%s'
|
||||
% (urllib.quote(local_user.email)),
|
||||
'5678',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url, status=403)
|
||||
assert output.json['err_desc'] == 'invalid orig'
|
||||
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/user/?format=json&orig=salut&email=%s' % (urllib.quote(local_user.email)),
|
||||
'5678',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url)
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
|
||||
def test_is_url_signed_check_nonce(pub, local_user, freezer):
|
||||
ORIG = 'xxx'
|
||||
KEY = 'xxx'
|
||||
|
||||
pub.site_options.add_section('api-secrets')
|
||||
pub.site_options.set('api-secrets', ORIG, KEY)
|
||||
pub.site_options.write(open(os.path.join(pub.app_dir, 'site-options.cfg'), 'w'))
|
||||
# test clean_nonces do not bark when nonces directory is empty
|
||||
if os.path.exists(os.path.join(pub.app_dir, 'nonces')):
|
||||
shutil.rmtree(os.path.join(pub.app_dir, 'nonces'))
|
||||
pub.clean_nonces(now=0)
|
||||
nonce_dir = os.path.join(pub.app_dir, 'nonces')
|
||||
assert not os.path.exists(nonce_dir) or not os.listdir(nonce_dir)
|
||||
signed_url = sign_url('?format=json&orig=%s&email=%s' % (ORIG, urllib.quote(local_user.email)), KEY)
|
||||
req = HTTPRequest(
|
||||
None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net', 'QUERY_STRING': signed_url[1:]}
|
||||
)
|
||||
req.process_inputs()
|
||||
pub.set_app_dir(req)
|
||||
pub._set_request(req)
|
||||
|
||||
assert is_url_signed()
|
||||
with pytest.raises(AccessForbiddenError) as exc_info:
|
||||
req.signed = False
|
||||
is_url_signed()
|
||||
assert exc_info.value.public_msg == 'nonce already used'
|
||||
# test that clean nonces works
|
||||
pub.clean_nonces()
|
||||
assert os.listdir(nonce_dir)
|
||||
|
||||
# 80 seconds in the future, nothing should be cleaned
|
||||
freezer.move_to(datetime.timedelta(seconds=80))
|
||||
pub.clean_nonces()
|
||||
assert os.listdir(nonce_dir)
|
||||
|
||||
# 90 seconds in the future, nonces should be removed
|
||||
freezer.move_to(datetime.timedelta(seconds=10))
|
||||
pub.clean_nonces()
|
||||
assert not os.listdir(nonce_dir)
|
||||
|
||||
|
||||
def test_get_user_compat_endpoint(pub, local_user):
|
||||
signed_url = sign_url(
|
||||
'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email), '1234'
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
output = get_app(pub).get(url)
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,184 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs import fields
|
||||
from wcs.carddef import CardDef
|
||||
from wcs.categories import CardDefCategory
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.roles import Role
|
||||
|
||||
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
|
||||
|
||||
|
||||
def test_cards(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
CardDefCategory.wipe()
|
||||
category = CardDefCategory()
|
||||
category.name = 'Category A'
|
||||
category.store()
|
||||
|
||||
CardDef.wipe()
|
||||
carddef = CardDef()
|
||||
carddef.name = 'test'
|
||||
carddef.fields = [fields.StringField(id='0', label='foobar', varname='foo')]
|
||||
carddef.workflow_roles = {'_viewer': role.id}
|
||||
carddef.digest_template = 'bla {{ form_var_foo }} xxx'
|
||||
carddef.store()
|
||||
|
||||
carddef.data_class().wipe()
|
||||
formdata = carddef.data_class()()
|
||||
formdata.data = {'0': 'blah'}
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'shared carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'private carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'owner'
|
||||
custom_view.user = local_user
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'datasource carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'datasource'
|
||||
custom_view.store()
|
||||
|
||||
resp = get_app(pub).get('/api/cards/@list', status=403)
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/@list'))
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['slug'] == 'test'
|
||||
assert resp.json['data'][0]['category_slug'] is None
|
||||
assert resp.json['data'][0]['category_name'] is None
|
||||
assert resp.json['data'][0]['custom_views'] == [
|
||||
{'id': 'datasource-carddef-custom-view', 'text': 'datasource carddef custom view'},
|
||||
{'id': 'shared-carddef-custom-view', 'text': 'shared carddef custom view'},
|
||||
]
|
||||
|
||||
carddef.category = category
|
||||
carddef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/@list'))
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['slug'] == 'test'
|
||||
assert resp.json['data'][0]['category_slug'] == 'category-a'
|
||||
assert resp.json['data'][0]['category_name'] == 'Category A'
|
||||
|
||||
# signed but anonymous
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/test/list?NameID='), status=403)
|
||||
|
||||
# signed without specifying any user -> get everything
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/test/list'))
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/test/list?NameID=%s' % local_user.name_identifiers[0]))
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['display_id'] == formdata.get_display_id()
|
||||
assert resp.json['data'][0]['display_name'] == formdata.get_display_name()
|
||||
assert resp.json['data'][0]['digest'] == formdata.digest
|
||||
assert resp.json['data'][0]['text'] == formdata.digest
|
||||
resp = get_app(pub).get(
|
||||
sign_uri('/api/cards/test/list?NameID=%s&full=on' % local_user.name_identifiers[0])
|
||||
)
|
||||
assert resp.json['data'][0]['fields']['foo'] == 'blah'
|
||||
assert resp.json['data'][0]['digest'] == formdata.digest
|
||||
assert resp.json['data'][0]['text'] == formdata.digest
|
||||
|
||||
# get schema
|
||||
resp = get_app(pub).get(sign_uri('/api/cards/test/@schema'), status=200)
|
||||
assert len(resp.json['fields']) == 1
|
||||
assert resp.json['fields'][0]['label'] == 'foobar'
|
||||
assert resp.json['fields'][0]['varname'] == 'foo'
|
||||
|
||||
|
||||
def test_cards_import_csv(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
CardDef.wipe()
|
||||
carddef = CardDef()
|
||||
carddef.name = 'test'
|
||||
carddef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foo'),
|
||||
fields.StringField(id='1', label='foobar2', varname='foo2'),
|
||||
]
|
||||
carddef.workflow_roles = {'_viewer': role.id}
|
||||
carddef.backoffice_submission_roles = [role.id]
|
||||
carddef.digest_template = 'bla {{ form_var_foo }} xxx'
|
||||
carddef.store()
|
||||
|
||||
carddef.data_class().wipe()
|
||||
|
||||
get_app(pub).get(sign_uri('/api/cards/test/import-csv'), status=405)
|
||||
get_app(pub).put(sign_uri('/api/cards/test/import-csv'), status=403)
|
||||
get_app(pub).put(
|
||||
sign_uri('/api/cards/test/import-csv', user=local_user),
|
||||
params=b'foobar;foobar2\nfirst entry;plop\nsecond entry;plop\n',
|
||||
headers={'content-type': 'text/csv'},
|
||||
)
|
||||
assert carddef.data_class().count() == 2
|
||||
assert set([x.data['0'] for x in carddef.data_class().select()]) == {'first entry', 'second entry'}
|
||||
|
|
@ -0,0 +1,245 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs.categories import Category
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.roles import Role
|
||||
|
||||
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
|
||||
|
||||
|
||||
def test_categories(pub):
|
||||
FormDef.wipe()
|
||||
Category.wipe()
|
||||
category = Category()
|
||||
category.name = 'Category'
|
||||
category.description = 'hello world'
|
||||
category.store()
|
||||
|
||||
resp = get_app(pub).get('/api/categories/', headers={'Accept': 'application/json'})
|
||||
assert resp.json['data'] == [] # no advertised forms
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.category_id = category.id
|
||||
formdef.fields = []
|
||||
formdef.keywords = 'mobile, test'
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test 2'
|
||||
formdef.category_id = category.id
|
||||
formdef.fields = []
|
||||
formdef.keywords = 'foobar'
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
resp = get_app(pub).get('/api/categories/')
|
||||
resp2 = get_app(pub).get('/categories', headers={'Accept': 'application/json'})
|
||||
assert resp.json == resp2.json
|
||||
assert resp.json['data'][0]['title'] == 'Category'
|
||||
assert resp.json['data'][0]['url'] == 'http://example.net/category/'
|
||||
assert resp.json['data'][0]['description'] == '<p>hello world</p>'
|
||||
assert set(resp.json['data'][0]['keywords']) == set(['foobar', 'mobile', 'test'])
|
||||
assert 'forms' not in resp.json['data'][0]
|
||||
|
||||
# check HTML description
|
||||
category.description = '<p><strong>hello world</strong></p>'
|
||||
category.store()
|
||||
resp = get_app(pub).get('/api/categories/')
|
||||
assert resp.json['data'][0]['description'] == category.description
|
||||
|
||||
|
||||
def test_categories_private(pub, local_user):
|
||||
FormDef.wipe()
|
||||
Category.wipe()
|
||||
category = Category()
|
||||
category.name = 'Category'
|
||||
category.description = 'hello world'
|
||||
category.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.category_id = category.id
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
# open form
|
||||
resp = get_app(pub).get('/api/categories/')
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
# private form, the category doesn't appear anymore
|
||||
formdef.roles = ['plop']
|
||||
formdef.store()
|
||||
resp = get_app(pub).get('/api/categories/')
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# not even for a signed request specifying an user
|
||||
resp = get_app(pub).get(sign_uri('http://example.net/api/categories/', local_user))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# but it appears if this is a signed request without user
|
||||
resp = get_app(pub).get(sign_uri('http://example.net/api/categories/'))
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
# or signed with an authorised user
|
||||
local_user.roles = ['plop']
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(sign_uri('http://example.net/api/categories/', local_user))
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
|
||||
def test_categories_formdefs(pub, local_user):
|
||||
FormDef.wipe()
|
||||
Category.wipe()
|
||||
category = Category()
|
||||
category.name = 'Category'
|
||||
category.description = 'hello world'
|
||||
category.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.category_id = category.id
|
||||
formdef.fields = []
|
||||
formdef.keywords = 'mobile, test'
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test 2'
|
||||
formdef.category_id = category.id
|
||||
formdef.fields = []
|
||||
formdef.keywords = 'foobar'
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
formdef2 = FormDef()
|
||||
formdef2.name = 'other test'
|
||||
formdef2.category_id = None
|
||||
formdef2.fields = []
|
||||
formdef2.store()
|
||||
formdef2.data_class().wipe()
|
||||
|
||||
formdef2 = FormDef()
|
||||
formdef2.name = 'test disabled'
|
||||
formdef2.category_id = category.id
|
||||
formdef2.fields = []
|
||||
formdef2.disabled = True
|
||||
formdef2.store()
|
||||
formdef2.data_class().wipe()
|
||||
|
||||
resp = get_app(pub).get('/api/categories/category/formdefs/', status=403)
|
||||
resp2 = get_app(pub).get('/category/json', status=403)
|
||||
resp = get_app(pub).get(sign_uri('/api/categories/category/formdefs/'))
|
||||
resp2 = get_app(pub).get(sign_uri('/category/json'))
|
||||
assert resp.json == resp2.json
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 2
|
||||
assert resp.json['data'][0]['title'] == 'test'
|
||||
assert resp.json['data'][0]['url'] == 'http://example.net/test/'
|
||||
assert resp.json['data'][0]['redirection'] is False
|
||||
assert resp.json['data'][0]['category'] == 'Category'
|
||||
assert resp.json['data'][0]['category_slug'] == 'category'
|
||||
assert 'count' not in resp.json['data'][0]
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/categories/category/formdefs/?include-count=on'))
|
||||
assert resp.json['data'][0]['title'] == 'test'
|
||||
assert resp.json['data'][0]['url'] == 'http://example.net/test/'
|
||||
assert resp.json['data'][0]['count'] == 0
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/categories/category/formdefs/?include-disabled=on'))
|
||||
assert len(resp.json['data']) == 3
|
||||
assert resp.json['data'][2]['title'] == 'test disabled'
|
||||
|
||||
get_app(pub).get('/api/categories/XXX/formdefs/', status=404)
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/categories/category/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = []
|
||||
local_user.store()
|
||||
# check it's not advertised ...
|
||||
formdef.backoffice_submission_roles = [role.id]
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/categories/category/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
resp = get_app(pub).get(
|
||||
sign_uri(
|
||||
'/api/categories/category/formdefs/?backoffice-submission=on&NameID=%s'
|
||||
% local_user.name_identifiers[0]
|
||||
)
|
||||
)
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
# ... unless user has correct roles
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(
|
||||
sign_uri(
|
||||
'/api/categories/category/formdefs/?backoffice-submission=on&NameID=%s'
|
||||
% local_user.name_identifiers[0]
|
||||
)
|
||||
)
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
|
||||
def test_categories_full(pub):
|
||||
test_categories(pub)
|
||||
resp = get_app(pub).get('/api/categories/?full=on')
|
||||
assert len(resp.json['data'][0]['forms']) == 2
|
||||
assert resp.json['data'][0]['forms'][0]['title'] == 'test'
|
||||
assert resp.json['data'][0]['forms'][1]['title'] == 'test 2'
|
||||
|
|
@ -0,0 +1,315 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import os
|
||||
import xml.etree.ElementTree as ET
|
||||
import zipfile
|
||||
|
||||
import pytest
|
||||
from django.utils.six import BytesIO
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs import fields
|
||||
from wcs.carddef import CardDef
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.qommon import ods
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.roles import Role
|
||||
|
||||
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 test_api_custom_view_access(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.workflow_roles = {'_receiver': role.id}
|
||||
formdef.fields = [fields.StringField(id='0', label='foobar', varname='foobar')]
|
||||
formdef.geolocations = {'base': 'Location'}
|
||||
formdef.store()
|
||||
|
||||
carddef = CardDef()
|
||||
carddef.name = 'test'
|
||||
carddef.fields = [fields.StringField(id='0', label='foobar', varname='foo')]
|
||||
carddef.workflow_roles = {'_viewer': role.id}
|
||||
carddef.digest_template = 'bla {{ form_var_foo }} xxx'
|
||||
carddef.geolocations = {'base': 'Location'}
|
||||
carddef.store()
|
||||
|
||||
pub.custom_view_class.wipe()
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'shared formdef custom view'
|
||||
custom_view.formdef = formdef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'private formdef custom view'
|
||||
custom_view.formdef = formdef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'owner'
|
||||
custom_view.user = local_user
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'shared carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'private carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'owner'
|
||||
custom_view.user = local_user
|
||||
custom_view.store()
|
||||
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'datasource carddef custom view'
|
||||
custom_view.formdef = carddef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {}
|
||||
custom_view.visibility = 'datasource'
|
||||
custom_view.store()
|
||||
|
||||
get_app(pub).get(sign_uri('/api/forms/test/list/shared-formdef-custom-view', user=local_user), status=200)
|
||||
get_app(pub).get(sign_uri('/api/forms/test/ods/shared-formdef-custom-view', user=local_user), status=200)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/forms/test/geojson/shared-formdef-custom-view', user=local_user), status=200
|
||||
)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/forms/test/list/private-formdef-custom-view', user=local_user), status=404
|
||||
)
|
||||
get_app(pub).get(sign_uri('/api/forms/test/ods/private-formdef-custom-view', user=local_user), status=404)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/forms/test/geojson/private-formdef-custom-view', user=local_user), status=404
|
||||
)
|
||||
|
||||
get_app(pub).get(sign_uri('/api/cards/test/list/shared-carddef-custom-view', user=local_user), status=200)
|
||||
get_app(pub).get(sign_uri('/api/cards/test/ods/shared-carddef-custom-view', user=local_user), status=200)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/geojson/shared-carddef-custom-view', user=local_user), status=200
|
||||
)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/list/private-carddef-custom-view', user=local_user), status=404
|
||||
)
|
||||
get_app(pub).get(sign_uri('/api/cards/test/ods/private-carddef-custom-view', user=local_user), status=404)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/geojson/private-carddef-custom-view', user=local_user), status=404
|
||||
)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/list/datasource-carddef-custom-view', user=local_user), status=200
|
||||
)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/ods/datasource-carddef-custom-view', user=local_user), status=200
|
||||
)
|
||||
get_app(pub).get(
|
||||
sign_uri('/api/cards/test/geojson/datasource-carddef-custom-view', user=local_user), status=200
|
||||
)
|
||||
|
||||
|
||||
def test_api_list_formdata_custom_view(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.workflow_roles = {'_receiver': role.id}
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
]
|
||||
formdef.store()
|
||||
|
||||
data_class = formdef.data_class()
|
||||
data_class.wipe()
|
||||
|
||||
for i in range(30):
|
||||
formdata = data_class()
|
||||
formdata.data = {'0': 'FOO BAR %d' % i}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
if i % 3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
formdata.store()
|
||||
|
||||
# add proper role to user
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
# check it now gets the data
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list', user=local_user))
|
||||
assert len(resp.json) == 30
|
||||
|
||||
pub.custom_view_class.wipe()
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'custom view'
|
||||
custom_view.formdef = formdef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {"filter": "done", "filter-status": "on"}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list/custom-view', user=local_user))
|
||||
assert len(resp.json['data']) == 20
|
||||
|
||||
|
||||
def test_api_ods_formdata_custom_view(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.workflow_roles = {'_receiver': role.id}
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
]
|
||||
formdef.store()
|
||||
|
||||
data_class = formdef.data_class()
|
||||
data_class.wipe()
|
||||
|
||||
for i in range(30):
|
||||
formdata = data_class()
|
||||
formdata.data = {'0': 'FOO BAR %d' % i}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
if i % 3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
formdata.store()
|
||||
|
||||
# add proper role to user
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
# check it now gets the data
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/ods', user=local_user))
|
||||
zipf = zipfile.ZipFile(BytesIO(resp.body))
|
||||
ods_sheet = ET.parse(zipf.open('content.xml'))
|
||||
assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 11
|
||||
|
||||
pub.custom_view_class.wipe()
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'custom view'
|
||||
custom_view.formdef = formdef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {"filter": "done", "filter-status": "on"}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/ods/custom-view', user=local_user))
|
||||
zipf = zipfile.ZipFile(BytesIO(resp.body))
|
||||
ods_sheet = ET.parse(zipf.open('content.xml'))
|
||||
assert len(ods_sheet.findall('.//{%s}table-row' % ods.NS['table'])) == 21
|
||||
|
||||
|
||||
def test_api_geojson_formdata_custom_view(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.workflow_roles = {'_receiver': role.id}
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
]
|
||||
formdef.geolocations = {'base': 'Location'}
|
||||
formdef.store()
|
||||
|
||||
data_class = formdef.data_class()
|
||||
data_class.wipe()
|
||||
|
||||
for i in range(30):
|
||||
formdata = data_class()
|
||||
formdata.data = {'0': 'FOO BAR %d' % i}
|
||||
formdata.geolocations = {'base': {'lat': 48, 'lon': 2}}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
if i % 3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
formdata.store()
|
||||
|
||||
# add proper role to user
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
# check it now gets the data
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/geojson', user=local_user))
|
||||
assert len(resp.json['features']) == 10
|
||||
|
||||
pub.custom_view_class.wipe()
|
||||
custom_view = pub.custom_view_class()
|
||||
custom_view.title = 'custom view'
|
||||
custom_view.formdef = formdef
|
||||
custom_view.columns = {'list': [{'id': '0'}]}
|
||||
custom_view.filters = {"filter": "done", "filter-status": "on"}
|
||||
custom_view.visibility = 'any'
|
||||
custom_view.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/geojson/custom-view', user=local_user))
|
||||
assert len(resp.json['features']) == 20
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -0,0 +1,853 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import base64
|
||||
import datetime
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
|
||||
import mock
|
||||
import pytest
|
||||
from django.utils.encoding import force_text
|
||||
from django.utils.six import StringIO
|
||||
from django.utils.six.moves.urllib import parse as urllib
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs import fields, qommon
|
||||
from wcs.api_utils import sign_url
|
||||
from wcs.data_sources import NamedDataSource
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.form import PicklableUpload
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.roles import Role
|
||||
from wcs.wf.jump import JumpWorkflowStatusItem
|
||||
from wcs.workflows import Workflow, WorkflowBackofficeFieldsFormDef
|
||||
|
||||
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
|
||||
|
||||
|
||||
def test_formdef_list(pub):
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.id = '14'
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.description = 'plop'
|
||||
formdef.keywords = 'mobile, test'
|
||||
formdef.workflow_roles = {'_receiver': str(role.id)}
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
# anonymous access -> 403
|
||||
resp1 = get_app(pub).get('/json', status=403)
|
||||
resp2 = get_app(pub).get('/', headers={'Accept': 'application/json'}, status=403)
|
||||
resp3 = get_app(pub).get('/api/formdefs/', status=403)
|
||||
|
||||
# signed request
|
||||
resp1 = get_app(pub).get(sign_uri('/json'))
|
||||
resp2 = get_app(pub).get(sign_uri('/'), headers={'Accept': 'application/json'})
|
||||
resp3 = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
assert resp1.json == resp2.json == resp3.json
|
||||
assert resp1.json['data'][0]['title'] == 'test'
|
||||
assert resp1.json['data'][0]['url'] == 'http://example.net/test/'
|
||||
assert resp1.json['data'][0]['redirection'] is False
|
||||
assert resp1.json['data'][0]['always_advertise'] is False
|
||||
assert resp1.json['data'][0]['description'] == 'plop'
|
||||
assert resp1.json['data'][0]['keywords'] == ['mobile', 'test']
|
||||
assert list(resp1.json['data'][0]['functions'].keys()) == ['_receiver']
|
||||
assert resp1.json['data'][0]['functions']['_receiver']['label'] == 'Recipient'
|
||||
assert resp1.json['data'][0]['functions']['_receiver']['role']['slug'] == role.slug
|
||||
assert resp1.json['data'][0]['functions']['_receiver']['role']['name'] == role.name
|
||||
assert 'count' not in resp1.json['data'][0]
|
||||
|
||||
# backoffice_submission formdef : none
|
||||
resp1 = get_app(pub).get('/api/formdefs/?backoffice-submission=on', status=403)
|
||||
resp1 = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp1.json['err'] == 0
|
||||
assert len(resp1.json['data']) == 0
|
||||
|
||||
formdef.data_class().wipe()
|
||||
|
||||
# a draft
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {}
|
||||
formdata.just_created()
|
||||
formdata.status = 'draft'
|
||||
formdata.store()
|
||||
|
||||
other_formdef = FormDef()
|
||||
other_formdef.name = 'test 2'
|
||||
other_formdef.fields = []
|
||||
other_formdef.store()
|
||||
other_formdata = other_formdef.data_class()()
|
||||
other_formdata.data = {}
|
||||
other_formdata.just_created()
|
||||
other_formdata.store()
|
||||
|
||||
# formdata created:
|
||||
# - 1 day ago (=3*4)
|
||||
# - 7 days ago (=2*2)
|
||||
# - 29 days ago (=1*1)
|
||||
# - 31 days ago (=0)
|
||||
for days in [1, 1, 1, 7, 7, 29, 31]:
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {}
|
||||
formdata.just_created()
|
||||
formdata.receipt_time = (datetime.datetime.now() - datetime.timedelta(days=days)).timetuple()
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?include-count=on'))
|
||||
if not pub.is_using_postgresql():
|
||||
assert resp.json['data'][0]['count'] == 8
|
||||
else:
|
||||
# 3*4 + 2*2 + 1*1
|
||||
assert resp.json['data'][0]['count'] == 17
|
||||
|
||||
|
||||
def test_limited_formdef_list(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.id = '14'
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.description = 'plop'
|
||||
formdef.workflow_roles = {'_receiver': str(role.id)}
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['authentication_required'] is False
|
||||
# not present in backoffice-submission formdefs
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# check it's not advertised
|
||||
formdef.roles = [role.id]
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
resp2 = get_app(pub).get(sign_uri('/api/formdefs/?NameID='))
|
||||
resp3 = get_app(pub).get(sign_uri('/api/formdefs/?NameID=XXX'))
|
||||
resp4 = get_app(pub).get(sign_uri('/api/formdefs/?NameID=%s' % local_user.name_identifiers[0]))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1 # advertised in naked calls (as done from combo)
|
||||
assert len(resp2.json['data']) == 0 # not advertised otherwise
|
||||
assert resp2.json == resp3.json == resp4.json
|
||||
# still not present in backoffice-submission formdefs
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# unless user has correct roles
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?NameID=%s' % local_user.name_identifiers[0]))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
local_user.roles = []
|
||||
local_user.store()
|
||||
|
||||
# check it's also included in anonymous/signed calls, but marked for
|
||||
# authentication
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
assert resp.json['data'][0]
|
||||
assert resp.json['data'][0]['authentication_required'] is True
|
||||
|
||||
# check it's advertised
|
||||
formdef.always_advertise = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
resp2 = get_app(pub).get(sign_uri('/api/formdefs/?NameID='))
|
||||
resp3 = get_app(pub).get(sign_uri('/api/formdefs/?NameID=XXX'))
|
||||
resp4 = get_app(pub).get(sign_uri('/api/formdefs/?NameID=%s' % local_user.name_identifiers[0]))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['authentication_required']
|
||||
assert resp.json == resp2.json == resp3.json == resp4.json
|
||||
|
||||
formdef.required_authentication_contexts = ['fedict']
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/'))
|
||||
assert resp.json['data'][0]['required_authentication_contexts'] == ['fedict']
|
||||
|
||||
|
||||
def test_formdef_list_redirection(pub):
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.disabled = True
|
||||
formdef.disabled_redirection = 'http://example.net'
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
resp1 = get_app(pub).get(sign_uri('/json'))
|
||||
assert resp1.json['err'] == 0
|
||||
assert resp1.json['data'][0]['title'] == 'test'
|
||||
assert resp1.json['data'][0]['url'] == 'http://example.net/test/'
|
||||
assert resp1.json['data'][0]['redirection'] is True
|
||||
assert 'count' not in resp1.json['data'][0]
|
||||
|
||||
|
||||
def test_backoffice_submission_formdef_list(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.id = '14'
|
||||
role.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.description = 'plop'
|
||||
formdef.workflow_roles = {'_receiver': str(role.id)}
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
formdef2 = FormDef()
|
||||
formdef2.name = 'ignore me'
|
||||
formdef2.fields = []
|
||||
formdef2.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# check it's not advertised ...
|
||||
formdef.backoffice_submission_roles = [role.id]
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# even if it's advertised on frontoffice
|
||||
formdef.always_advertise = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
# even if user is admin
|
||||
local_user.is_admin = True
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(
|
||||
sign_uri('/api/formdefs/?backoffice-submission=on&NameID=%s' % local_user.name_identifiers[0])
|
||||
)
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
local_user.is_admin = False
|
||||
local_user.store()
|
||||
|
||||
# ... unless user has correct roles
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(
|
||||
sign_uri('/api/formdefs/?backoffice-submission=on&NameID=%s' % local_user.name_identifiers[0])
|
||||
)
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert 'backoffice_submission_url' in resp.json['data'][0]
|
||||
|
||||
# but not advertised if it's a redirection
|
||||
formdef.disabled = True
|
||||
formdef.disabled_redirection = 'http://example.net'
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/formdefs/?backoffice-submission=on'))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
|
||||
def test_formdef_schema(pub):
|
||||
Workflow.wipe()
|
||||
workflow = Workflow(name='test')
|
||||
st1 = workflow.add_status('Status1', 'st1')
|
||||
jump = JumpWorkflowStatusItem()
|
||||
jump.status = 'st2'
|
||||
jump.timeout = 100
|
||||
st1.items.append(jump)
|
||||
st2 = workflow.add_status('Status2', 'st2')
|
||||
jump = JumpWorkflowStatusItem()
|
||||
jump.status = 'st3'
|
||||
st2.items.append(jump)
|
||||
st2 = workflow.add_status('Status3', 'st3')
|
||||
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
|
||||
workflow.backoffice_fields_formdef.fields = [
|
||||
fields.StringField(id='bo1', label='1st backoffice field', type='string', varname='backoffice_blah'),
|
||||
]
|
||||
workflow.store()
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar'),
|
||||
fields.ItemField(
|
||||
id='1',
|
||||
label='foobar1',
|
||||
varname='foobar1',
|
||||
data_source={
|
||||
'type': 'json',
|
||||
'value': 'http://datasource.com',
|
||||
},
|
||||
),
|
||||
fields.ItemsField(
|
||||
id='2',
|
||||
label='foobar2',
|
||||
varname='foobar2',
|
||||
data_source={
|
||||
'type': 'formula',
|
||||
'value': '[dict(id=i, text=\'label %s\' % i, foo=i) for i in range(10)]',
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
formdef.workflow_id = workflow.id
|
||||
formdef.store()
|
||||
|
||||
with mock.patch('wcs.qommon.misc.urlopen') as urlopen:
|
||||
urlopen.side_effect = lambda *args: StringIO(
|
||||
'''\
|
||||
{"data": [{"id": 0, "text": "zéro", "foo": "bar"}, \
|
||||
{"id": 1, "text": "uné", "foo": "bar1"}, \
|
||||
{"id": 2, "text": "deux", "foo": "bar2"}]}'''
|
||||
)
|
||||
resp = get_app(pub).get('/api/formdefs/test/schema')
|
||||
resp2 = get_app(pub).get('/test/schema')
|
||||
resp3 = get_app(pub).get(sign_url('/api/formdefs/test/schema?orig=coucou', '1234'))
|
||||
resp4 = get_app(pub).get(sign_url('/api/formdefs/test/schema?orig=coucou', '1234'))
|
||||
|
||||
# check schema
|
||||
assert resp.json == resp2.json
|
||||
assert set(resp.json.keys()) >= set(
|
||||
[
|
||||
'enable_tracking_codes',
|
||||
'url_name',
|
||||
'description',
|
||||
'workflow',
|
||||
'expiration_date',
|
||||
'discussion',
|
||||
'has_captcha',
|
||||
'always_advertise',
|
||||
'name',
|
||||
'disabled',
|
||||
'only_allow_one',
|
||||
'fields',
|
||||
'keywords',
|
||||
'publication_date',
|
||||
'detailed_emails',
|
||||
'disabled_redirection',
|
||||
]
|
||||
)
|
||||
assert resp.json['name'] == 'test'
|
||||
|
||||
# fields checks
|
||||
assert resp.json['fields'][0]['label'] == 'foobar'
|
||||
assert resp.json['fields'][0]['type'] == 'string'
|
||||
|
||||
assert resp.json['fields'][1]['label'] == 'foobar1'
|
||||
assert resp.json['fields'][1]['type'] == 'item'
|
||||
|
||||
# check structured items are only exported for authenticated callers
|
||||
assert resp.json['fields'][1]['items'] == []
|
||||
assert resp.json['fields'][2]['items'] == []
|
||||
assert 'structured_items' not in resp.json['fields'][1]
|
||||
assert 'structured_items' not in resp.json['fields'][2]
|
||||
|
||||
assert len(resp3.json['fields'][1]['structured_items']) == 3
|
||||
assert resp3.json['fields'][1]['structured_items'][0]['id'] == 0
|
||||
assert resp3.json['fields'][1]['structured_items'][0]['text'] == u'zéro'
|
||||
assert resp3.json['fields'][1]['structured_items'][0]['foo'] == 'bar'
|
||||
assert resp3.json['fields'][1]['items'][0] == u'zéro'
|
||||
|
||||
assert resp3.json['fields'][2]['label'] == 'foobar2'
|
||||
assert resp3.json['fields'][2]['type'] == 'items'
|
||||
assert len(resp3.json['fields'][2]['structured_items']) == 10
|
||||
assert resp3.json['fields'][2]['structured_items'][0]['id'] == 0
|
||||
assert resp3.json['fields'][2]['structured_items'][0]['text'] == 'label 0'
|
||||
assert resp3.json['fields'][2]['structured_items'][0]['foo'] == 0
|
||||
assert resp3.json['fields'][2]['items'][0] == 'label 0'
|
||||
|
||||
# if structured_items fails no values
|
||||
assert 'structured_items' not in resp4.json['fields'][1]
|
||||
assert resp4.json['fields'][1]['items'] == []
|
||||
|
||||
# workflow checks
|
||||
assert len(resp.json['workflow']['statuses']) == 3
|
||||
assert resp.json['workflow']['statuses'][0]['id'] == 'st1'
|
||||
assert resp.json['workflow']['statuses'][0]['endpoint'] is False
|
||||
assert resp.json['workflow']['statuses'][0]['waitpoint'] is True
|
||||
assert resp.json['workflow']['statuses'][1]['id'] == 'st2'
|
||||
assert resp.json['workflow']['statuses'][1]['endpoint'] is False
|
||||
assert resp.json['workflow']['statuses'][1]['waitpoint'] is False
|
||||
assert resp.json['workflow']['statuses'][2]['id'] == 'st3'
|
||||
assert resp.json['workflow']['statuses'][2]['endpoint'] is True
|
||||
assert resp.json['workflow']['statuses'][2]['waitpoint'] is True
|
||||
assert len(resp.json['workflow']['fields']) == 1
|
||||
|
||||
assert resp.json['workflow']['fields'][0]['label'] == '1st backoffice field'
|
||||
|
||||
get_app(pub).get('/api/formdefs/xxx/schema', status=404)
|
||||
|
||||
|
||||
def test_post_invalid_json(pub, local_user):
|
||||
resp = get_app(pub).post(
|
||||
'/api/formdefs/test/submit', params='not a json payload', content_type='application/json', status=400
|
||||
)
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_class'] == 'Invalid request'
|
||||
|
||||
|
||||
def test_formdef_submit(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [fields.StringField(id='0', label='foobar')]
|
||||
formdef.store()
|
||||
data_class = formdef.data_class()
|
||||
|
||||
resp = get_app(pub).post_json('/api/formdefs/test/submit', {'data': {}}, status=403)
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_desc'] == 'unsigned API call'
|
||||
|
||||
def url():
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/formdefs/test/submit'
|
||||
+ '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
return signed_url[len('http://example.net') :]
|
||||
|
||||
resp = get_app(pub).post_json(url(), {'data': {}})
|
||||
assert resp.json['err'] == 0
|
||||
assert resp.json['data']['url'] == ('http://example.net/test/%s/' % resp.json['data']['id'])
|
||||
assert resp.json['data']['backoffice_url'] == (
|
||||
'http://example.net/backoffice/management/test/%s/' % resp.json['data']['id']
|
||||
)
|
||||
assert resp.json['data']['api_url'] == ('http://example.net/api/forms/test/%s/' % resp.json['data']['id'])
|
||||
assert data_class.get(resp.json['data']['id']).status == 'wf-new'
|
||||
assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
|
||||
assert data_class.get(resp.json['data']['id']).tracking_code is None
|
||||
|
||||
local_user2 = get_publisher().user_class()
|
||||
local_user2.name = 'Test'
|
||||
local_user2.email = 'foo@localhost'
|
||||
local_user2.store()
|
||||
resp = get_app(pub).post_json(url(), {'data': {}, 'user': {'NameID': [], 'email': local_user2.email}})
|
||||
assert data_class.get(resp.json['data']['id']).user.email == local_user2.email
|
||||
|
||||
resp = get_app(pub).post(
|
||||
url(), json.dumps({'data': {}}), status=400
|
||||
) # missing Content-Type: application/json header
|
||||
assert resp.json['err_desc'] == 'expected JSON but missing appropriate content-type'
|
||||
|
||||
# check qualified content type are recognized
|
||||
resp = get_app(pub).post(url(), json.dumps({'data': {}}), content_type='application/json; charset=utf-8')
|
||||
assert resp.json['data']['url']
|
||||
|
||||
formdef.disabled = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).post_json(url(), {'data': {}}, status=403)
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_desc'] == 'disabled form'
|
||||
|
||||
formdef.disabled = False
|
||||
formdef.store()
|
||||
resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)
|
||||
formdef.backoffice_submission_roles = ['xx']
|
||||
formdef.store()
|
||||
resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}}, status=403)
|
||||
formdef.backoffice_submission_roles = [role.id]
|
||||
formdef.store()
|
||||
resp = get_app(pub).post_json(url(), {'meta': {'backoffice-submission': True}, 'data': {}})
|
||||
assert data_class.get(resp.json['data']['id']).status == 'wf-new'
|
||||
assert data_class.get(resp.json['data']['id']).backoffice_submission is True
|
||||
assert data_class.get(resp.json['data']['id']).user_id is None
|
||||
assert data_class.get(resp.json['data']['id']).submission_agent_id == str(local_user.id)
|
||||
|
||||
formdef.enable_tracking_codes = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).post_json(url(), {'data': {}})
|
||||
assert data_class.get(resp.json['data']['id']).tracking_code
|
||||
|
||||
resp = get_app(pub).post_json(url(), {'meta': {'draft': True}, 'data': {}})
|
||||
assert data_class.get(resp.json['data']['id']).status == 'draft'
|
||||
|
||||
resp = get_app(pub).post_json(
|
||||
url(),
|
||||
{
|
||||
'meta': {'backoffice-submission': True},
|
||||
'data': {},
|
||||
'context': {'channel': 'mail', 'comments': 'blah'},
|
||||
},
|
||||
)
|
||||
assert data_class.get(resp.json['data']['id']).status == 'wf-new'
|
||||
assert data_class.get(resp.json['data']['id']).backoffice_submission is True
|
||||
assert data_class.get(resp.json['data']['id']).user_id is None
|
||||
assert data_class.get(resp.json['data']['id']).submission_context == {'comments': 'blah'}
|
||||
assert data_class.get(resp.json['data']['id']).submission_channel == 'mail'
|
||||
|
||||
data_class.wipe()
|
||||
|
||||
|
||||
def test_formdef_submit_only_one(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.only_allow_one = True
|
||||
formdef.fields = [fields.StringField(id='0', label='foobar')]
|
||||
formdef.store()
|
||||
data_class = formdef.data_class()
|
||||
|
||||
def url():
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/formdefs/test/submit'
|
||||
+ '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
return signed_url[len('http://example.net') :]
|
||||
|
||||
resp = get_app(pub).post_json(url(), {'data': {}})
|
||||
assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
|
||||
|
||||
assert data_class.count() == 1
|
||||
|
||||
resp = get_app(pub).post_json(url(), {'data': {}}, status=403)
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_desc'] == 'only one formdata by user is allowed'
|
||||
|
||||
formdata = data_class.select()[0]
|
||||
formdata.user_id = '1000' # change owner
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).post_json(url(), {'data': {}}, status=200)
|
||||
assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
|
||||
assert data_class.count() == 2
|
||||
|
||||
|
||||
def test_formdef_submit_with_varname(pub, local_user):
|
||||
NamedDataSource.wipe()
|
||||
data_source = NamedDataSource(name='foobar')
|
||||
source = [{'id': '1', 'text': 'foo', 'more': 'XXX'}, {'id': '2', 'text': 'bar', 'more': 'YYY'}]
|
||||
data_source.data_source = {'type': 'formula', 'value': repr(source)}
|
||||
data_source.store()
|
||||
|
||||
data_source = NamedDataSource(name='foobar_jsonp')
|
||||
data_source.data_source = {'type': 'formula', 'value': 'http://example.com/jsonp'}
|
||||
data_source.store()
|
||||
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar0', varname='foobar0'),
|
||||
fields.ItemField(id='1', label='foobar1', varname='foobar1', data_source={'type': 'foobar'}),
|
||||
fields.ItemField(id='2', label='foobar2', varname='foobar2', data_source={'type': 'foobar_jsonp'}),
|
||||
fields.DateField(id='3', label='foobar3', varname='date'),
|
||||
fields.FileField(id='4', label='foobar4', varname='file'),
|
||||
fields.MapField(id='5', label='foobar5', varname='map'),
|
||||
fields.StringField(id='6', label='foobar6', varname='foobar6'),
|
||||
]
|
||||
formdef.store()
|
||||
data_class = formdef.data_class()
|
||||
|
||||
resp = get_app(pub).post_json('/api/formdefs/test/submit', {'data': {}}, status=403)
|
||||
assert resp.json['err'] == 1
|
||||
assert resp.json['err_desc'] == 'unsigned API call'
|
||||
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/formdefs/test/submit'
|
||||
+ '?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
url = signed_url[len('http://example.net') :]
|
||||
payload = {
|
||||
'data': {
|
||||
'foobar0': 'xxx',
|
||||
'foobar1': '1',
|
||||
'foobar1_structured': {
|
||||
'id': '1',
|
||||
'text': 'foo',
|
||||
'more': 'XXX',
|
||||
},
|
||||
'foobar2': 'bar',
|
||||
'foobar2_raw': '10',
|
||||
'date': '1970-01-01',
|
||||
'file': {
|
||||
'filename': 'test.txt',
|
||||
'content': force_text(base64.b64encode(b'test')),
|
||||
},
|
||||
'map': {
|
||||
'lat': 1.5,
|
||||
'lon': 2.25,
|
||||
},
|
||||
}
|
||||
}
|
||||
resp = get_app(pub).post_json(url, payload)
|
||||
assert resp.json['err'] == 0
|
||||
assert data_class.get(resp.json['data']['id']).status == 'wf-new'
|
||||
assert data_class.get(resp.json['data']['id']).user_id == str(local_user.id)
|
||||
assert data_class.get(resp.json['data']['id']).tracking_code is None
|
||||
assert data_class.get(resp.json['data']['id']).data['0'] == 'xxx'
|
||||
assert data_class.get(resp.json['data']['id']).data['1'] == '1'
|
||||
assert data_class.get(resp.json['data']['id']).data['1_structured'] == source[0]
|
||||
assert data_class.get(resp.json['data']['id']).data['2'] == '10'
|
||||
assert data_class.get(resp.json['data']['id']).data['2_display'] == 'bar'
|
||||
assert data_class.get(resp.json['data']['id']).data['3'] == time.struct_time(
|
||||
(1970, 1, 1, 0, 0, 0, 3, 1, -1)
|
||||
)
|
||||
|
||||
assert data_class.get(resp.json['data']['id']).data['4'].orig_filename == 'test.txt'
|
||||
assert data_class.get(resp.json['data']['id']).data['4'].get_content() == b'test'
|
||||
assert data_class.get(resp.json['data']['id']).data['5'] == '1.5;2.25'
|
||||
# test bijectivity
|
||||
assert (
|
||||
formdef.fields[3].get_json_value(data_class.get(resp.json['data']['id']).data['3'])
|
||||
== payload['data']['date']
|
||||
)
|
||||
for k in payload['data']['file']:
|
||||
data = data_class.get(resp.json['data']['id']).data['4']
|
||||
assert formdef.fields[4].get_json_value(data)[k] == payload['data']['file'][k]
|
||||
assert (
|
||||
formdef.fields[5].get_json_value(data_class.get(resp.json['data']['id']).data['5'])
|
||||
== payload['data']['map']
|
||||
)
|
||||
|
||||
data_class.wipe()
|
||||
|
||||
|
||||
def test_formdef_submit_from_wscall(pub, local_user):
|
||||
test_formdef_submit_with_varname(pub, local_user)
|
||||
formdef = FormDef.select()[0]
|
||||
workflow = Workflow.get_default_workflow()
|
||||
workflow.id = '2'
|
||||
workflow.backoffice_fields_formdef = WorkflowBackofficeFieldsFormDef(workflow)
|
||||
workflow.backoffice_fields_formdef.fields = [
|
||||
fields.StringField(id='bo1', label='1st backoffice field', type='string', varname='backoffice_blah'),
|
||||
]
|
||||
workflow.store()
|
||||
formdef.workflow = workflow
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
|
||||
upload = PicklableUpload('test.txt', 'text/plain', 'ascii')
|
||||
upload.receive([b'test'])
|
||||
|
||||
formdata.data = {
|
||||
'0': 'xxx',
|
||||
'1': '1',
|
||||
'1_display': '1',
|
||||
'1_structured': {
|
||||
'id': '1',
|
||||
'text': 'foo',
|
||||
'more': 'XXX',
|
||||
},
|
||||
'2': '10',
|
||||
'2_display': 'bar',
|
||||
'3': time.strptime('1970-01-01', '%Y-%m-%d'),
|
||||
'4': upload,
|
||||
'5': '1.5;2.25',
|
||||
'bo1': 'backoffice field',
|
||||
}
|
||||
formdata.just_created()
|
||||
formdata.evolution[-1].status = 'wf-new'
|
||||
formdata.store()
|
||||
|
||||
payload = json.loads(json.dumps(formdata.get_json_export_dict(), cls=qommon.misc.JSONEncoder))
|
||||
signed_url = sign_url('http://example.net/api/formdefs/test/submit?orig=coucou', '1234')
|
||||
url = signed_url[len('http://example.net') :]
|
||||
|
||||
resp = get_app(pub).post_json(url, payload)
|
||||
assert resp.json['err'] == 0
|
||||
new_formdata = formdef.data_class().get(resp.json['data']['id'])
|
||||
assert new_formdata.data['0'] == formdata.data['0']
|
||||
assert new_formdata.data['1'] == formdata.data['1']
|
||||
assert new_formdata.data['1_display'] == formdata.data['1_display']
|
||||
assert new_formdata.data['1_structured'] == formdata.data['1_structured']
|
||||
assert new_formdata.data['2'] == formdata.data['2']
|
||||
assert new_formdata.data['2_display'] == formdata.data['2_display']
|
||||
assert new_formdata.data['3'] == formdata.data['3']
|
||||
assert new_formdata.data['4'].get_content() == formdata.data['4'].get_content()
|
||||
assert new_formdata.data['5'] == formdata.data['5']
|
||||
assert new_formdata.data['bo1'] == formdata.data['bo1']
|
||||
assert not new_formdata.data.get('6')
|
||||
assert new_formdata.user_id is None
|
||||
|
||||
# add an extra attribute
|
||||
payload['extra'] = {'foobar6': 'YYY'}
|
||||
signed_url = sign_url('http://example.net/api/formdefs/test/submit?orig=coucou', '1234')
|
||||
url = signed_url[len('http://example.net') :]
|
||||
resp = get_app(pub).post_json(url, payload)
|
||||
assert resp.json['err'] == 0
|
||||
new_formdata = formdef.data_class().get(resp.json['data']['id'])
|
||||
assert new_formdata.data['0'] == formdata.data['0']
|
||||
assert new_formdata.data['6'] == 'YYY'
|
||||
|
||||
# add user
|
||||
formdata.user_id = local_user.id
|
||||
formdata.store()
|
||||
|
||||
payload = json.loads(json.dumps(formdata.get_json_export_dict(), cls=qommon.misc.JSONEncoder))
|
||||
signed_url = sign_url('http://example.net/api/formdefs/test/submit?orig=coucou', '1234')
|
||||
url = signed_url[len('http://example.net') :]
|
||||
|
||||
resp = get_app(pub).post_json(url, payload)
|
||||
assert resp.json['err'] == 0
|
||||
new_formdata = formdef.data_class().get(resp.json['data']['id'])
|
||||
assert str(new_formdata.user_id) == str(local_user.id)
|
||||
|
||||
# test missing map data
|
||||
del formdata.data['5']
|
||||
|
||||
payload = json.loads(json.dumps(formdata.get_json_export_dict(), cls=qommon.misc.JSONEncoder))
|
||||
signed_url = sign_url('http://example.net/api/formdefs/test/submit?orig=coucou', '1234')
|
||||
url = signed_url[len('http://example.net') :]
|
||||
|
||||
resp = get_app(pub).post_json(url, payload)
|
||||
assert resp.json['err'] == 0
|
||||
new_formdata = formdef.data_class().get(resp.json['data']['id'])
|
||||
assert new_formdata.data.get('5') is None
|
||||
|
||||
|
||||
def test_formdef_submit_structured(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='test')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.ItemField(
|
||||
id='0',
|
||||
label='foobar',
|
||||
varname='foobar',
|
||||
data_source={
|
||||
'type': 'json',
|
||||
'value': 'http://datasource.com',
|
||||
},
|
||||
),
|
||||
fields.ItemField(
|
||||
id='1',
|
||||
label='foobar1',
|
||||
varname='foobar1',
|
||||
data_source={
|
||||
'type': 'formula',
|
||||
'value': '[dict(id=i, text=\'label %s\' % i, foo=i) for i in range(10)]',
|
||||
},
|
||||
),
|
||||
]
|
||||
formdef.store()
|
||||
data_class = formdef.data_class()
|
||||
|
||||
def url():
|
||||
signed_url = sign_url(
|
||||
'http://example.net/api/formdefs/test/submit'
|
||||
'?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234',
|
||||
)
|
||||
return signed_url[len('http://example.net') :]
|
||||
|
||||
with mock.patch('wcs.qommon.misc.urlopen') as urlopen:
|
||||
urlopen.side_effect = lambda *args: StringIO(
|
||||
'''\
|
||||
{"data": [{"id": 0, "text": "zéro", "foo": "bar"}, \
|
||||
{"id": 1, "text": "uné", "foo": "bar1"}, \
|
||||
{"id": 2, "text": "deux", "foo": "bar2"}]}'''
|
||||
)
|
||||
resp = get_app(pub).post_json(
|
||||
url(),
|
||||
{
|
||||
'data': {
|
||||
'0': '0',
|
||||
"1": '3',
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
formdata = data_class.get(resp.json['data']['id'])
|
||||
assert formdata.status == 'wf-new'
|
||||
assert formdata.data['0'] == '0'
|
||||
assert formdata.data['0_display'] == 'zéro'
|
||||
assert formdata.data['0_structured'] == {
|
||||
'id': 0,
|
||||
'text': 'zéro',
|
||||
'foo': 'bar',
|
||||
}
|
||||
assert formdata.data['1'] == '3'
|
||||
assert formdata.data['1_display'] == 'label 3'
|
||||
assert formdata.data['1_structured'] == {
|
||||
'id': 3,
|
||||
'text': 'label 3',
|
||||
'foo': 3,
|
||||
}
|
||||
|
||||
data_class.wipe()
|
||||
|
|
@ -0,0 +1,457 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import datetime
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs import fields
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.form import PicklableUpload
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.ident.password_accounts import PasswordAccount
|
||||
from wcs.roles import Role
|
||||
from wcs.workflows import Workflow, WorkflowVariablesFieldsFormDef
|
||||
|
||||
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_roles(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='Hello World')
|
||||
role.emails = ['toto@example.com', 'zozo@example.com']
|
||||
role.details = 'kouign amann'
|
||||
role.store()
|
||||
|
||||
resp = get_app(pub).get('/api/roles', status=403)
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/roles'))
|
||||
assert resp.json['data'][0]['text'] == 'Hello World'
|
||||
assert resp.json['data'][0]['slug'] == 'hello-world'
|
||||
assert resp.json['data'][0]['emails'] == ['toto@example.com', 'zozo@example.com']
|
||||
assert resp.json['data'][0]['emails_to_members'] is False
|
||||
assert resp.json['data'][0]['details'] == 'kouign amann'
|
||||
|
||||
# also check old endpoint, for compatibility
|
||||
resp = get_app(pub).get(sign_uri('/roles'), headers={'Accept': 'application/json'})
|
||||
assert resp.json['data'][0]['text'] == 'Hello World'
|
||||
assert resp.json['data'][0]['slug'] == 'hello-world'
|
||||
assert resp.json['data'][0]['emails'] == ['toto@example.com', 'zozo@example.com']
|
||||
assert resp.json['data'][0]['emails_to_members'] is False
|
||||
assert resp.json['data'][0]['details'] == 'kouign amann'
|
||||
|
||||
|
||||
def test_users(pub, local_user):
|
||||
resp = get_app(pub).get('/api/users/', status=403)
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/'))
|
||||
assert resp.json['data'][0]['user_display_name'] == local_user.name
|
||||
assert resp.json['data'][0]['user_email'] == local_user.email
|
||||
assert resp.json['data'][0]['user_id'] == local_user.id
|
||||
|
||||
role = Role(name='Foo bar')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=jean'))
|
||||
assert resp.json['data'][0]['user_email'] == local_user.email
|
||||
assert len(resp.json['data'][0]['user_roles']) == 1
|
||||
assert resp.json['data'][0]['user_roles'][0]['name'] == 'Foo bar'
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=foobar'))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
from wcs.admin.settings import UserFieldsFormDef
|
||||
|
||||
formdef = UserFieldsFormDef(pub)
|
||||
formdef.fields.append(fields.StringField(id='3', label='test', type='string'))
|
||||
formdef.store()
|
||||
|
||||
local_user.form_data = {'3': 'HELLO'}
|
||||
local_user.set_attributes_from_formdata(local_user.form_data)
|
||||
local_user.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=HELLO'))
|
||||
assert len(resp.json['data']) == 1
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=foobar'))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
|
||||
def test_users_unaccent(pub, local_user):
|
||||
local_user.name = 'Jean Sénisme'
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=jean'))
|
||||
assert resp.json['data'][0]['user_email'] == local_user.email
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=senisme'))
|
||||
assert resp.json['data'][0]['user_email'] == local_user.email
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=sénisme'))
|
||||
assert resp.json['data'][0]['user_email'] == local_user.email
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/?q=blah'))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
|
||||
def test_user_by_nameid(pub, local_user):
|
||||
resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user), status=404)
|
||||
local_user.name_identifiers = ['xyz']
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user))
|
||||
assert str(resp.json['id']) == str(local_user.id)
|
||||
|
||||
|
||||
def test_user_roles(pub, local_user):
|
||||
local_user.name_identifiers = ['xyz']
|
||||
local_user.store()
|
||||
role = Role(name='Foo bar')
|
||||
role.store()
|
||||
local_user.roles = [role.id]
|
||||
local_user.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/users/xyz/', user=local_user))
|
||||
assert len(resp.json['user_roles']) == 1
|
||||
assert resp.json['user_roles'][0]['name'] == 'Foo bar'
|
||||
|
||||
|
||||
def test_user_forms(pub, local_user):
|
||||
Workflow.wipe()
|
||||
workflow = Workflow.get_default_workflow()
|
||||
workflow.id = '2'
|
||||
workflow.variables_formdef = WorkflowVariablesFieldsFormDef(workflow=workflow)
|
||||
workflow.variables_formdef.fields.append(
|
||||
fields.DateField(label='Test', type='date', varname='option_date')
|
||||
)
|
||||
workflow.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
fields.StringField(id='1', label='foobar2'),
|
||||
fields.DateField(id='2', label='date', type='date', varname='date'),
|
||||
]
|
||||
formdef.keywords = 'hello, world'
|
||||
formdef.disabled = False
|
||||
formdef.enable_tracking_codes = True
|
||||
formdef.workflow = workflow
|
||||
formdef.workflow_options = {'option_date': datetime.date(2020, 1, 15).timetuple()}
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {
|
||||
'0': 'foo@localhost',
|
||||
'1': 'xxx',
|
||||
'2': datetime.date(2020, 1, 15).timetuple(),
|
||||
}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.jump_status('new')
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
resp2 = get_app(pub).get(sign_uri('/myspace/forms', user=local_user))
|
||||
resp3 = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['form_name'] == 'test'
|
||||
assert resp.json['data'][0]['form_slug'] == 'test'
|
||||
assert resp.json['data'][0]['form_status'] == 'New'
|
||||
assert datetime.datetime.strptime(resp.json['data'][0]['form_receipt_datetime'], '%Y-%m-%dT%H:%M:%S')
|
||||
assert resp.json['data'][0]['keywords'] == ['hello', 'world']
|
||||
assert resp.json == resp2.json == resp3.json
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms?full=on', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert resp.json['data'][0]['fields']['foobar'] == 'foo@localhost'
|
||||
assert resp.json['data'][0]['fields']['date'] == '2020-01-15'
|
||||
assert resp.json['data'][0]['keywords'] == ['hello', 'world']
|
||||
assert resp.json['data'][0]['form_option_option_date'] == '2020-01-15'
|
||||
resp2 = get_app(pub).get(sign_uri('/api/user/forms?&full=on', user=local_user))
|
||||
assert resp.json == resp2.json
|
||||
|
||||
formdef.disabled = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
# check digest is part of contents
|
||||
formdef.digest_template = 'XYZ'
|
||||
formdef.data_class().get(formdata.id).store()
|
||||
assert formdef.data_class().get(formdata.id).digest == 'XYZ'
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
assert resp.json['data'][0]['form_digest'] == 'XYZ'
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms?NameID=xxx'))
|
||||
assert resp.json == {'err': 1, 'err_desc': 'unknown NameID', 'data': []}
|
||||
resp2 = get_app(pub).get(sign_uri('/api/user/forms?&NameID=xxx'))
|
||||
assert resp.json == resp2.json
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.user_id = local_user.id
|
||||
formdata.status = 'draft'
|
||||
formdata.receipt_time = datetime.datetime(2015, 1, 1).timetuple()
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms?include-drafts=true', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
formdef.disabled = False
|
||||
formdef.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms?include-drafts=true', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 2
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xyy'}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.receipt_time = (datetime.datetime.now() + datetime.timedelta(days=1)).timetuple()
|
||||
formdata.jump_status('new')
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/forms', user=local_user))
|
||||
assert len(resp.json['data']) == 2
|
||||
resp2 = get_app(pub).get(sign_uri('/api/user/forms?sort=desc', user=local_user))
|
||||
assert len(resp2.json['data']) == 2
|
||||
assert resp2.json['data'][0] == resp.json['data'][1]
|
||||
assert resp2.json['data'][1] == resp.json['data'][0]
|
||||
|
||||
|
||||
def test_user_forms_limit_offset(pub, local_user):
|
||||
if not pub.is_using_postgresql():
|
||||
pytest.skip('this requires SQL')
|
||||
return
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test limit offset'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
fields.StringField(id='1', label='foobar2'),
|
||||
]
|
||||
formdef.keywords = 'hello, world'
|
||||
formdef.disabled = False
|
||||
formdef.enable_tracking_codes = False
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
for i in range(50):
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {'0': 'foo@localhost', '1': str(i)}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.receipt_time = (datetime.datetime.now() + datetime.timedelta(days=i)).timetuple()
|
||||
formdata.jump_status('new')
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 50
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10' % local_user.id))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 10
|
||||
assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(1, 11)]
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10&offset=45' % local_user.id))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 5
|
||||
assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(46, 51)]
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms?limit=10&sort=desc' % local_user.id))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 10
|
||||
assert [x['form_number_raw'] for x in resp.json['data']] == [str(x) for x in range(50, 40, -1)]
|
||||
|
||||
|
||||
def test_user_forms_from_agent(pub, local_user):
|
||||
Role.wipe()
|
||||
role = Role(name='Foo bar')
|
||||
role.store()
|
||||
|
||||
agent_user = get_publisher().user_class()
|
||||
agent_user.name = 'Agent'
|
||||
agent_user.email = 'agent@example.com'
|
||||
agent_user.name_identifiers = ['ABCDE']
|
||||
agent_user.roles = [role.id]
|
||||
agent_user.store()
|
||||
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
fields.StringField(id='1', label='foobar2'),
|
||||
]
|
||||
formdef.store()
|
||||
formdef.data_class().wipe()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xxx'}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.jump_status('new')
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json['data'][0]['form_name'] == 'test'
|
||||
assert resp.json['data'][0]['form_slug'] == 'test'
|
||||
assert resp.json['data'][0]['form_status'] == 'New'
|
||||
assert resp.json['data'][0]['readable'] is False
|
||||
|
||||
formdef.skip_from_360_view = True
|
||||
formdef.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
formdef.workflow_roles = {'_receiver': str(role.id)}
|
||||
formdef.store()
|
||||
formdef.data_class().rebuild_security()
|
||||
resp = get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user))
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
agent_user.roles = []
|
||||
agent_user.store()
|
||||
get_app(pub).get(sign_uri('/api/users/%s/forms' % local_user.id, user=agent_user), status=403)
|
||||
|
||||
|
||||
def test_user_drafts(pub, local_user):
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
formdef.fields = [
|
||||
fields.StringField(id='0', label='foobar', varname='foobar'),
|
||||
fields.StringField(id='1', label='foobar2'),
|
||||
fields.FileField(id='2', label='foobar3', varname='file'),
|
||||
]
|
||||
formdef.keywords = 'hello, world'
|
||||
formdef.disabled = False
|
||||
formdef.enable_tracking_codes = True
|
||||
formdef.store()
|
||||
|
||||
formdef.data_class().wipe()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
upload = PicklableUpload('test.txt', 'text/plain', 'ascii')
|
||||
upload.receive([b'base64me'])
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xxx', '2': upload}
|
||||
formdata.user_id = local_user.id
|
||||
formdata.page_no = 1
|
||||
formdata.status = 'draft'
|
||||
formdata.receipt_time = datetime.datetime(2015, 1, 1).timetuple()
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts', user=local_user))
|
||||
resp2 = get_app(pub).get(sign_uri('/myspace/drafts', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
assert resp.json == resp2.json
|
||||
assert 'fields' not in resp.json['data'][0]
|
||||
assert resp.json['data'][0]['keywords'] == ['hello', 'world']
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts?full=on', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert 'fields' in resp.json['data'][0]
|
||||
assert resp.json['data'][0]['fields']['foobar'] == 'foo@localhost'
|
||||
assert 'url' in resp.json['data'][0]['fields']['file']
|
||||
assert 'content' not in resp.json['data'][0]['fields']['file'] # no file content in full lists
|
||||
assert resp.json['data'][0]['keywords'] == ['hello', 'world']
|
||||
|
||||
formdef.enable_tracking_codes = False
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 1
|
||||
|
||||
formdef.enable_tracking_codes = True
|
||||
formdef.disabled = True
|
||||
formdef.store()
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts', user=local_user))
|
||||
assert resp.json['err'] == 0
|
||||
assert len(resp.json['data']) == 0
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/api/user/drafts?NameID=xxx'))
|
||||
assert resp.json == {'err': 1, 'err_desc': 'unknown NameID', 'data': []}
|
||||
resp2 = get_app(pub).get(sign_uri('/api/user/drafts?&NameID=xxx'))
|
||||
assert resp.json == resp2.json
|
||||
|
|
@ -1,12 +1,11 @@
|
|||
import pytest
|
||||
from quixote import cleanup, get_response
|
||||
from utilities import clean_temporary_pub, create_temporary_pub
|
||||
|
||||
from wcs import sessions
|
||||
from wcs.api_utils import get_query_flag
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
|
||||
from utilities import create_temporary_pub, clean_temporary_pub
|
||||
|
||||
|
||||
def setup_module(module):
|
||||
cleanup()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,387 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from quixote import get_publisher
|
||||
from utilities import clean_temporary_pub, create_temporary_pub, get_app
|
||||
|
||||
from wcs import fields
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.ident.password_accounts import PasswordAccount
|
||||
from wcs.roles import Role
|
||||
from wcs.wf.jump import JumpWorkflowStatusItem
|
||||
from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem
|
||||
from wcs.workflows import Workflow
|
||||
|
||||
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'
|
||||
|
||||
# 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'
|
||||
|
||||
Role.wipe()
|
||||
role = Role(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_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()
|
||||
Role.wipe()
|
||||
role = Role(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'
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
import base64
|
||||
import datetime
|
||||
import hashlib
|
||||
import hmac
|
||||
|
||||
from django.utils.encoding import force_bytes
|
||||
from django.utils.six.moves.urllib import parse as urllib
|
||||
from django.utils.six.moves.urllib import parse as urlparse
|
||||
|
||||
|
||||
def sign_uri(uri, user=None, format='json'):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
scheme, netloc, path, params, query, fragment = urlparse.urlparse(uri)
|
||||
if query:
|
||||
query += '&'
|
||||
if format:
|
||||
query += 'format=%s&' % format
|
||||
query += 'orig=coucou&algo=sha256×tamp=' + timestamp
|
||||
if user:
|
||||
query += '&email=' + urllib.quote(user.email)
|
||||
query += '&signature=%s' % urllib.quote(
|
||||
base64.b64encode(hmac.new(b'1234', force_bytes(query), hashlib.sha256).digest())
|
||||
)
|
||||
return urlparse.urlunparse((scheme, netloc, path, params, query, fragment))
|
||||
Loading…
Reference in New Issue