altf4arnold/datatracker
1
0
Fork 0
Code Issues Pull requests Actions 3 Packages Projects Releases Wiki Activity

Create management command base class that sends emails on exceptions. Fixes #3356 and #3357. Commit ready for merge.

Browse source 
- Legacy-Id: 19493
This commit is contained in:
Jennifer Richards 2021-10-29 01:58:21 +00:00
parent 0d955b6328
commit 968b775315
8 changed files with 434 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
ietf/ipr/management/commands/process_email.py
View file

@ -4,29 +4,49 @@
import io
import sys
from textwrap import dedent
from django.core.management.base import BaseCommand, CommandError
from django.core.management import CommandError
from ietf.utils.management.base import EmailOnFailureCommand
from ietf.ipr.mail import process_response_email
import debug # pyflakes:ignore
class Command(BaseCommand):
class Command(EmailOnFailureCommand):
help = ("Process incoming email responses to ipr mail")
msg_bytes = None
def add_arguments(self, parser):
super().add_arguments(parser)
parser.add_argument('--email-file', dest='email', help='File containing email (default: stdin)')
def handle(self, *args, **options):
email = options.get('email', None)
msg = None
if not email:
msg = sys.stdin.read()
self.msg_bytes = msg.encode()
else:
msg = io.open(email, "r").read()
self.msg_bytes = io.open(email, "rb").read()
msg = self.msg_bytes.decode()
try:
process_response_email(msg)
except ValueError as e:
raise CommandError(e)
failure_subject = 'Error during ipr email processing'
failure_message = dedent("""\
An error occurred in the ipr process_email management command.
{error_summary}
""")
def make_failure_message(self, error, **extra):
msg = super().make_failure_message(error, **extra)
if self.msg_bytes is not None:
msg.add_attachment(
self.msg_bytes,
'application', 'octet-stream', # mime type
filename='original-message',
)
return msg

49
ietf/ipr/management/tests.py Normal file
View file

@ -0,0 +1,49 @@
# Copyright The IETF Trust 2021, All Rights Reserved
# -*- coding: utf-8 -*-
"""Tests of ipr management commands"""
import mock
from django.core.management import call_command
from django.test.utils import override_settings
from ietf.utils.test_utils import TestCase, name_of_file_containing
@override_settings(ADMINS=(('Some Admin', 'admin@example.com'),))
class ProcessEmailTests(TestCase):
@mock.patch('ietf.ipr.management.commands.process_email.process_response_email')
def test_process_email(self, process_mock):
"""The process_email command should process the correct email"""
with name_of_file_containing('contents') as filename:
call_command('process_email', email_file=filename)
self.assertEqual(process_mock.call_count, 1, 'process_response_email should be called once')
self.assertEqual(
process_mock.call_args.args,
('contents',),
'process_response_email should receive the correct contents'
)
@mock.patch('ietf.utils.management.base.send_smtp')
@mock.patch('ietf.ipr.management.commands.process_email.process_response_email')
def test_send_error_to_admin(self, process_mock, send_smtp_mock):
"""The process_email command should email the admins on error"""
# arrange an mock error during processing
process_mock.side_effect = RuntimeError('mock error')
with name_of_file_containing('contents') as filename:
call_command('process_email', email_file=filename)
self.assertTrue(send_smtp_mock.called)
(msg,) = send_smtp_mock.call_args.args
self.assertEqual(msg['to'], 'admin@example.com', 'Admins should be emailed on error')
self.assertIn('error', msg['subject'].lower(), 'Error email subject should indicate error')
self.assertTrue(msg.is_multipart(), 'Error email should have attachments')
parts = msg.get_payload()
self.assertEqual(len(parts), 3, 'Error email should contain message, traceback, and original message')
content = parts[0].get_payload()
traceback = parts[1].get_payload()
original = parts[2].get_payload(decode=True).decode() # convert octet-stream to string
self.assertIn('RuntimeError', content, 'Error type should be included in error email')
self.assertIn('mock.py', content, 'File where error occurred should be included in error email')
self.assertIn('traceback', traceback.lower(), 'Traceback should be attached to error email')
self.assertEqual(original, 'contents', 'Original message should be attached to error email')

