0001-implement-PUT-PATCH-and-DELETE-for-webservice-calls-.patch
tests/test_workflows.py | ||
---|---|---|
938 | 938 |
assert qs['evalme'] == [formdata.get_display_id()] |
939 | 939 |
assert qs['str'] == ['abcd'] |
940 | 940 | |
941 |
item = WebserviceCallStatusItem() |
|
942 |
item.method = 'DELETE' |
|
943 |
item.post = False |
|
944 |
item.url = 'http://remote.example.net/json' |
|
945 |
pub.substitutions.feed(formdata) |
|
946 |
item.perform(formdata) |
|
947 |
assert http_requests.get_last('method') == 'DELETE' |
|
948 | ||
949 |
item = WebserviceCallStatusItem() |
|
950 |
item.url = 'http://remote.example.net' |
|
951 |
item.method = 'PUT' |
|
952 |
item.post = False |
|
953 |
item.post_data = {'str': 'abcd', 'one': '=1', |
|
954 |
'evalme': '=form_number', 'error':'=1=3'} |
|
955 |
pub.substitutions.feed(formdata) |
|
956 |
item.perform(formdata) |
|
957 |
assert http_requests.get_last('url') == 'http://remote.example.net' |
|
958 |
assert http_requests.get_last('method') == 'PUT' |
|
959 |
payload = json.loads(http_requests.get_last('body')) |
|
960 |
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()} |
|
961 | ||
962 |
item = WebserviceCallStatusItem() |
|
963 |
item.url = 'http://remote.example.net' |
|
964 |
item.method = 'PATCH' |
|
965 |
item.post = False |
|
966 |
item.post_data = {'str': 'abcd', 'one': '=1', |
|
967 |
'evalme': '=form_number', 'error':'=1=3'} |
|
968 |
pub.substitutions.feed(formdata) |
|
969 |
item.perform(formdata) |
|
970 |
assert http_requests.get_last('url') == 'http://remote.example.net' |
|
971 |
assert http_requests.get_last('method') == 'PATCH' |
|
972 |
payload = json.loads(http_requests.get_last('body')) |
|
973 |
assert payload == {'one': 1, 'str': 'abcd', 'evalme': formdata.get_display_id()} |
|
974 | ||
941 | 975 |
def test_webservice_waitpoint(pub): |
942 | 976 |
item = WebserviceCallStatusItem() |
943 | 977 |
assert item.waitpoint |
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 | |
... | ... | |
106 | 108 |
wscall.request = {'method': 'POST', 'url': 'http://remote.example.net/json'} |
107 | 109 |
wscall.call() |
108 | 110 |
assert http_requests.get_last('body') is None |
111 | ||
112 | ||
113 |
def test_webservice_post_put_patch(pub): |
|
114 |
NamedWsCall.wipe() |
|
115 | ||
116 |
for method in ('POST', 'PUT', 'PATCH'): |
|
117 |
wscall = NamedWsCall() |
|
118 |
wscall.name = 'Hello world' |
|
119 |
wscall.request = {'method': 'PUT', 'post_data': {'toto': 'coin'}, |
|
120 |
'url': 'http://remote.example.net/json'} |
|
121 |
try: |
|
122 |
wscall.call() |
|
123 |
except: |
|
124 |
pass |
|
125 |
assert http_requests.get_last('url') == wscall.request['url'] |
|
126 |
assert http_requests.get_last('method') == wscall.request['method'] |
|
127 |
assert json.loads(http_requests.get_last('body')) == wscall.request['post_data'] |
|
128 | ||
129 | ||
130 |
def test_webservice_delete(pub): |
|
131 |
NamedWsCall.wipe() |
|
132 | ||
133 |
wscall = NamedWsCall() |
|
134 |
wscall.name = 'Hello world' |
|
135 |
wscall.request = {'method': 'DELETE', 'post_data': {'toto': 'coin'}, |
|
136 |
'url': 'http://remote.example.net/json'} |
|
137 |
try: |
|
138 |
wscall.call() |
|
139 |
except: |
|
140 |
pass |
|
141 |
assert http_requests.get_last('url') == wscall.request['url'] |
|
142 |
assert http_requests.get_last('method') == 'DELETE' |
wcs/wf/wscall.py | ||
---|---|---|
123 | 123 | |
124 | 124 |
@property |
125 | 125 |
def method(self): |
126 |
if self._method in ('GET', 'POST'): |
|
126 |
if self._method in ('GET', 'POST', 'PUT', 'PATCH', 'DELETE'):
|
|
127 | 127 |
return self._method |
128 | 128 |
if self.post or self.post_data: |
129 | 129 |
return 'POST' |
... | ... | |
165 | 165 |
value=self.qs_data or {}, |
166 | 166 |
element_value_type=ComputedExpressionWidget) |
167 | 167 |
methods = collections.OrderedDict( |
168 |
[('GET', _('GET')), ('POST', _('POST (JSON)'))]) |
|
168 |
[ |
|
169 |
('GET', _('GET')), |
|
170 |
('POST', _('POST (JSON)')) |
|
171 |
('PUT', _('PUT (JSON)')), |
|
172 |
('PATCH', _('PATCH (JSON)')), |
|
173 |
('DELETE', _('DELETE')), |
|
174 |
]) |
|
169 | 175 |
if 'method' in parameters: |
170 | 176 |
form.add(RadiobuttonsWidget, '%smethod' % prefix, |
171 | 177 |
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) |
... | ... | |
79 | 79 |
payload = None |
80 | 80 | |
81 | 81 |
# if post_data exists, payload is a dict built from it |
82 |
if method == 'POST' and post_data:
|
|
82 |
if method in ('PATCH', 'PUT', 'POST') and post_data:
|
|
83 | 83 |
payload = {} |
84 | 84 |
for (key, value) in post_data.items(): |
85 | 85 |
try: |
... | ... | |
89 | 89 | |
90 | 90 |
# if formdata has to be sent, it's the payload. If post_data exists, |
91 | 91 |
# it's added in formdata['extra'] |
92 |
if method == 'POST' and post_formdata:
|
|
92 |
if method in ('PATCH', 'PUT', 'POST') and post_formdata:
|
|
93 | 93 |
if formdata: |
94 | 94 |
formdata_dict = formdata.get_json_export_dict() |
95 | 95 |
if payload is not None: |
96 | 96 |
formdata_dict['extra'] = payload |
97 | 97 |
payload = formdata_dict |
98 | 98 | |
99 |
if method == 'POST':
|
|
99 |
if method in ('PATCH', 'PUT', 'POST'):
|
|
100 | 100 |
timeout = TIMEOUT |
101 | 101 |
if payload: |
102 | 102 |
headers['Content-type'] = 'application/json' |
... | ... | |
105 | 105 |
# increase timeout for huge loads, one second every 65536 |
106 | 106 |
# bytes, to match a country 512kbps DSL line. |
107 | 107 |
timeout += len(payload) / 65536 |
108 |
response, status, data, auth_header = http_post_request( |
|
109 |
url, payload, headers=headers, timeout=timeout) |
|
108 |
response, status, data, auth_header = qommon.misc._http_request( |
|
109 |
url, method=method, body=payload, headers=headers, timeout=timeout) |
|
110 |
elif method == 'DELETE': |
|
111 |
response, status, data, auth_header = qommon.misc._http_request( |
|
112 |
url, method='DELETE', headers=headers, timeout=TIMEOUT) |
|
110 | 113 |
else: |
111 | 114 |
response, status, data, auth_header = http_get_page( |
112 |
url, headers=headers, timeout=TIMEOUT)
|
|
115 |
url, headers=headers, timeout=TIMEOUT) |
|
113 | 116 |
return (response, status, data) |
114 | 117 | |
115 | 118 | |
... | ... | |
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 |
- |