0001-wscall-notify-and-record-errors-on-failure-44050.patch

Lauréline Guérin, 15 septembre 2020 14:06

Voir les différences: en ligne côte à côte

Subject: [PATCH] wscall: notify and record errors on failure (#44050)

 tests/test_workflows.py                  |  2 +-
 tests/test_wscall.py                     | 58 +++++++++++++++++++++-
 tests/test_wscalls_admin_pages.py        | 19 +++++++-
 tests/utilities.py                       |  1 +
 wcs/admin/wscalls.py                     | 12 +++++
 wcs/publisher.py                         | 10 ++--
 wcs/templates/wcs/backoffice/wscall.html |  7 +++
 wcs/wf/wscall.py                         | 18 +------
 wcs/wscalls.py                           | 61 +++++++++++++++++++++---
 9 files changed, 158 insertions(+), 30 deletions(-)

         with mock.patch('wcs.wscalls.get_secret_and_orig') as mocked_secret_and_orig:
             mocked_secret_and_orig.return_value = ('secret', 'localhost')
             with mock.patch('wcs.qommon.misc._http_request') as mocked_http_post:
                 mocked_http_post.return_value = ('response', '200', 'data', 'headers')
                 mocked_http_post.return_value = (mock.Mock(headers={}), 200, 'data', 'headers')
                 item.perform(formdata)
                 url = mocked_http_post.call_args[0][0]
                 payload = mocked_http_post.call_args[1]['body']

     import json
     import pytest
     from wcs import fields
     from wcs.formdef import FormDef
     from wcs.logged_errors import LoggedError
     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
     from utilities import get_app, create_temporary_pub, clean_temporary_pub
     @pytest.fixture
-...
             pass
         assert http_requests.get_last('url') == wscall.request['url']
         assert http_requests.get_last('method') == 'DELETE'
     @pytest.mark.parametrize('notify_on_errors', [True, False])
     @pytest.mark.parametrize('record_on_errors', [True, False])
     def test_webservice_on_error(http_requests, pub, emails, notify_on_errors, record_on_errors):
         pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}
         pub.write_cfg()
         NamedWsCall.wipe()
         LoggedError.wipe()
         FormDef.wipe()
         wscall = NamedWsCall()
         wscall.name = 'Hello world'
         wscall.notify_on_errors = notify_on_errors
         wscall.record_on_errors = record_on_errors
         wscall.store()
         assert wscall.slug == 'hello_world'
         formdef = FormDef()
         formdef.name = 'foobar'
         formdef.fields = [
             fields.CommentField(id='0', label='Foo Bar {{ webservice.hello_world }}', type='comment'),
+        ]
         formdef.store()
         for url_part in ['json', 'json-err0', 'json-errheader0']:
             wscall.request = {'url': 'http://remote.example.net/%s' % url_part}
             wscall.store()
             resp = get_app(pub).get('/foobar/')
             assert 'Foo Bar ' in resp.text
             assert emails.count() == 0
             assert LoggedError.count() == 0
         for url_part in ['400', '400-json', '404', '404-json', '500', 'json-err1', 'json-errheader1']:
             status_code = 200
             if not url_part.startswith('json'):
                 status_code = url_part[:3]
             wscall.request = {'url': 'http://remote.example.net/%s' % url_part}
             wscall.store()
             resp = get_app(pub).get('/foobar/')
             assert 'Foo Bar ' in resp.text
             if notify_on_errors:
                 assert emails.count() == 1
                 assert "[ERROR] ['WSCALL'] Exception: %s whatever" % status_code in emails.emails
                 emails.empty()
             else:
                 assert emails.count() == 0
             if record_on_errors:
                 assert LoggedError.count() == 1
                 LoggedError.wipe()
             else:
                 assert LoggedError.count() == 0

         NamedWsCall.wipe()
         wscall = NamedWsCall(name='xxx')
         wscall.description = 'description'
         wscall.notify_on_errors = True
         wscall.record_on_errors = True
         wscall.request = {
                 'url': 'http://remote.example.net/json',
                 'request_signature_key': 'xxx',
-...
         return wscall
     def test_wscalls_new(pub):
     @pytest.mark.parametrize('value', [True, False])
     def test_wscalls_new(pub, value):
         create_superuser(pub)
         NamedWsCall.wipe()
         app = login(get_app(pub))
-...
         # go to the page and add a webservice call
         resp = app.get('/backoffice/settings/wscalls/')
         resp = resp.click('New webservice call')
         assert resp.form['notify_on_errors'].value == 'yes'
         assert resp.form['record_on_errors'].value == 'yes'
         resp.form['name'] = 'a new webservice call'
         resp.form['description'] = 'description'
         resp.form['notify_on_errors'] = value
         resp.form['record_on_errors'] = value
         resp.form['request$url'] = 'http://remote.example.net/json'
         resp = resp.form.submit('submit')
         assert resp.location == 'http://example.net/backoffice/settings/wscalls/'
-...
         assert 'Edit webservice call' in resp.text
         assert NamedWsCall.get('a_new_webservice_call').name == 'a new webservice call'
         assert NamedWsCall.get('a_new_webservice_call').notify_on_errors == value
         assert NamedWsCall.get('a_new_webservice_call').record_on_errors == value
     def test_wscalls_view(pub, wscall):
-...
         resp = app.get('/backoffice/settings/wscalls/xxx/')
         resp = resp.click(href='edit')
         assert resp.form['name'].value == 'xxx'
         assert resp.form['notify_on_errors'].value == 'yes'
         assert resp.form['record_on_errors'].value == 'yes'
         assert 'slug' in resp.form.fields
         resp.form['description'] = 'bla bla bla'
         resp.form['notify_on_errors'] = False
         resp.form['record_on_errors'] = False
         resp = resp.form.submit('submit')
         assert resp.location == 'http://example.net/backoffice/settings/wscalls/xxx/'
         resp = resp.follow()
         assert NamedWsCall.get('xxx').description == 'bla bla bla'
         assert NamedWsCall.get('xxx').notify_on_errors is False
         assert NamedWsCall.get('xxx').record_on_errors is False
         resp = app.get('/backoffice/settings/wscalls/xxx/')
         resp = resp.click(href='edit')
         assert resp.form['name'].value == 'xxx'
         assert resp.form['notify_on_errors'].value is None
         assert resp.form['record_on_errors'].value is None
         assert 'slug' in resp.form.fields
         resp.form['slug'] = 'yyy'
         resp = resp.form.submit('submit')

tests/utilities.py
324	324
325	325	status, data, headers = {
326	326	'http://remote.example.net/204': (204, None, None),
	327	'http://remote.example.net/400': (400, 'bad request', None),
327	328	'http://remote.example.net/400-json': (400, '{"err": 1, "err_desc": ":("}', None),
328	329	'http://remote.example.net/404': (404, 'page not found', None),
329	330	'http://remote.example.net/404-json': (404, '{"err": 1}', None),

                     value=self.wscall.request,
                     title=_('Request'), required=True)
             form.add(
                 CheckboxWidget,
                 'notify_on_errors',
                 title=_('Notify on errors'),
                 value=self.wscall.notify_on_errors if self.wscall.slug else True)
             form.add(
                 CheckboxWidget,
                 'record_on_errors',
                 title=_('Record on errors'),
                 value=self.wscall.record_on_errors if self.wscall.slug else True)
             if not self.wscall.is_readonly():
                 form.add_submit('submit', _('Submit'))
             form.add_submit('cancel', _('Cancel'))
-...
             self.wscall.name = name
             self.wscall.description = form.get_widget('description').parse()
             self.wscall.notify_on_errors = form.get_widget('notify_on_errors').parse()
             self.wscall.record_on_errors = form.get_widget('record_on_errors').parse()
             self.wscall.request = form.get_widget('request').parse()
             if self.wscall.slug:
                 self.wscall.slug = slug

             conn.commit()
             cur.close()
         def notify_of_exception(self, exc_tuple, context=None):
         def notify_of_exception(self, exc_tuple, context=None, record=True, notify=True):
             exc_type, exc_value, tb = exc_tuple
             error_summary = traceback.format_exception_only(exc_type, exc_value)
             error_summary = error_summary[0][0:-1] # de-listify and strip newline
             error_summary = error_summary[0][0:-1]  # de-listify and strip newline
             if context:
                 error_summary = '%s %s' % (context, error_summary)
-...
                             exc_type, exc_value,
                             tb))
             self.log_internal_error(error_summary, plain_error_msg, record=True)
             self.log_internal_error(error_summary, plain_error_msg, record=record, notify=notify)
         def log_internal_error(self, error_summary, plain_error_msg, record=False):
         def log_internal_error(self, error_summary, plain_error_msg, record=False, notify=True):
             tech_id = None
             if record:
                 logged_exception = LoggedError.record_exception(
                         error_summary, plain_error_msg, publisher=self)
                 if logged_exception:
                     tech_id = logged_exception.tech_id
             if not notify:
                 return
             try:
                 self.logger.log_internal_error(error_summary, plain_error_msg, tech_id)
             except socket.error:

       <li>{% trans "Method:" %} {% if wscall.request.method == 'POST' %}POST{% else %}GET{% endif %}</li>
     </ul>
     </div>
     <div class="bo-block">
     <ul>
       <li>{% trans "Notify on errors:" %} {{ wscall.notify_on_errors|yesno }}</li>
       <li>{% trans "Record on errors:" %} {{ wscall.record_on_errors|yesno }}</li>
     </ul>
     </div>
     {% endblock %}

     from ..qommon.misc import json_loads
     from wcs.workflows import (WorkflowStatusItem, register_item_class,
             AbortActionException, AttachmentEvolutionPart)
     from wcs.wscalls import call_webservice
     from wcs.wscalls import call_webservice, get_app_error_code
     class JournalWsCallErrorPart: #pylint: disable=C1001
