parent
594c6db59d
commit
1e200cd30c
|
@ -1,14 +1,17 @@
|
|||
import pytest
|
||||
import shutil
|
||||
import os
|
||||
import hmac
|
||||
import base64
|
||||
import hashlib
|
||||
import urllib
|
||||
import urlparse
|
||||
import datetime
|
||||
import time
|
||||
|
||||
from quixote import cleanup, get_publisher
|
||||
from wcs.users import User
|
||||
from wcs.roles import Role
|
||||
from wcs.formdef import FormDef
|
||||
from wcs.categories import Category
|
||||
from wcs import fields
|
||||
|
@ -16,20 +19,15 @@ from wcs.api import sign_url
|
|||
|
||||
from utilities import get_app, create_temporary_pub
|
||||
|
||||
pub, req, app_dir, user = None, None, None, None
|
||||
pub, req, app_dir = None, None, None
|
||||
|
||||
|
||||
def setup_module(module):
|
||||
cleanup()
|
||||
|
||||
global pub, req, app_dir, user
|
||||
global pub, req, app_dir
|
||||
pub = create_temporary_pub()
|
||||
|
||||
user = User()
|
||||
user.name = 'Jean Darmette'
|
||||
user.email = 'jean.darmette@triffouilis.fr'
|
||||
user.store()
|
||||
|
||||
file(os.path.join(pub.app_dir, 'site-options.cfg'), 'w').write('''\
|
||||
[api-secrets]
|
||||
coucou = 1234
|
||||
|
@ -40,15 +38,27 @@ def teardown_module(module):
|
|||
shutil.rmtree(pub.APP_DIR)
|
||||
|
||||
|
||||
def sign_uri(uri):
|
||||
@pytest.fixture
|
||||
def local_user():
|
||||
User.wipe()
|
||||
user = User()
|
||||
user.name = 'Jean Darmette'
|
||||
user.email = 'jean.darmette@triffouilis.fr'
|
||||
user.store()
|
||||
return user
|
||||
|
||||
def sign_uri(uri, user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(user.email) + '×tamp=' + timestamp
|
||||
signature = urllib.quote(
|
||||
scheme, netloc, path, params, query, fragment = urlparse.urlparse(uri)
|
||||
if query:
|
||||
query += '&'
|
||||
query += 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(user.email) + '×tamp=' + timestamp
|
||||
query += '&signature=%s' % urllib.quote(
|
||||
base64.b64encode(
|
||||
hmac.new('1234',
|
||||
query,
|
||||
hashlib.sha256).digest()))
|
||||
return uri + '?%s&signature=%s' % (query, signature)
|
||||
return urlparse.urlunparse((scheme, netloc, path, params, query, fragment))
|
||||
|
||||
def test_user_page_redirect():
|
||||
output = get_app(pub).get('/user')
|
||||
|
@ -98,9 +108,9 @@ def test_get_user_from_api_query_string_error_missing_email():
|
|||
output = get_app(pub).get('/user?%s&signature=%s' % (query, signature), status=403)
|
||||
assert output.json['err_desc'] == 'missing email or NameID fields'
|
||||
|
||||
def test_get_user_from_api_query_string_error_success_sha1():
|
||||
def test_get_user_from_api_query_string_error_success_sha1(local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha1&email=' + urllib.quote(user.email) + '×tamp=' + timestamp
|
||||
query = 'format=json&orig=coucou&algo=sha1&email=' + urllib.quote(local_user.email) + '×tamp=' + timestamp
|
||||
signature = urllib.quote(
|
||||
base64.b64encode(
|
||||
hmac.new('1234',
|
||||
|
@ -109,9 +119,9 @@ def test_get_user_from_api_query_string_error_success_sha1():
|
|||
output = get_app(pub).get('/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():
|
||||
def test_get_user_from_api_query_string_error_invalid_signature_algo_mismatch(local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(user.email) + '×tamp=' + timestamp
|
||||
query = 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(local_user.email) + '×tamp=' + timestamp
|
||||
signature = urllib.quote(
|
||||
base64.b64encode(
|
||||
hmac.new('1234',
|
||||
|
@ -120,20 +130,20 @@ def test_get_user_from_api_query_string_error_invalid_signature_algo_mismatch():
|
|||
output = get_app(pub).get('/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():
|
||||
def test_get_user_from_api_query_string_error_success_sha256(local_user):
|
||||
timestamp = datetime.datetime.utcnow().isoformat()[:19] + 'Z'
|
||||
query = 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(user.email) + '×tamp=' + timestamp
|
||||
query = 'format=json&orig=coucou&algo=sha256&email=' + urllib.quote(local_user.email) + '×tamp=' + timestamp
|
||||
signature = urllib.quote(
|
||||
base64.b64encode(
|
||||
hmac.new('1234',
|
||||
query,
|
||||
query,
|
||||
hashlib.sha256).digest()))
|
||||
output = get_app(pub).get('/user?%s&signature=%s' % (query, signature))
|
||||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
def test_sign_url():
|
||||
def test_sign_url(local_user):
|
||||
signed_url = sign_url(
|
||||
'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(user.email),
|
||||
'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'1234'
|
||||
)
|
||||
url = signed_url[len('http://example.net'):]
|
||||
|
@ -141,7 +151,7 @@ def test_sign_url():
|
|||
assert output.json['user_display_name'] == u'Jean Darmette'
|
||||
|
||||
signed_url = sign_url(
|
||||
'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(user.email),
|
||||
'http://example.net/user?format=json&orig=coucou&email=%s' % urllib.quote(local_user.email),
|
||||
'12345'
|
||||
)
|
||||
url = signed_url[len('http://example.net'):]
|
||||
|
@ -210,7 +220,7 @@ def test_categories():
|
|||
assert resp.json['data'][0]['url'] == 'http://example.net/category/'
|
||||
assert resp.json['data'][0]['description'] == 'hello world'
|
||||
|
||||
def test_formdata():
|
||||
def test_formdata(local_user):
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
|
@ -223,18 +233,18 @@ def test_formdata():
|
|||
formdata = formdef.data_class()()
|
||||
date = time.strptime('2014-01-20', '%Y-%m-%d')
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xxx', '2': date}
|
||||
formdata.user_id = user.id
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/test/%s/' % formdata.id))
|
||||
resp = get_app(pub).get(sign_uri('/test/%s/' % formdata.id, user=local_user))
|
||||
assert 'last_update_time' in resp.json
|
||||
assert resp.json['user']['name'] == user.name
|
||||
assert resp.json['user']['name'] == local_user.name
|
||||
assert resp.json['fields']['foobar'] == 'foo@localhost'
|
||||
assert resp.json['fields']['date'] == '2014-01-20'
|
||||
assert len(resp.json['fields']) == 2 # foobar2 has no varname, not in json
|
||||
|
||||
def test_myspace_forms():
|
||||
def test_myspace_forms(local_user):
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
|
@ -243,21 +253,21 @@ def test_myspace_forms():
|
|||
fields.StringField(id='1', label='foobar2'),]
|
||||
formdef.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/myspace/forms'))
|
||||
resp = get_app(pub).get(sign_uri('/myspace/forms', user=local_user))
|
||||
assert len(resp.json) == 0
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xxx'}
|
||||
formdata.user_id = user.id
|
||||
formdata.user_id = local_user.id
|
||||
formdata.just_created()
|
||||
formdata.jump_status('new')
|
||||
formdata.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/myspace/forms'))
|
||||
resp = get_app(pub).get(sign_uri('/myspace/forms', user=local_user))
|
||||
assert len(resp.json) == 1
|
||||
assert resp.json[0]['form_status'] == 'New'
|
||||
|
||||
def test_myspace_drafts():
|
||||
def test_myspace_drafts(local_user):
|
||||
FormDef.wipe()
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test'
|
||||
|
@ -266,16 +276,91 @@ def test_myspace_drafts():
|
|||
fields.StringField(id='1', label='foobar2'),]
|
||||
formdef.store()
|
||||
|
||||
resp = get_app(pub).get(sign_uri('/myspace/drafts'))
|
||||
resp = get_app(pub).get(sign_uri('/myspace/drafts', user=local_user))
|
||||
assert len(resp.json) == 0
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.data = {'0': 'foo@localhost', '1': 'xxx'}
|
||||
formdata.user_id = user.id
|
||||
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('/myspace/drafts'))
|
||||
resp = get_app(pub).get(sign_uri('/myspace/drafts', user=local_user))
|
||||
assert len(resp.json) == 1
|
||||
|
||||
def test_api_list_formdata(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'),
|
||||
fields.ItemField(id='1', label='foobar3', varname='foobar3', type='item',
|
||||
items=['foo', 'bar', 'baz']),
|
||||
]
|
||||
formdef.store()
|
||||
|
||||
data_class = formdef.data_class()
|
||||
data_class.wipe()
|
||||
|
||||
for i in range(30):
|
||||
formdata = data_class()
|
||||
date = time.strptime('2014-01-20', '%Y-%m-%d')
|
||||
formdata.data = {'0': 'FOO BAR %d' % i}
|
||||
if i%4 == 0:
|
||||
formdata.data['1'] = 'foo'
|
||||
formdata.data['1_display'] = 'foo'
|
||||
elif i%4 == 1:
|
||||
formdata.data['1'] = 'bar'
|
||||
formdata.data['1_display'] = 'bar'
|
||||
else:
|
||||
formdata.data['1'] = 'baz'
|
||||
formdata.data['1_display'] = 'baz'
|
||||
|
||||
formdata.just_created()
|
||||
if i%3 == 0:
|
||||
formdata.jump_status('new')
|
||||
else:
|
||||
formdata.jump_status('finished')
|
||||
formdata.store()
|
||||
|
||||
# check access is denied if the user has not the appropriate role
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list', user=local_user), status=403)
|
||||
|
||||
# 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
|
||||
assert 'receipt_time' in resp.json[0]
|
||||
assert not 'fields' in resp.json[0]
|
||||
|
||||
# check getting full formdata
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?full=on', user=local_user))
|
||||
assert len(resp.json) == 30
|
||||
assert 'receipt_time' in resp.json[0]
|
||||
assert 'fields' in resp.json[0]
|
||||
|
||||
# check filtered results
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-foobar3=foo', user=local_user))
|
||||
assert len(resp.json) == 8
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-foobar3=bar', user=local_user))
|
||||
assert len(resp.json) == 8
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter-foobar3=baz', user=local_user))
|
||||
assert len(resp.json) == 14
|
||||
|
||||
# check filter on status
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter=pending', user=local_user))
|
||||
assert len(resp.json) == 10
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter=done', user=local_user))
|
||||
assert len(resp.json) == 20
|
||||
resp = get_app(pub).get(sign_uri('/api/forms/test/list?filter=all', user=local_user))
|
||||
assert len(resp.json) == 30
|
||||
|
|
Loading…
Reference in New Issue