59
ietf/nomcom/management/commands/feedback_email.py
View file

@ -4,47 +4,82 @@
import io
import sys
from textwrap import dedent
from django.core.management.base import BaseCommand, CommandError
from django.core.management import CommandError
from ietf.utils.log import log
from ietf.utils.management.base import EmailOnFailureCommand
from ietf.nomcom.models import NomCom
from ietf.nomcom.utils import create_feedback_email
from ietf.nomcom.fields import EncryptedException
import debug # pyflakes:ignore
class Command(BaseCommand):
class Command(EmailOnFailureCommand):
help = ("Receive nomcom email, encrypt and save it.")
nomcom = None
msg = None # incoming message
def add_arguments(self, parser):
super().add_arguments(parser)
parser.add_argument('--nomcom-year', dest='year', help='NomCom year')
parser.add_argument('--email-file', dest='email', help='File containing email (default: stdin)')
def handle(self, *args, **options):
email = options.get('email', None)
year = options.get('year', None)
msg = None
nomcom = None
help_message = 'Usage: feeback_email --nomcom-year <nomcom-year> --email-file <email-file>'
if not year:
log("Error: missing nomcom-year")
raise CommandError("Missing nomcom-year\n\n"+help_message)
if not email:
msg = io.open(sys.stdin.fileno(), 'rb').read()
else:
msg = io.open(email, "rb").read()
raise CommandError("Missing nomcom-year\n\n" + help_message)
try:
nomcom = NomCom.objects.get(group__acronym__icontains=year,
self.nomcom = NomCom.objects.get(group__acronym__icontains=year,
group__state__slug='active')
except NomCom.DoesNotExist:
raise CommandError("NomCom %s does not exist or it isn't active" % year)
if not email:
self.msg = io.open(sys.stdin.fileno(), 'rb').read()
else:
self.msg = io.open(email, "rb").read()
try:
feedback = create_feedback_email(nomcom, msg)
feedback = create_feedback_email(self.nomcom, self.msg)
log("Received nomcom email from %s" % feedback.author)
except (EncryptedException, ValueError) as e:
raise CommandError(e)
# Configuration for the email to be sent on failure
failure_email_includes_traceback = False # error messages might contain pieces of the feedback email
failure_subject = '{nomcom}: error during feedback email processing'
failure_message = dedent("""\
An error occurred in the nomcom feedback_email management command while
processing feedback for {nomcom}.
{error_summary}
""")
@property
def failure_recipients(self):
return self.nomcom.chair_emails() if self.nomcom else super().failure_recipients
def make_failure_message(self, error, **extra):
failure_message = super().make_failure_message(
error,
nomcom=self.nomcom or 'nomcom',
**extra
)
if self.nomcom and self.msg:
# Attach incoming message if we have it and are sending to the nomcom chair.
# Do not attach it if we are sending to the admins. Send as a generic
# mime type because we don't know for sure that it was actually a valid
# message.
failure_message.add_attachment(
self.msg,
'application', 'octet-stream', # mime type
filename='original-message',
)
return failure_message

84
ietf/nomcom/management/tests.py Normal file
View file

@ -0,0 +1,84 @@
# Copyright The IETF Trust 2021, All Rights Reserved
# -*- coding: utf-8 -*-
"""Tests of nomcom management commands"""
import mock
from collections import namedtuple
from django.core.management import call_command
from django.test.utils import override_settings
from ietf.nomcom.factories import NomComFactory
from ietf.utils.test_utils import TestCase, name_of_file_containing
@override_settings(ADMINS=(('Some Admin', 'admin@example.com'),))
class FeedbackEmailTests(TestCase):
def setUp(self):
self.year = 2021
self.nomcom = NomComFactory(group__acronym=f'nomcom{self.year}')
@mock.patch('ietf.utils.management.base.send_smtp')
def test_send_error_to_admins(self, send_smtp_mock):
"""If a nomcom chair cannot be identified, mail goes to admins
This email should not contain either the full traceback or the original message.
"""
# Call with the wrong nomcom year so the admin will be contacted
with name_of_file_containing('feedback message') as filename:
call_command('feedback_email', nomcom_year=self.year + 1, email_file=filename)
self.assertTrue(send_smtp_mock.called)
(msg,) = send_smtp_mock.call_args.args # get the message to be sent
self.assertEqual(msg['to'], 'admin@example.com', 'Email recipient should be the admins')
self.assertIn('error', msg['subject'], 'Email subject should indicate error')
self.assertFalse(msg.is_multipart(), 'Nomcom feedback error sent to admin should not have attachments')
content = msg.get_payload()
self.assertIn('CommandError', content, 'Admin email should contain error type')
self.assertIn('feedback_email.py', content, 'Admin email should contain file where error occurred')
self.assertNotIn('traceback', content.lower(), 'Admin email should not contain traceback')
self.assertNotIn(f'NomCom {self.year} does not exist', content,
'Admin email should not contain error message')
# not going to check the line - that's too likely to change
@mock.patch('ietf.utils.management.base.send_smtp')
@mock.patch('ietf.nomcom.management.commands.feedback_email.create_feedback_email')
def test_send_error_to_chair(self, create_feedback_mock, send_smtp_mock):
# mock an exception in create_feedback_email()
create_feedback_mock.side_effect = RuntimeError('mock error')
with name_of_file_containing('feedback message') as filename:
call_command('feedback_email', nomcom_year=self.year, email_file=filename)
self.assertTrue(send_smtp_mock.called)
(msg,) = send_smtp_mock.call_args.args # get the message to be sent
self.assertCountEqual(
[addr.strip() for addr in msg['to'].split(',')],
self.nomcom.chair_emails(),
'Email recipient should be the nomcom chair(s)',
)
self.assertIn('error', msg['subject'], 'Email subject should indicate error')
self.assertTrue(msg.is_multipart(), 'Chair feedback error should have attachments')
parts = msg.get_payload()
content = parts[0].get_payload()
# decode=True decodes the base64 encoding, .decode() converts the octet-stream bytes to a string
attachment = parts[1].get_payload(decode=True).decode()
self.assertIn('RuntimeError', content, 'Nomcom email should contain error type')
self.assertIn('mock.py', content, 'Nomcom email should contain file where error occurred')
self.assertIn('feedback message', attachment, 'Nomcom email should include original message')
@mock.patch('ietf.nomcom.management.commands.feedback_email.create_feedback_email')
def test_feedback_email(self, create_feedback_mock):
"""The feedback_email command should create feedback"""
# mock up the return value
create_feedback_mock.return_value = namedtuple('mock_feedback', 'author')('author@example.com')
with name_of_file_containing('feedback message') as filename:
call_command('feedback_email', nomcom_year=self.year, email_file=filename)
self.assertEqual(create_feedback_mock.call_count, 1, 'create_feedback_email() should be called once')
self.assertEqual(
create_feedback_mock.call_args.args,
(self.nomcom, b'feedback message'),
'feedback_email should process the correct email for the correct nomcom'
)

9
ietf/nomcom/models.py
View file

@ -102,6 +102,15 @@ class NomCom(models.Model):
else:
raise EncryptedException(error)
def chair_emails(self):
if not hasattr(self, '_cached_chair_emails'):
if self.group:
self._cached_chair_emails = list(
self.group.role_set.filter(name_id='chair').values_list('email__address', flat=True)
)
else:
self._cached_chair_emails = []
return self._cached_chair_emails
def delete_nomcom(sender, **kwargs):
nomcom = kwargs.get('instance', None)

96
ietf/utils/management/base.py Normal file
View file

@ -0,0 +1,96 @@
# Copyright The IETF Trust 2013-2020, All Rights Reserved
# -*- coding: utf-8 -*-
from email.message import EmailMessage
from textwrap import dedent
from traceback import format_exception, extract_tb
from django.conf import settings
from django.core.management.base import BaseCommand
from ietf.utils.mail import send_smtp
import debug # pyflakes:ignore
class EmailOnFailureCommand(BaseCommand):
"""Command that sends email when an exception occurs
Subclasses can override failure_message, failure_subject, and failure_recipients
to customize the behavior. Both failure_subject and failure_message are formatted
with keywords for interpolation. By default, the following substitutions are
available:
{error} - the exception instance
{error_summary} - multiline summary of error type and location where it occurred
More interpolation values can be added through the **extra argument to
make_failure_message().
By default, the full traceback will be attached to the notification email.
To disable this, set failure_email_includes_traceback to False.
When a command is executed, its handle() method will be called as usual.
If an exception occurs, instead of printing this to the terminal and
exiting with an error, a message generated via the make_failure_message()
method will be sent to failure_recipients. The command will exit successfully
to the shell.
This can be prevented for debugging by passing the --no-failure-email option.
In this case, the usual error handling will be used. To make this available,
the subclass must call super().add_arguments() in its own add_arguments() method.
"""
failure_message = dedent("""\
An exception occurred:
{error}
""")
failure_subject = 'Exception in management command'
failure_email_includes_traceback = True
@property
def failure_recipients(self):
return tuple(item[1] for item in settings.ADMINS)
def execute(self, *args, **options):
try:
super().execute(*args, **options)
except Exception as error:
if options['email_on_failure']:
msg = self.make_failure_message(error)
send_smtp(msg)
else:
raise
def _summarize_error(self, error):
frame = extract_tb(error.__traceback__)[-1]
return dedent(f"""\
Error details:
Exception type: {type(error).__module__}.{type(error).__name__}
File: {frame.filename}
Line: {frame.lineno}""")
def make_failure_message(self, error, **extra):
"""Generate an EmailMessage to report an error"""
format_values = dict(
error=error,
error_summary=self._summarize_error(error),
)
format_values.update(**extra)
msg = EmailMessage()
msg['To'] = self.failure_recipients
msg['From'] = settings.SERVER_EMAIL
msg['Subject'] = self.failure_subject.format(**format_values)
msg.set_content(
self.failure_message.format(**format_values)
)
if self.failure_email_includes_traceback:
msg.add_attachment(
''.join(format_exception(None, error, error.__traceback__)),
filename='traceback.txt',
)
return msg
def add_arguments(self, parser):
parser.add_argument('--no-failure-email', dest='email_on_failure', action='store_false',
help='Disable sending email on failure')

107
ietf/utils/management/tests.py Normal file
View file

@ -0,0 +1,107 @@
# Copyright The IETF Trust 2013-2020, All Rights Reserved
# -*- coding: utf-8 -*-
import mock
from django.core.management import call_command, CommandError
from django.test import override_settings
from ietf.utils.mail import outbox, empty_outbox
from ietf.utils.management.base import EmailOnFailureCommand
from ietf.utils.test_utils import TestCase
@mock.patch.object(EmailOnFailureCommand, 'handle')
class EmailOnFailureCommandTests(TestCase):
def test_calls_handle(self, handle_method):
call_command(EmailOnFailureCommand())
self.assertEqual(handle_method.call_count, 1)
def test_sends_email(self, handle_method):
handle_method.side_effect = CommandError('error during the command')
empty_outbox()
admins = (
('admin one', 'admin1@example.com'),
('admin two', 'admin2@example.com'),
)
with override_settings(ADMINS=admins, SERVER_EMAIL='server@example.com'):
call_command(EmailOnFailureCommand())
self.assertEqual(len(outbox), 1)
msg = outbox[0]
self.assertEqual(msg['to'], 'admin1@example.com, admin2@example.com',
'Outgoing email recipients did not default to settings.ADMINS')
self.assertEqual(msg['from'], 'server@example.com',
'Outgoing email sender did not default to settings.SERVER_EMAIL')
self.assertTrue(msg.is_multipart())
parts = msg.get_payload()
self.assertEqual(len(parts), 2)
self.assertIn('error during the command', parts[0].get_content())
self.assertIn('error during the command', parts[1].get_content())
self.assertIn('Traceback', parts[1].get_content())
def test_disable_email(self, handle_method):
handle_method.side_effect = CommandError('error during the command')
empty_outbox()
with self.assertRaises(CommandError):
call_command(EmailOnFailureCommand(), '--no-failure-email')
self.assertEqual(len(outbox), 0)
def test_customize_email(self, handle_method):
class _SubclassCommand(EmailOnFailureCommand):
failure_message = 'simple message with the {error} and {other}\n'
failure_recipients = 'someone@example.com'
failure_subject = 'subject of the email'
def make_failure_message(self, error, **extra):
msg = super().make_failure_message(error, other='additional info', **extra)
msg.add_attachment('attached\n')
return msg
handle_method.side_effect = CommandError('error during the command')
empty_outbox()
with override_settings(
ADMINS=('a1', 'admin@example.com'),
SERVER_EMAIL='server@example.com',
):
call_command(_SubclassCommand())
self.assertEqual(len(outbox), 1)
msg = outbox[0]
self.assertEqual(msg['to'], 'someone@example.com',
'Outgoing email recipients were not customized')
self.assertEqual(msg['from'], 'server@example.com',
'Outgoing email sender did not default to settings.SERVER_EMAIL')
self.assertEqual(msg['subject'], 'subject of the email',
'Outgoing email subject was not customized')
self.assertTrue(msg.is_multipart())
parts = msg.get_payload()
self.assertEqual(len(parts), 3, 'Attachment was not added')
self.assertEqual(
parts[0].get_content(),
'simple message with the error during the command and additional info\n',
)
self.assertIn('error during the command', parts[1].get_content())
self.assertIn('Traceback', parts[1].get_content())
self.assertEqual('attached\n', parts[2].get_content())
def test_disable_traceback(self, handle_method):
"""Traceback should not be included when disabled"""
class _SubclassCommand(EmailOnFailureCommand):
failure_email_includes_traceback = False
handle_method.side_effect = CommandError('error during the command')
empty_outbox()
with override_settings(
ADMINS=('a1', 'admin@example.com'),
SERVER_EMAIL='server@example.com',
):
call_command(_SubclassCommand())
self.assertEqual(len(outbox), 1)
msg = outbox[0]
if msg.is_multipart():
parts = msg.get_payload()
self.assertEqual(len(parts), 1, 'Traceback should not be attached')
content = parts[0].get_content()
else:
content = msg.get_payload()
self.assertNotIn('Traceback', content)

14
ietf/utils/test_utils.py
View file

@ -41,9 +41,11 @@ import html5lib
import sys
from urllib.parse import unquote
from unittest.util import strclass
from bs4 import BeautifulSoup
from contextlib import contextmanager
from pathlib import Path
from tempfile import NamedTemporaryFile
import django.test
from django.conf import settings
@ -100,6 +102,16 @@ def reload_db_objects(*objects):
else:
return t
@contextmanager
def name_of_file_containing(contents):
"""Get a context with the name of an email file"""
f = NamedTemporaryFile('w', delete=False)
f.write(contents)
f.close()
yield f.name # hand the filename to the context
Path(f.name).unlink() # clean up after context exits
def assert_ical_response_is_valid(test_inst, response, expected_event_summaries=None,
expected_event_uids=None, expected_event_count=None):
"""Validate an HTTP response containing iCal data