-...
                         exc_info=sys.exc_info())
                 return
             app_error_code = 0
             app_error_code = get_app_error_code(response, data, self.response_type)
             app_error_code_header = response.headers.get('x-error-code')
             if app_error_code_header:
                 # result is good only if header value is '0'
                 try:
                     app_error_code = int(app_error_code_header)
                 except ValueError as e:
                     app_error_code = app_error_code_header
             elif self.response_type == 'json':
                 try:
                     d = json_loads(data)
                 except (ValueError, TypeError) as e:
                     pass
                 else:
                     if isinstance(d, dict) and d.get('err'):
                         app_error_code = d['err']
             if self.varname:
                 workflow_data.update({

     from .qommon import _, force_str
     from .qommon.misc import simplify, get_variadic_url, JSONEncoder, json_loads
     from .qommon.xml_storage import XmlStorableObject
     from .qommon.form import (CompositeWidget, StringWidget, WidgetDict,
             ComputedExpressionWidget, RadiobuttonsWidget, CheckboxWidget)
     from .qommon.form import (
         CompositeWidget, StringWidget, WidgetDict,
         ComputedExpressionWidget, RadiobuttonsWidget, CheckboxWidget)
     from .qommon import misc
     from .qommon.template import Template
-...
     from wcs.workflows import WorkflowStatusItem
     def call_webservice(url, qs_data=None, request_signature_key=None,
     def get_app_error_code(response, data, response_type):
         app_error_code = 0
         app_error_code_header = response.headers.get('x-error-code')
         if app_error_code_header:
             # result is good only if header value is '0'
             try:
                 app_error_code = int(app_error_code_header)
             except ValueError:
                 app_error_code = app_error_code_header
         elif response_type == 'json':
             try:
                 d = json_loads(data)
             except (ValueError, TypeError):
                 pass
             else:
                 if isinstance(d, dict) and d.get('err'):
                     app_error_code = d['err']
         return app_error_code
     def call_webservice(
             url, qs_data=None, request_signature_key=None,
             method=None, post_data=None, post_formdata=None, formdata=None,
             cache=False, **kwargs):
             cache=False, notify_on_errors=False, record_on_errors=False, **kwargs):
         url = url.strip()
         if Template.is_template_string(url):
-...
             if cache is True and request and hasattr(request, 'wscalls_cache'):
                 request.wscalls_cache[unsigned_url] = (status, data)
         app_error_code = get_app_error_code(response, data, 'json')
         if (app_error_code != 0 or status >= 400) and (notify_on_errors or record_on_errors):
             summary = '<no response>'
             if response is not None:
                 summary = '%s %s' % (status, response.reason)
             try:
                 raise Exception(summary)
             except Exception:
                 exc_info = sys.exc_info()
             get_publisher().notify_of_exception(
                 exc_info, context=['WSCALL'], notify=notify_on_errors, record=record_on_errors)
         return (response, status, data)
-...
         slug = None
         description = None
         request = None
         notify_on_errors = False
         record_on_errors = False
         # declarations for serialization
         XML_NODES = [('name', 'str'), ('slug', 'str'), ('description', 'str'),
                 ('request', 'request'),]
         XML_NODES = [
             ('name', 'str'),
             ('slug', 'str'),
             ('description', 'str'),
             ('request', 'request'),
             ('notify_on_errors', 'bool'),
             ('record_on_errors', 'bool'),
+        ]
         def __init__(self, name=None):
             XmlStorableObject.__init__(self)
-...
             return {'webservice': WsCallsSubstitutionProxy()}
         def call(self):
             (response, status, data) = call_webservice(cache=True, **self.request)
             (response, status, data) = call_webservice(
                 cache=True,
                 notify_on_errors=self.notify_on_errors,
                 record_on_errors=self.record_on_errors,
                 **self.request)
             return json_loads(data)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-wscall-notify-and-record-errors-on-failure-44050.patch