0001-misc-mark-logged-error-emails-with-a-tech_id-referen.patch

Frédéric Péters, 29 octobre 2019 09:57

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

Subject: [PATCH] misc: mark logged error emails with a tech_id reference
 (#36807)

 tests/test_backoffice_pages.py | 51 ++++++++++++++++++++++++++++++++++
 wcs/logged_errors.py           |  3 +-
 wcs/publisher.py               | 35 ++++++++++++++++++++---
 wcs/qommon/emails.py           | 10 +++++--
 wcs/qommon/logger.py           |  8 ++++--
 wcs/qommon/publisher.py        | 26 -----------------
 6 files changed, 97 insertions(+), 36 deletions(-)

         assert (' with the number %s.' % number31.get_display_id()) in resp.body
         assert not 'Error during webservice call' in resp.body
     def test_backoffice_wscall_error_email(http_requests, pub, emails):
         pub.cfg['debug'] = {'error_email': 'errors@localhost.invalid'}
         pub.write_cfg()
         user = create_user(pub)
         create_environment(pub)
         formdef = FormDef.get_by_urlname('form-title')
         form_class = formdef.data_class()
         number31 = [x for x in form_class.select() if x.data['1'] == 'FOO BAR 30'][0]
         # attach a custom workflow
         workflow = Workflow(name='wscall')
         st1 = workflow.add_status('Status1', number31.status.split('-')[1])
         wscall = WebserviceCallStatusItem()
         wscall.id = '_wscall'
         wscall.varname = 'xxx'
         wscall.url = 'http://remote.example.net/xml'
         wscall.action_on_bad_data = ':stop'
         wscall.record_errors = True
         st1.items.append(wscall)
         wscall.parent = st1
         again = ChoiceWorkflowStatusItem()
         again.id = '_again'
         again.label = 'Again'
         again.by = ['_receiver']
         again.status = st1.id
         st1.items.append(again)
         again.parent = st1
         workflow.store()
         formdef.workflow_id = workflow.id
         formdef.store()
         app = login(get_app(pub))
         resp = app.get('/backoffice/management/form-title/%s/' % number31.id)
         assert (' with the number %s.' % number31.get_display_id()) in resp.body
         assert 'Again' in resp.body
         resp = resp.forms[0].submit('button_again')
         resp = resp.follow()
         assert 'Error during webservice call' in resp.body
         # check email box
         error_email = emails.get('[ERROR] [WSCALL] ValueError: No JSON object could be decoded')
         assert '/form-title/%s/' % number31.id in error_email['payload']
         assert error_email['msg']['References']
     def test_backoffice_wscall_attachment(http_requests, pub):
         create_user(pub)
         create_environment(pub)

             error.occurences_count += 1
             error.latest_occurence_timestamp = datetime.datetime.now()
             error.store()
             return error
         @classmethod
         def record_exception(cls, error_summary, plain_error_msg, publisher):
-...
                 return
             formdef = FormDef.get_by_urlname(formdef_urlname)
             formdata = formdef.data_class().get(formdata_id, ignore_errors=True)
             cls.record(error_summary, plain_error_msg, formdata=formdata,
             return cls.record(error_summary, plain_error_msg, formdata=formdata,
                        formdef=formdef, workflow=formdef.workflow)
         @property

     import os
     import random
     import sys
     import traceback
     import zipfile
     from django.utils.six.moves import cPickle
-...
     except ImportError:
         pass
     from .qommon.publisher import set_publisher_class, QommonPublisher
     from .qommon.publisher import set_publisher_class, QommonPublisher, get_request
     # this is terribly ugly but import RootDirectory will import a bunch of things,
     # and some of them need a publisher to be set
-...
             conn.commit()
             cur.close()
         def notify_of_exception(self, exc_tuple, context=None):
             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
             if context:
                 error_summary = '%s %s' % (context, error_summary)
             plain_error_msg = str(self._generate_plaintext_error(
                             get_request(),
                             self,
                             exc_type, exc_value,
                             tb))
             self.log_internal_error(error_summary, plain_error_msg, record=True)
         def log_internal_error(self, error_summary, plain_error_msg, record=False):
             super(WcsPublisher, self).log_internal_error(error_summary,
                     plain_error_msg, record=record)
             tech_id = None
             if record:
                 LoggedError.record_exception(error_summary, plain_error_msg, publisher=self)
                 logged_exception = LoggedError.record_exception(
                         error_summary, plain_error_msg, publisher=self)
                 if logged_exception:
                     tech_id = logged_exception.tech_id
             try:
                 self.logger.log_internal_error(error_summary, plain_error_msg, tech_id)
             except socket.error:
                 # will happen if there is no mail server available and exceptions
                 # were configured to be mailed.
                 pass
             except OSError:
                 # this could happen on file descriptor exhaustion
                 pass
         def apply_global_action_timeouts(self):
             from wcs.workflows import Workflow, WorkflowGlobalActionTimeoutTrigger

             return part
         get_logger().warn('Failed to build MIME part from %r', attachment)
     def email(subject, mail_body, email_rcpt, replyto = None, bcc = None,
     def email(subject, mail_body, email_rcpt, replyto=None, bcc=None,
             email_from=None, exclude_current_user=False, email_type=None,
             want_html=True, hide_recipients = False, fire_and_forget = False,
             smtp_timeout = None, attachments = ()):
             want_html=True, hide_recipients=False, fire_and_forget=False,
             smtp_timeout=None, attachments=(),
             extra_headers=None):
         if not get_request():
             # we are not processing a request, no sense delaying the handling
-...
             msg['Reply-To'] = replyto
         msg['X-Qommon-Id'] = os.path.basename(get_publisher().app_dir)
         if extra_headers:
             for key, value in extra_headers.items():
                 msg[key] = value
         if type(email_rcpt) is list:
             rcpts = email_rcpt[:]

     class ApplicationLogger(DefaultLogger):
         def log_internal_error(self, error_summary, error_msg):
         def log_internal_error(self, error_summary, error_msg, tech_id=None):
             self.log('exception caught')
             self.error_log.write(error_msg)
             if self.error_email:
                 from .emails import email
                 headers = {}
                 if tech_id:
                     headers['References'] = '<%s@%s>' % (tech_id, os.path.basename(get_publisher().app_dir))
                 email(subject='[ERROR] %s' % error_summary,
                       mail_body=error_msg,
                       email_from=self.error_email,
                       email_rcpt=[self.error_email],
                       want_html=False,
                       fire_and_forget=True)
                       fire_and_forget=True,
                       extra_headers=headers)
     class BotFilter(logging.Filter):

             return error_file.getvalue()
         def notify_of_exception(self, exc_tuple, context=None):
             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
             if context:
                 error_summary = '%s %s' % (context, error_summary)
             plain_error_msg = str(self._generate_plaintext_error(
                             get_request(),
                             self,
                             exc_type, exc_value,
                             tb))
             self.log_internal_error(error_summary, plain_error_msg, record=True)
         def log_internal_error(self, error_summary, plain_error_msg, record=False):
             try:
                 self.logger.log_internal_error(error_summary, plain_error_msg)
             except socket.error:
                 # will happen if there is no mail server available and exceptions
                 # were configured to be mailed.
                 pass
             except OSError:
                 # this could happen on file descriptor exhaustion
                 pass
         def finish_successful_request(self):
             if not self.get_request().ignore_session:
                 self.session_manager.finish_successful_request()
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-misc-mark-logged-error-emails-with-a-tech_id-referen.patch