0001-implement-PUT-PATCH-and-DELETE-for-webservice-calls-.patch
tests/test_workflows.py | ||
---|---|---|
910 | 910 |
assert qs['evalme'] == [formdata.get_display_id()] |
911 | 911 |
assert qs['str'] == ['abcd'] |
912 | 912 | |
913 |
item = WebserviceCallStatusItem() |
|
914 |
item.method = 'DELETE' |
|
915 |
item.post = False |
|
916 |
item.url = 'http://remote.example.net/json' |
|
917 |
pub.substitutions.feed(formdata) |
|
918 |
item.perform(formdata) |
|
919 |
assert http_requests.get_last('method') == 'DELETE' |
|
920 | ||
921 |
item = WebserviceCallStatusItem() |
|
922 |
item.url = 'http://remote.example.net' |
|
923 |
item.method = 'PUT' |
|
924 |
item.post = False |
|
925 |
item.post_data = {'str': 'abcd', 'one': '=1', |
|
926 |
'evalme': '=form_number', 'error':'=1=3'} |
|
927 |
pub.substitutions.feed(formdata) |
|
928 |
item.perform(formdata) |
|
929 |
assert http_requests.get_last('url') == 'http://remote.example.net' |
|
930 |
assert http_requests.get_last('method') == 'PUT' |
|
931 |
payload = json.loads(http_requests.get_last('body')) |
|
932 |
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()} |
|
933 | ||
934 |
item = WebserviceCallStatusItem() |
|
935 |
item.url = 'http://remote.example.net' |
|
936 |
item.method = 'PATCH' |
|
937 |
item.post = False |
|
938 |
item.post_data = {'str': 'abcd', 'one': '=1', |
|
939 |
'evalme': '=form_number', 'error':'=1=3'} |
|
940 |
pub.substitutions.feed(formdata) |
|
941 |
item.perform(formdata) |
|
942 |
assert http_requests.get_last('url') == 'http://remote.example.net' |
|
943 |
assert http_requests.get_last('method') == 'PATCH' |
|
944 |
payload = json.loads(http_requests.get_last('body')) |
|
945 |
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()} |
|
913 | 946 | |
914 | 947 |
def test_timeout(pub): |
915 | 948 |
workflow = Workflow(name='timeout') |
tests/test_wscall.py | ||
---|---|---|
1 | 1 |
import json |
2 |
import mock |
|
3 | 2 |
import pytest |
4 | 3 | |
5 | 4 |
from wcs.qommon.http_request import HTTPRequest |
... | ... | |
17 | 16 |
pub.load_site_options() |
18 | 17 |
return pub |
19 | 18 | |
19 | ||
20 | 20 |
def teardown_module(module): |
21 | 21 |
clean_temporary_pub() |
22 | 22 | |
... | ... | |
48 | 48 |
assert 'bye' in NamedWsCall.keys() |
49 | 49 |
assert not 'hello_1' in NamedWsCall.keys() |
50 | 50 | |
51 | ||
51 | 52 |
def test_webservice_substitution_variable(pub): |
52 | 53 |
NamedWsCall.wipe() |
53 | 54 | |
... | ... | |
61 | 62 |
variables = pub.substitutions.get_context_variables() |
62 | 63 |
assert variables['webservice'].hello_world == {'foo': 'bar'} |
63 | 64 | |
65 | ||
64 | 66 |
def test_webservice_auto_sign(pub): |
65 | 67 |
NamedWsCall.wipe() |
66 | 68 | |
... | ... | |
98 | 100 |
wscall.request = {'method': 'POST', 'url': 'http://remote.example.net/json'} |
99 | 101 |
wscall.call() |
100 | 102 |
assert http_requests.get_last('body') is None |
103 | ||
104 | ||
105 |
def test_webservice_post_put_patch(pub): |
|
106 |
NamedWsCall.wipe() |
|
107 | ||
108 |
for method in ('POST', 'PUT', 'PATCH'): |
|
109 |
wscall = NamedWsCall() |
|
110 |
wscall.name = 'Hello world' |
|
111 |
wscall.request = {'method': 'PUT', 'post_data': {'toto': 'coin'}, |
|
112 |
'url': 'http://remote.example.net/json'} |
|
113 |
try: |
|
114 |
wscall.call() |
|
115 |
except: |
|
116 |
pass |
|
117 |
assert http_requests.get_last('url') == wscall.request['url'] |
|
118 |
assert http_requests.get_last('method') == wscall.request['method'] |
|
119 |
assert json.loads(http_requests.get_last('body')) == wscall.request['post_data'] |
|
120 | ||
121 | ||
122 |
def test_webservice_delete(pub): |
|
123 |
NamedWsCall.wipe() |
|
124 | ||
125 |
wscall = NamedWsCall() |
|
126 |
wscall.name = 'Hello world' |
|
127 |
wscall.request = {'method': 'DELETE', 'post_data': {'toto': 'coin'}, |
|
128 |
'url': 'http://remote.example.net/json'} |
|
129 |
try: |
|
130 |
wscall.call() |
|
131 |
except: |
|
132 |
pass |
|
133 |
assert http_requests.get_last('url') == wscall.request['url'] |
|
134 |
assert http_requests.get_last('method') == 'DELETE' |
wcs/wf/wscall.py | ||
---|---|---|
114 | 114 | |
115 | 115 |
@property |
116 | 116 |
def method(self): |
117 |
if self._method in ('GET', 'POST'): |
|
117 |
if self._method in ('GET', 'POST', 'PUT', 'PATCH', 'DELETE'):
|
|
118 | 118 |
return self._method |
119 | 119 |
if self.post or self.post_data: |
120 | 120 |
return 'POST' |
... | ... | |
156 | 156 |
value=self.qs_data or {}, |
157 | 157 |
element_value_type=ComputedExpressionWidget) |
158 | 158 |
methods = collections.OrderedDict( |
159 |
[('GET', _('GET')), ('POST', _('POST (JSON)'))]) |
|
159 |
[ |
|
160 |
('GET', _('GET')), |
|
161 |
('POST', _('POST (JSON)')) |
|
162 |
('PUT', _('PUT (JSON)')), |
|
163 |
('PATCH', _('PATCH (JSON)')), |
|
164 |
('DELETE', _('DELETE')), |
|
165 |
]) |
|
160 | 166 |
if 'method' in parameters: |
161 | 167 |
form.add(RadiobuttonsWidget, '%smethod' % prefix, |
162 | 168 |
title=_('Method'), |
wcs/wscalls.py | ||
---|---|---|
23 | 23 | |
24 | 24 |
from quixote import get_publisher |
25 | 25 | |
26 |
from qommon.misc import (simplify, http_get_page, http_post_request, |
|
27 |
get_variadic_url, JSONEncoder, json_loads)
|
|
26 |
import qommon.misc |
|
27 |
from qommon.misc import (simplify, http_get_page, get_variadic_url, JSONEncoder, json_loads)
|
|
28 | 28 |
from qommon.xml_storage import XmlStorableObject |
29 | 29 |
from qommon.form import (CompositeWidget, StringWidget, WidgetDict, |
30 | 30 |
ComputedExpressionWidget, RadiobuttonsWidget, CheckboxWidget) |
... | ... | |
78 | 78 |
payload = None |
79 | 79 | |
80 | 80 |
# if post_data exists, payload is a dict built from it |
81 |
if method == 'POST' and post_data:
|
|
81 |
if method in ('PATCH', 'PUT', 'POST') and post_data:
|
|
82 | 82 |
payload = {} |
83 | 83 |
for (key, value) in post_data.items(): |
84 | 84 |
try: |
... | ... | |
88 | 88 | |
89 | 89 |
# if formdata has to be sent, it's the payload. If post_data exists, |
90 | 90 |
# it's added in formdata['extra'] |
91 |
if method == 'POST' and post_formdata:
|
|
91 |
if method in ('PATCH', 'PUT', 'POST') and post_formdata:
|
|
92 | 92 |
if formdata: |
93 | 93 |
formdata_dict = formdata.get_json_export_dict() |
94 | 94 |
if payload is not None: |
95 | 95 |
formdata_dict['extra'] = payload |
96 | 96 |
payload = formdata_dict |
97 | 97 | |
98 |
if method == 'POST':
|
|
98 |
if method in ('PATCH', 'PUT', 'POST'):
|
|
99 | 99 |
timeout = TIMEOUT |
100 | 100 |
if payload: |
101 | 101 |
payload = json.dumps(payload, cls=JSONEncoder, |
... | ... | |
103 | 103 |
# increase timeout for huge loads, one second every 65536 |
104 | 104 |
# bytes, to match a country 512kbps DSL line. |
105 | 105 |
timeout += len(payload) / 65536 |
106 |
response, status, data, auth_header = http_post_request( |
|
107 |
url, payload, headers=headers, timeout=timeout) |
|
106 |
response, status, data, auth_header = qommon.misc._http_request( |
|
107 |
url, method=method, body=payload, headers=headers, timeout=timeout) |
|
108 |
elif method == 'DELETE': |
|
109 |
response, status, data, auth_header = qommon.misc._http_request( |
|
110 |
url, method='DELETE', headers=headers, timeout=TIMEOUT) |
|
108 | 111 |
else: |
109 | 112 |
response, status, data, auth_header = http_get_page( |
110 |
url, headers=headers, timeout=TIMEOUT)
|
|
113 |
url, headers=headers, timeout=TIMEOUT) |
|
111 | 114 |
return (response, status, data) |
112 | 115 | |
113 | 116 | |
... | ... | |
216 | 219 |
value=value.get('qs_data') or {}, |
217 | 220 |
element_value_type=ComputedExpressionWidget) |
218 | 221 |
methods = collections.OrderedDict( |
219 |
[('GET', _('GET')), ('POST', _('POST (JSON)'))]) |
|
222 |
[ |
|
223 |
('GET', _('GET')), |
|
224 |
('POST', _('POST (JSON)')) |
|
225 |
('PUT', _('PUT (JSON)')), |
|
226 |
('PATCH', _('PATCH (JSON)')), |
|
227 |
('DELETE', _('DELETE')), |
|
228 |
]) |
|
220 | 229 |
self.add(RadiobuttonsWidget, 'method', |
221 | 230 |
title=_('Method'), |
222 | 231 |
options=methods.items(), |
223 |
- |