235 lines
7.7 KiB
Python
235 lines
7.7 KiB
Python
import io
|
|
import xml.etree.ElementTree as ET
|
|
|
|
import pytest
|
|
from webtest import Upload
|
|
|
|
from wcs.qommon.http_request import HTTPRequest
|
|
from wcs.wscalls import NamedWsCall
|
|
|
|
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
|
|
from .test_all import create_superuser
|
|
|
|
|
|
@pytest.fixture
|
|
def pub(request):
|
|
pub = create_temporary_pub(sql_mode=True)
|
|
|
|
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()
|
|
|
|
return pub
|
|
|
|
|
|
def teardown_module(module):
|
|
clean_temporary_pub()
|
|
|
|
|
|
@pytest.fixture
|
|
def wscall():
|
|
NamedWsCall.wipe()
|
|
wscall = NamedWsCall(name='xxx')
|
|
wscall.description = 'description'
|
|
wscall.notify_on_errors = True
|
|
wscall.record_on_errors = True
|
|
wscall.request = {
|
|
'url': 'http://remote.example.net/json',
|
|
'request_signature_key': 'xxx',
|
|
'qs_data': {'a': 'b'},
|
|
'method': 'POST',
|
|
'post_data': {'c': 'd'},
|
|
}
|
|
wscall.store()
|
|
return wscall
|
|
|
|
|
|
@pytest.mark.parametrize('value', [True, False])
|
|
def test_wscalls_new(pub, value):
|
|
create_superuser(pub)
|
|
NamedWsCall.wipe()
|
|
app = login(get_app(pub))
|
|
|
|
# go to the page and cancel
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click('New webservice call')
|
|
resp = resp.forms[0].submit('cancel')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/'
|
|
|
|
# go to the page and add a webservice call
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click('New webservice call')
|
|
assert resp.form['notify_on_errors'].value is None
|
|
assert resp.form['record_on_errors'].value == 'yes'
|
|
resp.form['name'] = 'a new webservice call'
|
|
resp.form['description'] = 'description'
|
|
resp.form['notify_on_errors'] = value
|
|
resp.form['record_on_errors'] = value
|
|
resp.form['request$url'] = 'http://remote.example.net/json'
|
|
resp = resp.form.submit('submit')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/'
|
|
resp = resp.follow()
|
|
assert 'a new webservice call' in resp.text
|
|
resp = resp.click('a new webservice call')
|
|
assert 'Webservice Call - a new webservice call' in resp.text
|
|
resp = resp.click('Edit')
|
|
assert 'Edit webservice call' in resp.text
|
|
|
|
assert NamedWsCall.get('a_new_webservice_call').name == 'a new webservice call'
|
|
assert NamedWsCall.get('a_new_webservice_call').notify_on_errors == value
|
|
assert NamedWsCall.get('a_new_webservice_call').record_on_errors == value
|
|
|
|
|
|
def test_wscalls_view(pub, wscall):
|
|
create_superuser(pub)
|
|
|
|
app = login(get_app(pub))
|
|
resp = app.get('/backoffice/settings/wscalls/%s/' % wscall.id)
|
|
assert 'http://remote.example.net/json' in resp.text
|
|
|
|
|
|
def test_wscalls_edit(pub, wscall):
|
|
create_superuser(pub)
|
|
|
|
app = login(get_app(pub))
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/xxx/')
|
|
resp = resp.click(href='edit')
|
|
assert resp.form['name'].value == 'xxx'
|
|
assert resp.form['notify_on_errors'].value == 'yes'
|
|
assert resp.form['record_on_errors'].value == 'yes'
|
|
assert 'slug' in resp.form.fields
|
|
resp.form['description'] = 'bla bla bla'
|
|
resp.form['notify_on_errors'] = False
|
|
resp.form['record_on_errors'] = False
|
|
resp = resp.form.submit('submit')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/xxx/'
|
|
resp = resp.follow()
|
|
|
|
assert NamedWsCall.get('xxx').description == 'bla bla bla'
|
|
assert NamedWsCall.get('xxx').notify_on_errors is False
|
|
assert NamedWsCall.get('xxx').record_on_errors is False
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/xxx/')
|
|
resp = resp.click(href='edit')
|
|
assert resp.form['name'].value == 'xxx'
|
|
assert resp.form['notify_on_errors'].value is None
|
|
assert resp.form['record_on_errors'].value is None
|
|
assert 'slug' in resp.form.fields
|
|
resp.form['slug'] = 'yyy'
|
|
resp = resp.form.submit('submit')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/yyy/'
|
|
|
|
|
|
def test_wscalls_delete(pub, wscall):
|
|
create_superuser(pub)
|
|
|
|
app = login(get_app(pub))
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/xxx/')
|
|
resp = resp.click(href='delete')
|
|
resp = resp.form.submit('cancel')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/'
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/xxx/')
|
|
resp = resp.click(href='delete')
|
|
resp = resp.form.submit('submit')
|
|
assert resp.location == 'http://example.net/backoffice/settings/wscalls/'
|
|
assert NamedWsCall.count() == 0
|
|
|
|
|
|
def test_wscalls_export(pub, wscall):
|
|
create_superuser(pub)
|
|
|
|
app = login(get_app(pub))
|
|
resp = app.get('/backoffice/settings/wscalls/xxx/')
|
|
|
|
resp = resp.click(href='export')
|
|
xml_export = resp.text
|
|
|
|
ds = io.StringIO(xml_export)
|
|
wscall2 = NamedWsCall.import_from_xml(ds)
|
|
assert wscall2.name == 'xxx'
|
|
|
|
|
|
def test_wscalls_import(pub, wscall):
|
|
create_superuser(pub)
|
|
|
|
wscall.slug = 'foobar'
|
|
wscall.store()
|
|
wscall_xml = ET.tostring(wscall.export_to_xml(include_id=True))
|
|
NamedWsCall.wipe()
|
|
assert NamedWsCall.count() == 0
|
|
|
|
app = login(get_app(pub))
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click(href='import')
|
|
resp.forms[0]['file'] = Upload('wscall.wcs', wscall_xml)
|
|
resp = resp.forms[0].submit()
|
|
assert NamedWsCall.count() == 1
|
|
assert {wc.slug for wc in NamedWsCall.select()} == {'foobar'}
|
|
|
|
# check slug
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click(href='import')
|
|
resp.forms[0]['file'] = Upload('wscall.wcs', wscall_xml)
|
|
resp = resp.forms[0].submit()
|
|
assert NamedWsCall.count() == 2
|
|
assert {wc.slug for wc in NamedWsCall.select()} == {'foobar', 'xxx'}
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click(href='import')
|
|
resp.forms[0]['file'] = Upload('wscall.wcs', wscall_xml)
|
|
resp = resp.forms[0].submit()
|
|
assert NamedWsCall.count() == 3
|
|
assert {wc.slug for wc in NamedWsCall.select()} == {'foobar', 'xxx', 'xxx_1'}
|
|
|
|
# import an invalid file
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click(href='import')
|
|
resp.form['file'] = Upload('wscall.wcs', b'garbage')
|
|
resp = resp.form.submit()
|
|
assert 'Invalid File' in resp.text
|
|
|
|
|
|
def test_wscalls_empty_param_values(pub):
|
|
create_superuser(pub)
|
|
NamedWsCall.wipe()
|
|
app = login(get_app(pub))
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click('New webservice call')
|
|
resp.form['name'] = 'a new webservice call'
|
|
resp.form['description'] = 'description'
|
|
resp.form['request$qs_data$element0key'] = 'foo'
|
|
resp.form['request$post_data$element0key'] = 'bar'
|
|
resp = resp.form.submit('submit').follow()
|
|
|
|
wscall = NamedWsCall.get('a_new_webservice_call')
|
|
assert wscall.request['qs_data'] == {'foo': ''}
|
|
assert wscall.request['post_data'] == {'bar': ''}
|
|
|
|
|
|
def test_wscalls_timeout(pub):
|
|
create_superuser(pub)
|
|
NamedWsCall.wipe()
|
|
app = login(get_app(pub))
|
|
|
|
resp = app.get('/backoffice/settings/wscalls/')
|
|
resp = resp.click('New webservice call')
|
|
resp.form['name'] = 'a new webservice call'
|
|
resp.form['description'] = 'description'
|
|
resp.form['request$timeout'] = 'plop'
|
|
resp = resp.form.submit('submit')
|
|
assert resp.pyquery('[data-widget-name="request$timeout"].widget-with-error')
|
|
assert (
|
|
resp.pyquery('[data-widget-name="request$timeout"] .error').text()
|
|
== 'Timeout must be empty or a number.'
|
|
)
|
|
resp.form['request$timeout'] = '10'
|
|
resp = resp.form.submit('submit')
|
|
|
|
wscall = NamedWsCall.get('a_new_webservice_call')
|
|
assert wscall.request['timeout'] == '10'
|