tests: add basic testing of wscall workflow
This commit is contained in:
parent
dc65fd5b7b
commit
5d14ce6f0a
|
@ -14,8 +14,10 @@ from wcs.wf.jump import JumpWorkflowStatusItem
|
|||
from wcs.wf.register_comment import RegisterCommenterWorkflowStatusItem
|
||||
from wcs.wf.remove import RemoveWorkflowStatusItem
|
||||
from wcs.wf.roles import AddRoleWorkflowStatusItem, RemoveRoleWorkflowStatusItem
|
||||
from wcs.wf.wscall import WebserviceCallStatusItem
|
||||
|
||||
from utilities import create_temporary_pub, MockSubstitutionVariables, emails
|
||||
from utilities import (create_temporary_pub, MockSubstitutionVariables, emails,
|
||||
http_requests)
|
||||
|
||||
def setup_module(module):
|
||||
cleanup()
|
||||
|
@ -325,3 +327,50 @@ def test_email():
|
|||
item.perform(formdata)
|
||||
assert emails.count() == 1
|
||||
assert emails.get('foobar')['kwargs']['email_rcpt'] == ['xyz@localhost']
|
||||
|
||||
def test_webservice_call():
|
||||
pub.substitutions.feed(MockSubstitutionVariables())
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'baz'
|
||||
formdef.fields = []
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
item = WebserviceCallStatusItem()
|
||||
item.url = 'http://remote.example.net'
|
||||
item.perform(formdata)
|
||||
assert http_requests.get_last('url') == 'http://remote.example.net'
|
||||
assert http_requests.get_last('method') == 'POST'
|
||||
|
||||
item = WebserviceCallStatusItem()
|
||||
item.url = 'http://remote.example.net'
|
||||
item.post = False
|
||||
item.perform(formdata)
|
||||
assert http_requests.get_last('url') == 'http://remote.example.net'
|
||||
assert http_requests.get_last('method') == 'GET'
|
||||
|
||||
item = WebserviceCallStatusItem()
|
||||
item.url = 'http://remote.example.net'
|
||||
item.post = False
|
||||
item.request_signature_key = 'xxx'
|
||||
item.perform(formdata)
|
||||
assert 'signature=' in http_requests.get_last('url')
|
||||
assert http_requests.get_last('method') == 'GET'
|
||||
|
||||
item = WebserviceCallStatusItem()
|
||||
item.url = 'http://remote.example.net'
|
||||
item.post = False
|
||||
item.request_signature_key = '[empty]'
|
||||
item.perform(formdata)
|
||||
assert not 'signature=' in http_requests.get_last('url')
|
||||
|
||||
item = WebserviceCallStatusItem()
|
||||
item.url = 'http://remote.example.net'
|
||||
item.post = False
|
||||
item.request_signature_key = '[bar]'
|
||||
item.perform(formdata)
|
||||
assert 'signature=' in http_requests.get_last('url')
|
||||
|
|
|
@ -58,4 +58,35 @@ emails = EmailsMocking()
|
|||
|
||||
class MockSubstitutionVariables(object):
|
||||
def get_substitution_variables(self):
|
||||
return {'bar': 'Foobar', 'foo': '1 < 3', 'email': 'sub@localhost'}
|
||||
return {'bar': 'Foobar', 'foo': '1 < 3', 'email': 'sub@localhost',
|
||||
'empty': ''}
|
||||
|
||||
|
||||
class HttpRequestsMocking(object):
|
||||
def __init__(self):
|
||||
self.requests = []
|
||||
import wcs.qommon.misc
|
||||
import qommon.misc
|
||||
wcs.qommon.misc._http_request = self.http_request
|
||||
qommon.misc._http_request = self.http_request
|
||||
|
||||
def http_request(self, url, method='GET', body=None, headers={}, timeout=None):
|
||||
self.requests.append(
|
||||
{'url': url,
|
||||
'method': method,
|
||||
'body': body,
|
||||
'headers': headers,
|
||||
'timeout': timeout})
|
||||
|
||||
response = ''
|
||||
status = 200
|
||||
data = ''
|
||||
return None, status, data, None
|
||||
|
||||
def get_last(self, attribute):
|
||||
return self.requests[-1][attribute]
|
||||
|
||||
def empty(self):
|
||||
self.requests = []
|
||||
|
||||
http_requests = HttpRequestsMocking()
|
||||
|
|
Loading…
Reference in New Issue