mock des requêtes dans les tests (#85086) #1067
|
|
@ -12,8 +12,9 @@ from wcs.carddef import CardDef
|
|||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.upload_storage import PicklableUpload
|
||||
from wcs.testdef import TestDef, TestResult
|
||||
from wcs.testdef import TestDef, TestResult, WebserviceResponse
|
||||
from wcs.workflow_tests import WorkflowTests
|
||||
from wcs.wscalls import NamedWsCall
|
||||
|
||||
from ..utilities import clean_temporary_pub, create_temporary_pub, get_app, login
|
||||
from .test_all import create_superuser
|
||||
|
|
@ -32,6 +33,7 @@ def pub():
|
|||
TestDef.wipe()
|
||||
TestResult.wipe()
|
||||
WorkflowTests.wipe()
|
||||
WebserviceResponse.wipe()
|
||||
return pub
|
||||
|
||||
|
||||
|
|
@ -188,6 +190,11 @@ def test_tests_import_export(pub):
|
|||
testdef.name = 'First test'
|
||||
testdef.store()
|
||||
|
||||
response = WebserviceResponse()
|
||||
response.testdef_id = testdef.id
|
||||
response.name = 'Response xxx'
|
||||
response.store()
|
||||
|
||||
app = login(get_app(pub))
|
||||
|
||||
resp = app.get('/backoffice/forms/1/tests/%s/' % testdef.id)
|
||||
|
|
@ -198,6 +205,7 @@ def test_tests_import_export(pub):
|
|||
resp = resp.form.submit().follow()
|
||||
assert 'First test' not in resp.text
|
||||
assert WorkflowTests.count() == 0
|
||||
assert WebserviceResponse.count() == 0
|
||||
|
||||
resp = resp.click('Import')
|
||||
resp.form['file'] = Upload('export.wcs', export_resp.body)
|
||||
|
|
@ -826,6 +834,106 @@ def test_tests_result_recorded_errors(pub):
|
|||
assert escape('Invalid filter "unknown"') in resp.text
|
||||
|
||||
|
||||
def test_tests_result_sent_requests(pub, http_requests):
|
||||
create_superuser(pub)
|
||||
|
||||
wscall = NamedWsCall()
|
||||
wscall.name = 'Hello world'
|
||||
wscall.request = {'url': 'http://remote.example.net/json'}
|
||||
wscall.store()
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test title'
|
||||
formdef.fields = [
|
||||
fields.PageField(
|
||||
id='0',
|
||||
label='1st page',
|
||||
post_conditions=[
|
||||
{
|
||||
'condition': {'type': 'django', 'value': 'form_var_computed_foo == "bar"'},
|
||||
'error_message': '',
|
||||
}
|
||||
],
|
||||
),
|
||||
fields.ComputedField(
|
||||
id='1',
|
||||
label='Computed',
|
||||
varname='computed',
|
||||
value_template='{{ webservice.hello_world }}',
|
||||
freeze_on_initial_value=True,
|
||||
),
|
||||
]
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
|
|
||||
formdata.store()
|
||||
|
||||
testdef = TestDef.create_from_formdata(formdef, formdata)
|
||||
testdef.name = 'First test'
|
||||
testdef.store()
|
||||
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/forms/1/tests/results/run').follow()
|
||||
|
||||
assert 'Success!' in resp.text
|
||||
assert http_requests.count() == 1
|
||||
http_requests.empty()
|
||||
|
||||
resp = resp.click('Display details')
|
||||
|
||||
assert 'Sent requests:' in resp.text
|
||||
assert 'GET http://remote.example.net/json' in resp.text
|
||||
assert 'Used webservice response:' not in resp.text
|
||||
|
||||
response = WebserviceResponse()
|
||||
response.testdef_id = testdef.id
|
||||
response.name = 'Response xxx'
|
||||
response.url = 'http://remote.example.net/json'
|
||||
response.payload = '{"foo": "wrong"}'
|
||||
response.store()
|
||||
|
||||
resp = app.get('/backoffice/forms/1/tests/results/run').follow()
|
||||
|
||||
assert 'Success!' not in resp.text
|
||||
assert http_requests.count() == 0
|
||||
|
||||
resp = resp.click('Display details')
|
||||
result_url = resp.request.url
|
||||
|
||||
assert 'Sent requests:' in resp.text
|
||||
assert 'GET http://remote.example.net/json' in resp.text
|
||||
assert 'Used webservice response:' in resp.text
|
||||
|
||||
resp = resp.click('Response xxx')
|
||||
assert 'Edit webservice response' in resp.text
|
||||
|
||||
response.remove_self()
|
||||
resp = app.get(result_url)
|
||||
|
||||
assert 'Used webservice response:' in resp.text
|
||||
assert 'Response xxx' not in resp.text
|
||||
assert 'deleted' in resp.text
|
||||
|
||||
wscall.request['method'] = 'POST'
|
||||
wscall.store()
|
||||
|
||||
resp = app.get('/backoffice/forms/1/tests/results/run').follow()
|
||||
|
||||
assert 'Success!' not in resp.text
|
||||
assert http_requests.count() == 0
|
||||
|
||||
resp = resp.click('Display details')
|
||||
|
||||
assert 'Sent requests:' in resp.text
|
||||
assert 'POST http://remote.example.net/json' in resp.text
|
||||
assert 'Request was blocked since it is not a GET request.' in resp.text
|
||||
assert 'Recorded errors:' not in resp.text
|
||||
|
||||
resp = resp.click('You can create corresponding webservice response here.')
|
||||
assert 'Webservice responses' in resp.text
|
||||
|
||||
|
||||
def test_tests_run_order(pub):
|
||||
create_superuser(pub)
|
||||
|
||||
|
|
@ -1022,3 +1130,71 @@ def test_tests_exclude_self(pub):
|
|||
|
||||
resp = resp.form.submit('submit').follow()
|
||||
assert 'First test' in resp.text
|
||||
|
||||
|
||||
def test_tests_webservice_response(pub):
|
||||
create_superuser(pub)
|
||||
|
||||
formdef = FormDef()
|
||||
formdef.name = 'test title'
|
||||
formdef.store()
|
||||
|
||||
formdata = formdef.data_class()()
|
||||
formdata.just_created()
|
||||
formdata.store()
|
||||
|
||||
testdef = TestDef.create_from_formdata(formdef, formdata)
|
||||
testdef.name = 'First test'
|
||||
testdef.store()
|
||||
|
||||
app = login(get_app(pub))
|
||||
resp = app.get('/backoffice/forms/1/tests/%s/' % testdef.id)
|
||||
|
||||
resp = resp.click('Webservice response')
|
||||
assert 'There are no webservice responses yet.' in resp.text
|
||||
|
||||
resp = resp.click('New')
|
||||
resp.form['name'] = 'Test response'
|
||||
resp = resp.form.submit().follow()
|
||||
|
||||
resp.form['url'] = 'http://example.com/'
|
||||
resp = resp.form.submit().follow()
|
||||
|
||||
assert 'There are no webservice responses yet.' not in resp.text
|
||||
|
||||
resp = resp.click('Test response')
|
||||
resp.form['payload'] = '{"a": "b"}'
|
||||
resp = resp.form.submit().follow()
|
||||
|
||||
response = testdef.get_webservice_responses()[0]
|
||||
assert response.name == 'Test response'
|
||||
assert response.url == 'http://example.com/'
|
||||
assert response.payload == '{"a": "b"}'
|
||||
|
||||
resp = resp.click('Duplicate').follow()
|
||||
assert 'Test response' in resp.text
|
||||
assert 'not configured' not in resp.text
|
||||
assert 'Test response (copy)' in resp.text
|
||||
|
||||
response = testdef.get_webservice_responses()[1]
|
||||
assert response.name == 'Test response (copy)'
|
||||
assert response.url == 'http://example.com/'
|
||||
assert response.payload == '{"a": "b"}'
|
||||
|
||||
resp = resp.click('Remove', href=response.id)
|
||||
resp = resp.form.submit().follow()
|
||||
|
||||
assert 'Test response (copy)' not in resp.text
|
||||
|
||||
resp = resp.click('Test response')
|
||||
resp.form['payload'] = ''
|
||||
resp = resp.form.submit().follow()
|
||||
|
||||
assert 'Test response' in resp.text
|
||||
assert '(not configured)' in resp.text
|
||||
|
||||
resp = resp.click('Test response')
|
||||
resp.form['payload'] = '{"a"}'
|
||||
resp = resp.form.submit()
|
||||
|
||||
assert "Invalid JSON: Expecting ':' delimiter: line 1 column 5 (char 4)" in resp.text
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import io
|
|||
import json
|
||||
import time
|
||||
import xml.etree.ElementTree as ET
|
||||
from unittest import mock
|
||||
|
||||
import pytest
|
||||
import responses
|
||||
|
|
@ -15,7 +16,7 @@ from wcs.carddef import CardDef
|
|||
from wcs.formdef import FormDef
|
||||
from wcs.qommon.http_request import HTTPRequest
|
||||
from wcs.qommon.upload_storage import PicklableUpload
|
||||
from wcs.testdef import TestDef, TestDefXmlProxy, TestError, TestResult
|
||||
from wcs.testdef import TestDef, TestDefXmlProxy, TestError, TestResult, WebserviceResponse
|
||||
from wcs.wscalls import NamedWsCall
|
||||
|
||||
from .utilities import clean_temporary_pub, create_temporary_pub
|
||||
|
|
@ -31,6 +32,7 @@ def pub():
|
|||
|
||||
FormDef.wipe()
|
||||
BlockDef.wipe()
|
||||
WebserviceResponse.wipe()
|
||||
return pub
|
||||
|
||||
|
||||
|
|
@ -1038,8 +1040,62 @@ def test_computed_field_support_webservice(pub, http_requests):
|
|||
formdata.receipt_time = make_aware(datetime.datetime(2021, 1, 1, 0, 0))
|
||||
|
||||
testdef = TestDef.create_from_formdata(formdef, formdata)
|
||||
testdef.store()
|
||||
testdef.run(formdef)
|
||||
|
||||
assert len(testdef.sent_requests) == 1
|
||||
assert testdef.sent_requests[0]['method'] == 'GET'
|
||||
assert testdef.sent_requests[0]['url'] == 'http://remote.example.net/json'
|
||||
|
||||
response = WebserviceResponse()
|
||||
response.testdef_id = testdef.id
|
||||
response.name = 'Fake response'
|
||||
response.url = 'http://remote.example.net/json'
|
||||
response.payload = '{"foo": "bar"}'
|
||||
response.store()
|
||||
|
||||
testdef.run(formdef)
|
||||
|
||||
assert len(testdef.sent_requests) == 1
|
||||
assert testdef.sent_requests[0]['url'] == 'http://remote.example.net/json'
|
||||
assert testdef.sent_requests[0]['webservice_response_id'] == response.id
|
||||
|
||||
response.payload = '{"foo": "baz"}'
|
||||
response.store()
|
||||
|
||||
with pytest.raises(TestError) as excinfo:
|
||||
testdef.run(formdef)
|
||||
assert str(excinfo.value) == 'Page 1 post condition was not met (form_var_computed_foo == "bar").'
|
||||
assert len(testdef.sent_requests) == 1
|
||||
assert testdef.sent_requests[0]['url'] == 'http://remote.example.net/json'
|
||||
assert testdef.sent_requests[0]['webservice_response_id'] == response.id
|
||||
|
||||
response.url = 'http://example.com/json'
|
||||
response.store()
|
||||
|
||||
testdef.run(formdef)
|
||||
assert len(testdef.sent_requests) == 1
|
||||
assert testdef.sent_requests[0]['url'] == 'http://remote.example.net/json'
|
||||
assert testdef.sent_requests[0]['webservice_response_id'] is None
|
||||
|
||||
response.url = None
|
||||
response.store()
|
||||
|
||||
testdef.run(formdef)
|
||||
assert len(testdef.sent_requests) == 1
|
||||
assert testdef.sent_requests[0]['url'] == 'http://remote.example.net/json'
|
||||
assert testdef.sent_requests[0]['webservice_response_id'] is None
|
||||
|
||||
testdef = TestDef.create_from_formdata(formdef, formdata)
|
||||
with mock.patch('wcs.testdef.MockWebserviceResponseAdapter._send', side_effect=KeyError('missing key')):
|
||||
with pytest.raises(TestError):
|
||||
testdef.run(formdef)
|
||||
|
||||
assert len(testdef.sent_requests) == 0
|
||||
assert testdef.recorded_errors == [
|
||||
"Unexpected error when mocking webservice call for url http://remote.example.net/json: 'missing key'."
|
||||
]
|
||||
|
||||
|
||||
def test_computed_field_value_too_long(pub):
|
||||
formdef = FormDef()
|
||||
|
|
|
|||
|
|
@ -14,6 +14,9 @@
|
|||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import copy
|
||||
import json
|
||||
|
||||
from django.template.loader import render_to_string
|
||||
from django.utils.timezone import now
|
||||
from quixote import get_publisher, get_request, get_response, get_session, redirect
|
||||
|
|
@ -27,9 +30,9 @@ from wcs.forms.common import FormStatusPage
|
|||
from wcs.qommon import _, misc, template
|
||||
from wcs.qommon.afterjobs import AfterJob
|
||||
from wcs.qommon.errors import TraversalError
|
||||
from wcs.qommon.form import FileWidget, Form, RadiobuttonsWidget, SingleSelectWidget, StringWidget
|
||||
from wcs.qommon.form import FileWidget, Form, RadiobuttonsWidget, SingleSelectWidget, StringWidget, TextWidget
|
||||
from wcs.sql_criterias import Equal, Null, StrictNotEqual
|
||||
from wcs.testdef import TestDef, TestError, TestResult
|
||||
from wcs.testdef import TestDef, TestError, TestResult, WebserviceResponse
|
||||
from wcs.workflow_tests import WorkflowTestError
|
||||
|
||||
|
||||
|
|
@ -132,6 +135,7 @@ class TestPage(FormBackOfficeStatusPage):
|
|||
('edit-data', 'edit_data'),
|
||||
'duplicate',
|
||||
('workflow', 'workflow_tests'),
|
||||
('webservice-responses', 'webservice_responses'),
|
||||
]
|
||||
|
||||
def __init__(self, component, objectdef):
|
||||
|
|
@ -144,6 +148,7 @@ class TestPage(FormBackOfficeStatusPage):
|
|||
super().__init__(objectdef, filled)
|
||||
|
||||
self.workflow_tests = WorkflowTestsDirectory(self.testdef, self.formdef)
|
||||
self.webservice_responses = WebserviceResponseDirectory(self.testdef)
|
||||
|
||||
@property
|
||||
def edit_data(self):
|
||||
|
|
@ -427,6 +432,11 @@ class TestResultDetailPage(Directory):
|
|||
except (KeyError, ValueError):
|
||||
raise TraversalError()
|
||||
|
||||
try:
|
||||
self.testdef = TestDef.get(self.result['id'])
|
||||
except KeyError:
|
||||
self.testdef = None
|
||||
|
||||
def _q_traverse(self, path):
|
||||
get_response().breadcrumb.append(
|
||||
(str(self.result_index) + '/', _('Details of %(test_name)s') % {'test_name': self.result['name']})
|
||||
|
|
@ -437,28 +447,35 @@ class TestResultDetailPage(Directory):
|
|||
context = {
|
||||
'result': self.result['details'],
|
||||
'test_name': self.result['name'],
|
||||
'testdef': self.testdef,
|
||||
'workflow_test_action': self.get_workflow_test_action(
|
||||
self.result['details']['workflow_test_action_uuid']
|
||||
),
|
||||
}
|
||||
|
||||
for request in self.result['details'].get('sent_requests', []):
|
||||
if request['webservice_response_id']:
|
||||
try:
|
||||
request['webservice_response'] = [
|
||||
x
|
||||
for x in self.testdef.get_webservice_responses()
|
||||
if x.id == request['webservice_response_id']
|
||||
][0]
|
||||
except IndexError:
|
||||
pass
|
||||
|
||||
return render_to_string('wcs/backoffice/test-result-detail.html', context=context)
|
||||
|
||||
def get_workflow_test_action(self, action_uuid):
|
||||
if not action_uuid:
|
||||
if not action_uuid or not self.testdef:
|
||||
return
|
||||
|
||||
try:
|
||||
testdef = TestDef.get(self.result['id'])
|
||||
except KeyError:
|
||||
return
|
||||
|
||||
try:
|
||||
action = [x for x in testdef.workflow_tests.actions if x.uuid == action_uuid][0]
|
||||
action = [x for x in self.testdef.workflow_tests.actions if x.uuid == action_uuid][0]
|
||||
except IndexError:
|
||||
return
|
||||
|
||||
action.url = testdef.get_admin_url() + 'workflow/#%s' % action.id
|
||||
action.url = self.testdef.get_admin_url() + 'workflow/#%s' % action.id
|
||||
return action
|
||||
|
||||
|
||||
|
|
@ -593,6 +610,7 @@ class TestsAfterJob(AfterJob):
|
|||
'details': {
|
||||
'recorded_errors': test.recorded_errors,
|
||||
'missing_required_fields': test.missing_required_fields,
|
||||
'sent_requests': test.sent_requests,
|
||||
'workflow_test_action_uuid': test.exception.action_uuid if test.exception else None,
|
||||
'error_details': test.exception.details if test.exception else None,
|
||||
},
|
||||
|
|
@ -603,3 +621,135 @@ class TestsAfterJob(AfterJob):
|
|||
test_result.store()
|
||||
|
||||
return test_result
|
||||
|
||||
|
||||
class WebserviceResponsePage(Directory):
|
||||
_q_exports = ['', 'delete', 'duplicate']
|
||||
|
||||
def __init__(self, component, testdef):
|
||||
self.testdef = testdef
|
||||
try:
|
||||
self.webservice_response = [x for x in testdef.get_webservice_responses() if x.id == component][0]
|
||||
except IndexError:
|
||||
raise TraversalError()
|
||||
|
||||
def _q_index(self):
|
||||
form = Form(enctype='multipart/form-data')
|
||||
form.add(StringWidget, 'name', size=50, title=_('Name'), value=self.webservice_response.name)
|
||||
form.add(
|
||||
StringWidget,
|
||||
'url',
|
||||
title=_('URL'),
|
||||
value=self.webservice_response.url,
|
||||
size=80,
|
||||
)
|
||||
|
||||
def validate_json(value):
|
||||
try:
|
||||
json.loads(value)
|
||||
except ValueError as e:
|
||||
raise ValueError(_('Invalid JSON: %s') % e)
|
||||
|
||||
form.add(
|
||||
TextWidget,
|
||||
'payload',
|
||||
title=_('Response payload (JSON)'),
|
||||
value=self.webservice_response.payload,
|
||||
validation_function=validate_json,
|
||||
)
|
||||
|
||||
form.add_submit('submit', _('Submit'))
|
||||
form.add_submit('cancel', _('Cancel'))
|
||||
|
||||
if form.get_widget('cancel').parse():
|
||||
return redirect('.')
|
||||
|
||||
if not form.is_submitted() or form.has_errors():
|
||||
get_response().breadcrumb.append(('edit', _('Edit webservice response')))
|
||||
r = TemplateIO(html=True)
|
||||
r += htmltext('<h2>%s</h2>') % (_('Edit webservice response'))
|
||||
r += form.render()
|
||||
return r.getvalue()
|
||||
|
||||
self.webservice_response.name = form.get_widget('name').parse()
|
||||
self.webservice_response.payload = form.get_widget('payload').parse()
|
||||
self.webservice_response.url = form.get_widget('url').parse()
|
||||
self.webservice_response.store()
|
||||
|
||||
return redirect('..')
|
||||
|
||||
def delete(self):
|
||||
form = Form(enctype='multipart/form-data')
|
||||
form.add_submit('delete', _('Delete'))
|
||||
form.add_submit('cancel', _('Cancel'))
|
||||
|
||||
if form.get_widget('cancel').parse():
|
||||
return redirect('.')
|
||||
if not form.is_submitted() or form.has_errors():
|
||||
get_response().breadcrumb.append(('delete', _('Delete')))
|
||||
r = TemplateIO(html=True)
|
||||
r += htmltext('<h2>%s %s</h2>') % (_('Deleting:'), self.webservice_response)
|
||||
r += form.render()
|
||||
return r.getvalue()
|
||||
|
||||
self.webservice_response.remove_self()
|
||||
return redirect('..')
|
||||
|
||||
def duplicate(self):
|
||||
new_webservice_response = copy.deepcopy(self.webservice_response)
|
||||
new_webservice_response.id = None
|
||||
new_webservice_response.name = '%s %s' % (new_webservice_response.name, _('(copy)'))
|
||||
new_webservice_response.store()
|
||||
return redirect('..')
|
||||
|
||||
|
||||
class WebserviceResponseDirectory(Directory):
|
||||
_q_exports = ['', 'new']
|
||||
|
||||
def __init__(self, testdef):
|
||||
self.testdef = testdef
|
||||
|
||||
def _q_traverse(self, path):
|
||||
get_response().breadcrumb.append(('webservice-responses/', _('Webservice responses')))
|
||||
return super()._q_traverse(path)
|
||||
|
||||
def _q_lookup(self, component):
|
||||
return WebserviceResponsePage(component, self.testdef)
|
||||
|
||||
def _q_index(self):
|
||||
context = {
|
||||
'webservice_responses': self.testdef.get_webservice_responses(),
|
||||
'has_sidebar': True,
|
||||
}
|
||||
get_response().add_javascript(['popup.js'])
|
||||
get_response().set_title(_('Webservice responses'))
|
||||
return template.QommonTemplateResponse(
|
||||
templates=['wcs/backoffice/test-webservice-responses.html'],
|
||||
context=context,
|
||||
is_django_native=True,
|
||||
)
|
||||
|
||||
def new(self):
|
||||
form = Form(enctype='multipart/form-data')
|
||||
form.add(StringWidget, 'name', title=_('Name'), required=True, size=50)
|
||||
|
||||
form.add_submit('submit', _('Submit'))
|
||||
form.add_submit('cancel', _('Cancel'))
|
||||
|
||||
if form.get_widget('cancel').parse():
|
||||
return redirect('.')
|
||||
|
||||
if not form.is_submitted() or form.has_errors():
|
||||
get_response().breadcrumb.append(('new', _('New')))
|
||||
get_response().set_title(_('New webservice response'))
|
||||
r = TemplateIO(html=True)
|
||||
r += htmltext('<h2>%s</h2>') % _('New webservice response')
|
||||
r += form.render()
|
||||
return r.getvalue()
|
||||
|
||||
webservice_response = WebserviceResponse()
|
||||
webservice_response.testdef_id = self.testdef.id
|
||||
webservice_response.name = form.get_widget('name').parse()
|
||||
webservice_response.store()
|
||||
|
||||
return redirect(self.testdef.get_admin_url() + 'webservice-responses/%s/' % webservice_response.id)
|
||||
|
|
|
|||
|
|
@ -3124,3 +3124,10 @@ div.infonotice.columns-default-value-message {
|
|||
ul.biglist li.workflow-test-action:target {
|
||||
border-left: solid;
|
||||
}
|
||||
|
||||
ul.objects-list.single-links li a.link-action-icon.duplicate {
|
||||
margin-right: 3em;
|
||||
&::before {
|
||||
content: "\f24d"; /* clone */
|
||||
fpeters
commented
Pour duplicate on a une icône hors fontawesome, mais ça n'est pas important, on verra plus tard. Pour duplicate on a une icône hors fontawesome, mais ça n'est pas important, on verra plus tard.
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,6 +34,37 @@
|
|||
{% trans "Missing required fields:" %} {{ result.missing_required_fields|join:"," }}
|
||||
</li>
|
||||
{% endif %}
|
||||
{% if result.sent_requests %}
|
||||
<li>{% trans "Sent requests:" %}</li>
|
||||
<ul>
|
||||
{% for request in result.sent_requests %}
|
||||
<li>
|
||||
{{ request.method }} {{ request.url }}
|
||||
<ul>
|
||||
{% if request.webservice_response_id %}
|
||||
<li>
|
||||
{% trans "Used webservice response:" %}
|
||||
{% if request.webservice_response %}
|
||||
<a href="{{ testdef.get_admin_url }}webservice-responses/{{ request.webservice_response.id }}/">
|
||||
{{ request.webservice_response.name }}
|
||||
</a>
|
||||
{% else %}
|
||||
{% trans "deleted" %}
|
||||
{% endif %}
|
||||
</li>
|
||||
{% elif request.forbidden_method %}
|
||||
<li>
|
||||
{% trans "Request was blocked since it is not a GET request." %}
|
||||
<a href="{{ testdef.get_admin_url }}webservice-responses/">
|
||||
{% trans "You can create corresponding webservice response here." %}
|
||||
</a>
|
||||
</li>
|
||||
{% endif %}
|
||||
</ul>
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
{% endif %}
|
||||
</ul>
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,32 @@
|
|||
{% extends "wcs/backoffice.html" %}
|
||||
{% load i18n %}
|
||||
|
||||
{% block appbar-title %}{% trans "Webservice responses" %}{% endblock %}
|
||||
|
||||
{% block sidebar-content %}
|
||||
<h3>{% trans "Actions" %}</h3>
|
||||
<a class="button button-paragraph" href="new" rel="popup">{% trans "New" %}</a>
|
||||
{% endblock %}
|
||||
|
||||
{% block body %}
|
||||
<div class="section">
|
||||
{% if webservice_responses %}
|
||||
<ul class="objects-list single-links">
|
||||
{% for response in webservice_responses %}
|
||||
<li>
|
||||
<a href="{{ response.id }}/">
|
||||
{{ response }}
|
||||
{% if not response.is_configured %}
|
||||
<i>({% trans "not configured" %})</i>
|
||||
{% endif %}
|
||||
</a>
|
||||
<a rel="popup" class="delete" href="{{ response.id }}/delete">{% trans "Remove" %}</a>
|
||||
<a class="link-action-icon duplicate" href="{{ response.id }}/duplicate">{% trans "Duplicate" %}</a>
|
||||
</li>
|
||||
{% endfor %}
|
||||
</ul>
|
||||
{% else %}
|
||||
<div><p>{% trans "There are no webservice responses yet." %}<p></div>
|
||||
{% endif %}
|
||||
</div>
|
||||
{% endblock %}
|
||||
|
|
@ -10,5 +10,6 @@
|
|||
<h3>{% trans "Navigation" %}</h3>
|
||||
<ul class="sidebar--buttons">
|
||||
<li><a class="button button-paragraph" rel="popup" href="edit">{% trans "Options" %}</a></li>
|
||||
<li><a class="button button-paragraph" href="webservice-responses/">{% trans "Webservice responses" %}</a></li>
|
||||
<li><a class="button button-paragraph" href="inspect">{% trans "Inspect" %}</a></li>
|
||||
</ul>
|
||||
|
|
|
|||
113
wcs/testdef.py
113
wcs/testdef.py
|
|
@ -14,13 +14,17 @@
|
|||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, see <http://www.gnu.org/licenses/>.
|
||||
|
||||
import http
|
||||
import io
|
||||
import json
|
||||
import socket
|
||||
import xml.etree.ElementTree as ET
|
||||
from contextlib import contextmanager
|
||||
|
||||
import requests
|
||||
from django.core.handlers.wsgi import WSGIRequest
|
||||
from quixote import get_publisher, get_session_manager
|
||||
from urllib3 import HTTPResponse
|
||||
|
||||
from wcs import sql
|
||||
from wcs.compat import CompatHTTPRequest
|
||||
|
|
@ -130,6 +134,9 @@ class TestDef(sql.TestDef):
|
|||
def workflow_tests(self, value):
|
||||
self._workflow_tests = value
|
||||
|
||||
def get_webservice_responses(self):
|
||||
return WebserviceResponse.select([Equal('testdef_id', self.id)], order_by='name')
|
||||
|
||||
def get_admin_url(self):
|
||||
base_url = get_publisher().get_backoffice_url()
|
||||
objects_dir = 'forms' if self.object_type == 'formdefs' else 'cards'
|
||||
|
|
@ -150,6 +157,10 @@ class TestDef(sql.TestDef):
|
|||
for workflow_tests in workflow_tests_list:
|
||||
workflow_tests.remove_self()
|
||||
|
||||
responses = WebserviceResponse.select([Equal('testdef_id', id)])
|
||||
for response in responses:
|
||||
response.remove_self()
|
||||
|
||||
@classmethod
|
||||
def select_for_objectdef(cls, objectdef):
|
||||
return cls.select(
|
||||
|
|
@ -213,6 +224,7 @@ class TestDef(sql.TestDef):
|
|||
self.recorded_errors.append(str(error_summary or exception))
|
||||
|
||||
real_record_error = get_publisher().record_error
|
||||
real_http_adapter = getattr(get_publisher(), '._http_adapter', None)
|
||||
|
||||
true_request = get_publisher().get_request()
|
||||
wsgi_request = WSGIRequest({'REQUEST_METHOD': 'POST', 'wsgi.input': io.StringIO()})
|
||||
|
|
@ -222,13 +234,16 @@ class TestDef(sql.TestDef):
|
|||
get_publisher()._set_request(fake_request)
|
||||
fake_request.session = get_session_manager().new_session(None)
|
||||
get_publisher().record_error = record_error
|
||||
get_publisher()._http_adapter = MockWebserviceResponseAdapter(self)
|
||||
yield
|
||||
finally:
|
||||
get_publisher()._set_request(true_request)
|
||||
get_publisher().record_error = real_record_error
|
||||
get_publisher()._http_adapter = real_http_adapter
|
||||
|
||||
def run(self, objectdef):
|
||||
self.exception = None
|
||||
self.sent_requests = []
|
||||
self.recorded_errors = []
|
||||
self.missing_required_fields = []
|
||||
with self.fake_request():
|
||||
|
|
@ -505,3 +520,101 @@ class TestResult(sql.TestResult):
|
|||
base_url = get_publisher().get_backoffice_url()
|
||||
objects_dir = 'forms' if self.object_type == 'formdefs' else 'cards'
|
||||
return '%s/%s/%s/tests/results/%s/' % (base_url, objects_dir, self.object_id, self.id)
|
||||
|
||||
|
||||
class WebserviceResponseError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class MockWebserviceResponseAdapter(requests.adapters.HTTPAdapter):
|
||||
def __init__(self, testdef, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.testdef = testdef
|
||||
|
||||
def send(self, request, *args, **kwargs):
|
||||
try:
|
||||
return self._send(request, *args, **kwargs)
|
||||
except WebserviceResponseError:
|
||||
raise requests.exceptions.RequestError
|
||||
except Exception as e:
|
||||
# Webservice call can happen through templates which catch all exceptions.
|
||||
# Record error to ensure we have a trace nonetheless.
|
||||
get_publisher().record_error(
|
||||
_('Unexpected error when mocking webservice call for url %(url)s: %(error)s.')
|
||||
% {'url': request.url.split('?')[0], 'error': str(e)}
|
||||
)
|
||||
raise e
|
||||
|
||||
def _send(self, request, *args, **kwargs):
|
||||
request_info = {
|
||||
'url': request.url.split('?')[0],
|
||||
'method': request.method,
|
||||
'webservice_response_id': None,
|
||||
'forbidden_method': False,
|
||||
}
|
||||
self.testdef.sent_requests.append(request_info)
|
||||
|
||||
for response in self.testdef.get_webservice_responses():
|
||||
if response.is_configured() and response.match_request(request):
|
||||
break
|
||||
else:
|
||||
if request.method != 'GET':
|
||||
request_info['forbidden_method'] = True
|
||||
raise WebserviceResponseError
|
||||
return super().send(request, *args, **kwargs)
|
||||
|
||||
request_info['webservice_response_id'] = response.id
|
||||
|
||||
headers = {
|
||||
'Content-Type': 'application/json',
|
||||
}
|
||||
|
||||
raw_response = HTTPResponse(
|
||||
status=200,
|
||||
body=io.BytesIO(response.payload.encode()),
|
||||
headers=headers,
|
||||
original_response=self.make_original_response(headers),
|
||||
preload_content=False,
|
||||
)
|
||||
|
||||
return self.build_response(request, raw_response)
|
||||
|
||||
def make_original_response(self, headers):
|
||||
dummy_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
original_response = http.client.HTTPResponse(sock=dummy_socket)
|
||||
|
||||
original_headers = http.client.HTTPMessage()
|
||||
for k, v in headers.items():
|
||||
original_headers.add_header(k, v)
|
||||
original_response.msg = original_headers
|
||||
|
||||
return original_response
|
||||
|
||||
|
||||
class WebserviceResponse(XmlStorableObject):
|
||||
_names = 'webservice-response'
|
||||
xml_root_node = 'webservice-response'
|
||||
|
||||
testdef_id = None
|
||||
name = ''
|
||||
payload = None
|
||||
url = None
|
||||
|
||||
XML_NODES = [
|
||||
('testdef_id', 'int'),
|
||||
('name', 'str'),
|
||||
('payload', 'str'),
|
||||
('url', 'str'),
|
||||
]
|
||||
|
||||
def __str__(self):
|
||||
return self.name
|
||||
|
||||
def is_configured(self):
|
||||
return self.payload is not None and self.url
|
||||
|
||||
def match_request(self, request):
|
||||
if request.url.split('?')[0] != self.url:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
|
|
|||
Loading…
Reference in New Issue
.