168 lines
4.8 KiB
Python
168 lines
4.8 KiB
Python
import json
|
|
import pytest
|
|
|
|
from wcs.qommon.http_request import HTTPRequest
|
|
from wcs.qommon.template import Template
|
|
from wcs.wscalls import NamedWsCall
|
|
|
|
from utilities import create_temporary_pub, clean_temporary_pub
|
|
|
|
|
|
@pytest.fixture
|
|
def pub():
|
|
pub = create_temporary_pub()
|
|
req = HTTPRequest(None, {'SCRIPT_NAME': '/', 'SERVER_NAME': 'example.net'})
|
|
pub.set_app_dir(req)
|
|
pub._set_request(req)
|
|
pub.load_site_options()
|
|
return pub
|
|
|
|
|
|
def teardown_module(module):
|
|
clean_temporary_pub()
|
|
|
|
|
|
def test_named_wscall(pub):
|
|
# create object
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello'
|
|
wscall.request = {'url': 'http://example.net', 'qs_data': {'a': 'b'}}
|
|
wscall.store()
|
|
assert wscall.slug == 'hello'
|
|
|
|
# get object
|
|
wscall = NamedWsCall.get('hello')
|
|
assert wscall.name == 'Hello'
|
|
assert wscall.request.get('url') == 'http://example.net'
|
|
assert wscall.request.get('qs_data') == {'a': 'b'}
|
|
|
|
# create with same name, should get a different slug
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello'
|
|
wscall.request = {'url': 'http://example.net'}
|
|
wscall.store()
|
|
assert wscall.slug == 'hello_1'
|
|
|
|
# change slug, shoulg get a new id
|
|
wscall.slug = 'bye'
|
|
wscall.store()
|
|
assert 'bye' in NamedWsCall.keys()
|
|
assert not 'hello_1' in NamedWsCall.keys()
|
|
|
|
|
|
def test_webservice_substitution_variable(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'url': 'http://remote.example.net/json'}
|
|
wscall.store()
|
|
assert wscall.slug == 'hello_world'
|
|
|
|
variables = pub.substitutions.get_context_variables()
|
|
assert variables['webservice'].hello_world == {'foo': 'bar'}
|
|
|
|
|
|
def test_webservice_auto_sign(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'url': 'http://blah.example.net'}
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert not 'signature=' in http_requests.get_last('url')
|
|
|
|
wscall.request = {'url': 'http://idp.example.net'}
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert 'orig=example.net' in http_requests.get_last('url')
|
|
assert 'signature=' in http_requests.get_last('url')
|
|
|
|
# erroneous space
|
|
wscall.request = {'url': ' http://idp.example.net'}
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert 'orig=example.net' in http_requests.get_last('url')
|
|
assert 'signature=' in http_requests.get_last('url')
|
|
|
|
wscall.request['request_signature_key'] = 'blah'
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert not 'orig=example.net' in http_requests.get_last('url')
|
|
assert 'signature=' in http_requests.get_last('url')
|
|
|
|
|
|
def test_webservice_post_with_no_payload(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'method': 'POST', 'url': 'http://remote.example.net/json'}
|
|
wscall.call()
|
|
assert http_requests.get_last('body') is None
|
|
|
|
|
|
def test_wscall_ezt(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'url': 'http://remote.example.net/json'}
|
|
wscall.store()
|
|
assert wscall.slug == 'hello_world'
|
|
|
|
variables = pub.substitutions.get_context_variables()
|
|
|
|
template = Template('<p>{{ webservice.hello_world.foo }}</p>')
|
|
assert template.render(variables) == '<p>bar</p>'
|
|
|
|
template = Template('<p>[webservice.hello_world.foo]</p>')
|
|
assert template.render(variables) == '<p>bar</p>'
|
|
|
|
# undefined webservice
|
|
template = Template('<p>{{ webservice.hello.foo }}</p>')
|
|
assert template.render(variables) == '<p></p>'
|
|
template = Template('<p>[webservice.hello.foo]</p>')
|
|
assert template.render(variables) == '<p>[webservice.hello.foo]</p>'
|
|
|
|
|
|
def test_webservice_post_put_patch(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
for method in ('POST', 'PUT', 'PATCH'):
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'method': method, 'post_data': {'toto': 'coin'},
|
|
'url': 'http://remote.example.net/json'}
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert http_requests.get_last('url') == wscall.request['url']
|
|
assert http_requests.get_last('method') == wscall.request['method']
|
|
assert json.loads(http_requests.get_last('body')) == wscall.request['post_data']
|
|
|
|
|
|
def test_webservice_delete(http_requests, pub):
|
|
NamedWsCall.wipe()
|
|
|
|
wscall = NamedWsCall()
|
|
wscall.name = 'Hello world'
|
|
wscall.request = {'method': 'DELETE', 'post_data': {'toto': 'coin'},
|
|
'url': 'http://remote.example.net/json'}
|
|
try:
|
|
wscall.call()
|
|
except:
|
|
pass
|
|
assert http_requests.get_last('url') == wscall.request['url']
|
|
assert http_requests.get_last('method') == 'DELETE'
|