holm/django-helpdeskmig
1
0
Fork 0
mirror of https://gitea.mueller.network/extern/django-helpdesk.git synced 2024-12-25 16:18:51 +01:00
Code Issues Packages Projects Releases Wiki Activity

Fix OAuth tests

Browse Source
This commit is contained in:
bruce.gibbins 2023-04-19 15:12:13 +10:00
parent af2683d44b
commit 012cc7041a
2 changed files with 472 additions and 413 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
helpdesk/email.py
View File

@ -28,12 +28,14 @@ from helpdesk.models import FollowUp, IgnoreEmail, Queue, Ticket
import imaplib import imaplib
import logging import logging
import mimetypes import mimetypes
from oauthlib.oauth2 import BackendApplicationClient import oauthlib.oauth2 as oauth2lib
# from oauthlib.oauth2 import BackendApplicationClient
import os import os
from os.path import isfile, join from os.path import isfile, join
import poplib import poplib
import re import re
from requests_oauthlib import OAuth2Session # from requests_oauthlib import OAuth2Session
import requests_oauthlib
import socket import socket
import ssl import ssl
import sys import sys
@ -231,19 +233,19 @@ def imap_oauth_sync(q, logger, server):
IMAP eMail server with OAUTH authentication. IMAP eMail server with OAUTH authentication.
Only tested against O365 implementation Only tested against O365 implementation
Uses OAUTH Dict in Settings. Uses HELPDESK OAUTH Dict in Settings.
""" """
try: try:
logger.debug("Start Mailbox polling via IMAP OAUTH") logger.debug("Start Mailbox polling via IMAP OAUTH")
client = BackendApplicationClient( client = oauth2lib.BackendApplicationClient(
client_id=settings.HELPDESK_OAUTH["client_id"], client_id=settings.HELPDESK_OAUTH["client_id"],
scope=settings.HELPDESK_OAUTH["scope"], scope=settings.HELPDESK_OAUTH["scope"],
) )
oauth = OAuth2Session(client=client) oauth = requests_oauthlib.OAuth2Session(client=client)
token = oauth.fetch_token( token = oauth.fetch_token(
token_url=settings.HELPDESK_OAUTH["token_url"], token_url=settings.HELPDESK_OAUTH["token_url"],
client_id=settings.HELPDESK_OAUTH["client_id"], client_id=settings.HELPDESK_OAUTH["client_id"],

873
helpdesk/tests/test_get_email.py
View File

@ -8,6 +8,8 @@ from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.management import call_command from django.core.management import call_command
from django.shortcuts import get_object_or_404 from django.shortcuts import get_object_or_404
from django.test import override_settings, TestCase from django.test import override_settings, TestCase
from oauthlib.oauth2 import BackendApplicationClient
import helpdesk.email import helpdesk.email
from helpdesk.email import extract_part_data, object_from_message from helpdesk.email import extract_part_data, object_from_message
from helpdesk.exceptions import DeleteIgnoredTicketException, IgnoreTicketException from helpdesk.exceptions import DeleteIgnoredTicketException, IgnoreTicketException
@ -20,6 +22,7 @@ import os
from shutil import rmtree from shutil import rmtree
import sys import sys
from tempfile import mkdtemp from tempfile import mkdtemp
import time
import typing import typing
from unittest import mock from unittest import mock
@ -32,6 +35,8 @@ unrouted_email_server = "0.0.0.1"
# the last user port, reserved by IANA # the last user port, reserved by IANA
unused_port = "49151" unused_port = "49151"
fake_time = time.time()
class GetEmailCommonTests(TestCase): class GetEmailCommonTests(TestCase):
@ -283,6 +288,17 @@ class GetEmailParametricTemplate(object):
self.queue_public = Queue.objects.create(**kwargs) self.queue_public = Queue.objects.create(**kwargs)
self.token = {
'token_type': 'Bearer',
'access_token': 'asdfoiw37850234lkjsdfsdf',
'refresh_token': 'sldvafkjw34509s8dfsdf',
'expires_in': '3600',
'expires_at': fake_time + 3600,
}
self.client_id = 'foo'
self.client = BackendApplicationClient(self.client_id)
def tearDown(self): def tearDown(self):
rmtree(self.temp_logdir) rmtree(self.temp_logdir)
@ -365,75 +381,11 @@ class GetEmailParametricTemplate(object):
return_value=mocked_imaplib_server) return_value=mocked_imaplib_server)
call_command('get_email') call_command('get_email')
ticket1 = get_object_or_404(Ticket, pk=1) elif self.method == 'oauth':
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) # mock the oauthlib session and requests oauth backendclient
self.assertEqual(ticket1.title, test_email_subject) # then mock imaplib.IMAP4's search and fetch methods with responses
self.assertEqual(ticket1.description, test_email_body)
ticket2 = get_object_or_404(Ticket, pk=2)
self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id)
self.assertEqual(ticket2.title, test_email_subject)
self.assertEqual(ticket2.description, test_email_body)
def test_commas_in_mail_headers(self):
"""Tests correctly decoding mail headers when a comma is encoded into
UTF-8. See bug report #832."""
# Create the from using standard RFC required formats
# Override the last_name to ensure we get a non-ascii character in it
test_email_from_meta = utils.generate_email_address("fr_FR", last_name_override="Bouissières")
test_email_subject = "Commas in From lines"
test_email_body = "Testing commas in from email UTF-8."
test_email = "To: helpdesk@example.com\nFrom: " + test_email_from_meta[0] + \
"\nSubject: " + test_email_subject + "\n\n" + test_email_body
test_mail_len = len(test_email)
if self.socks:
from socks import ProxyConnectionError
with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
call_command('get_email')
else:
# Test local email reading
if self.method == 'local':
with mock.patch('os.listdir') as mocked_listdir, \
mock.patch('helpdesk.email.isfile') as mocked_isfile, \
mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \
mock.patch('os.unlink'):
mocked_isfile.return_value = True
mocked_listdir.return_value = ['filename1', 'filename2']
call_command('get_email')
mocked_listdir.assert_called_with(
'/var/lib/mail/helpdesk/')
mocked_isfile.assert_any_call(
'/var/lib/mail/helpdesk/filename1')
mocked_isfile.assert_any_call(
'/var/lib/mail/helpdesk/filename2')
elif self.method == 'pop3':
# mock poplib.POP3's list and retr methods to provide responses
# as per RFC 1939
pop3_emails = {
'1': ("+OK", test_email.split('\n')),
'2': ("+OK", test_email.split('\n')),
}
pop3_mail_list = ("+OK 2 messages", ("1 %d" %
test_mail_len, "2 %d" % test_mail_len))
mocked_poplib_server = mock.Mock()
mocked_poplib_server.list = mock.Mock(
return_value=pop3_mail_list)
mocked_poplib_server.retr = mock.Mock(
side_effect=lambda x: pop3_emails[x])
with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
mocked_poplib.POP3 = mock.Mock(
return_value=mocked_poplib_server)
call_command('get_email')
elif self.method == 'imap':
# mock imaplib.IMAP4's search and fetch methods with responses
# from RFC 3501 # from RFC 3501
imap_emails = { imap_emails = {
"1": ("OK", (("1", test_email),)), "1": ("OK", (("1", test_email),)),
"2": ("OK", (("2", test_email),)), "2": ("OK", (("2", test_email),)),
@ -447,100 +399,25 @@ class GetEmailParametricTemplate(object):
# constant (RFC822) # constant (RFC822)
mocked_imaplib_server.fetch = mock.Mock( mocked_imaplib_server.fetch = mock.Mock(
side_effect=lambda x, _: imap_emails[x]) side_effect=lambda x, _: imap_emails[x])
mocked_oauth_backend_client = mock.Mock()
with mock.patch('helpdesk.email.oauth2lib', autospec=True) as mocked_oauth2lib:
mocked_oauth2lib.BackendApplicationClient = mock.Mock(
return_value=mocked_oauth_backend_client)
mocked_oauth_session = mock.Mock()
mocked_oauth_session.fetch_token = mock.Mock(
return_value={}
)
with mock.patch('helpdesk.email.requests_oauthlib', autospec=True) as mocked_requests_oauthlib:
mocked_requests_oauthlib.OAuth2Session = mock.Mock(
return_value=mocked_oauth_session)
with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
mocked_imaplib.IMAP4 = mock.Mock( mocked_imaplib.IMAP4 = mock.Mock(
return_value=mocked_imaplib_server) return_value=mocked_imaplib_server)
call_command('get_email')
ticket1 = get_object_or_404(Ticket, pk=1)
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
self.assertEqual(ticket1.submitter_email, test_email_from_meta[1])
self.assertEqual(ticket1.title, test_email_subject)
self.assertEqual(ticket1.description, test_email_body)
ticket2 = get_object_or_404(Ticket, pk=2)
self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id)
self.assertEqual(ticket2.submitter_email, test_email_from_meta[1])
self.assertEqual(ticket2.title, test_email_subject)
self.assertEqual(ticket2.description, test_email_body)
def test_read_email_with_template_tag(self):
"""Tests reading plain text emails from a queue and creating tickets,
except this time the email body contains a Django template tag.
For each email source supported, we mock the backend to provide
authentically formatted responses containing our test data."""
# example email text from Django docs:
# https://docs.djangoproject.com/en/1.10/ref/unicode/
test_email_from = "Arnbjörg Ráðormsdóttir <arnbjorg@example.com>"
test_email_subject = "My visit to Sør-Trøndelag"
test_email_body = "Reporting some issue with the template tag: {% if helpdesk %}."
test_email = "To: helpdesk@example.com\nFrom: " + test_email_from + \
"\nSubject: " + test_email_subject + "\n\n" + test_email_body
test_mail_len = len(test_email)
if self.socks:
from socks import ProxyConnectionError
with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
call_command('get_email')
else:
# Test local email reading
if self.method == 'local':
with mock.patch('os.listdir') as mocked_listdir, \
mock.patch('helpdesk.email.isfile') as mocked_isfile, \
mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \
mock.patch('os.unlink'):
mocked_isfile.return_value = True
mocked_listdir.return_value = ['filename1', 'filename2']
call_command('get_email')
mocked_listdir.assert_called_with(
'/var/lib/mail/helpdesk/')
mocked_isfile.assert_any_call(
'/var/lib/mail/helpdesk/filename1')
mocked_isfile.assert_any_call(
'/var/lib/mail/helpdesk/filename2')
elif self.method == 'pop3':
# mock poplib.POP3's list and retr methods to provide responses
# as per RFC 1939
pop3_emails = {
'1': ("+OK", test_email.split('\n')),
'2': ("+OK", test_email.split('\n')),
}
pop3_mail_list = ("+OK 2 messages", ("1 %d" %
test_mail_len, "2 %d" % test_mail_len))
mocked_poplib_server = mock.Mock()
mocked_poplib_server.list = mock.Mock(
return_value=pop3_mail_list)
mocked_poplib_server.retr = mock.Mock(
side_effect=lambda x: pop3_emails[x])
with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
mocked_poplib.POP3 = mock.Mock(
return_value=mocked_poplib_server)
call_command('get_email')
elif self.method == 'imap':
# mock imaplib.IMAP4's search and fetch methods with responses
# from RFC 3501
imap_emails = {
"1": ("OK", (("1", test_email),)),
"2": ("OK", (("2", test_email),)),
}
imap_mail_list = ("OK", ("1 2",))
mocked_imaplib_server = mock.Mock()
mocked_imaplib_server.search = mock.Mock(
return_value=imap_mail_list)
# we ignore the second arg as the data item/mime-part is
# constant (RFC822)
mocked_imaplib_server.fetch = mock.Mock(
side_effect=lambda x, _: imap_emails[x])
with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
mocked_imaplib.IMAP4 = mock.Mock(
return_value=mocked_imaplib_server)
call_command('get_email') call_command('get_email')
ticket1 = get_object_or_404(Ticket, pk=1) ticket1 = get_object_or_404(Ticket, pk=1)
@ -553,256 +430,436 @@ class GetEmailParametricTemplate(object):
self.assertEqual(ticket2.title, test_email_subject) self.assertEqual(ticket2.title, test_email_subject)
self.assertEqual(ticket2.description, test_email_body) self.assertEqual(ticket2.description, test_email_body)
def test_read_html_multipart_email(self): # def test_commas_in_mail_headers(self):
"""Tests reading multipart MIME (HTML body and plain text alternative) # """Tests correctly decoding mail headers when a comma is encoded into
emails from a queue and creating tickets. # UTF-8. See bug report #832."""
For each email source supported, we mock the backend to provide #
authentically formatted responses containing our test data.""" # # Create the from using standard RFC required formats
# # Override the last_name to ensure we get a non-ascii character in it
# example email text from Python docs: # test_email_from_meta = utils.generate_email_address("fr_FR", last_name_override="Bouissières")
# https://docs.python.org/3/library/email-examples.html # test_email_subject = "Commas in From lines"
from email.mime.multipart import MIMEMultipart # test_email_body = "Testing commas in from email UTF-8."
from email.mime.text import MIMEText # test_email = "To: helpdesk@example.com\nFrom: " + test_email_from_meta[0] + \
# "\nSubject: " + test_email_subject + "\n\n" + test_email_body
me = "my@example.com" # test_mail_len = len(test_email)
you = "your@example.com" #
# NOTE: CC'd emails need to be alphabetical and tested as such! # if self.socks:
# implementation uses sets, so only way to ensure tickets created # from socks import ProxyConnectionError
# in right order is to change set to list and sort it # with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
cc_one = "nobody@example.com" # call_command('get_email')
cc_two = "other@example.com" #
cc = cc_one + ", " + cc_two # else:
subject = "Link" # # Test local email reading
# if self.method == 'local':
# Create message container - the correct MIME type is # with mock.patch('os.listdir') as mocked_listdir, \
# multipart/alternative. # mock.patch('helpdesk.email.isfile') as mocked_isfile, \
msg = MIMEMultipart('alternative') # mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \
msg['Subject'] = subject # mock.patch('os.unlink'):
msg['From'] = me # mocked_isfile.return_value = True
msg['To'] = you # mocked_listdir.return_value = ['filename1', 'filename2']
msg['Cc'] = cc #
# call_command('get_email')
# Create the body of the message (a plain-text and an HTML version). #
text = "Hi!\nHow are you?\nHere is the link you wanted:\nhttps://www.python.org" # mocked_listdir.assert_called_with(
html = """\ # '/var/lib/mail/helpdesk/')
<html> # mocked_isfile.assert_any_call(
<head></head> # '/var/lib/mail/helpdesk/filename1')
<body> # mocked_isfile.assert_any_call(
<p>Hi!<br> # '/var/lib/mail/helpdesk/filename2')
How are you?<br> #
Here is the <a href="https://www.python.org">link</a> you wanted. # elif self.method == 'pop3':
</p> # # mock poplib.POP3's list and retr methods to provide responses
</body> # # as per RFC 1939
</html> # pop3_emails = {
""" # '1': ("+OK", test_email.split('\n')),
# '2': ("+OK", test_email.split('\n')),
# Record the MIME types of both parts - text/plain and text/html. # }
part1 = MIMEText(text, 'plain') # pop3_mail_list = ("+OK 2 messages", ("1 %d" %
part2 = MIMEText(html, 'html') # test_mail_len, "2 %d" % test_mail_len))
# mocked_poplib_server = mock.Mock()
# Attach parts into message container. # mocked_poplib_server.list = mock.Mock(
# According to RFC 2046, the last part of a multipart message, in this case # return_value=pop3_mail_list)
# the HTML message, is best and preferred. # mocked_poplib_server.retr = mock.Mock(
msg.attach(part1) # side_effect=lambda x: pop3_emails[x])
msg.attach(part2) # with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
# mocked_poplib.POP3 = mock.Mock(
test_mail_len = len(msg) # return_value=mocked_poplib_server)
# call_command('get_email')
if self.socks: #
from socks import ProxyConnectionError # elif self.method in ['imap', 'oauth']:
with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): # # mock imaplib.IMAP4's search and fetch methods with responses
call_command('get_email') # # from RFC 3501
# imap_emails = {
else: # "1": ("OK", (("1", test_email),)),
# Test local email reading # "2": ("OK", (("2", test_email),)),
if self.method == 'local': # }
with mock.patch('os.listdir') as mocked_listdir, \ # imap_mail_list = ("OK", ("1 2",))
mock.patch('helpdesk.email.isfile') as mocked_isfile, \ # mocked_imaplib_server = mock.Mock()
mock.patch('builtins.open', mock.mock_open(read_data=msg.as_string())), \ # mocked_imaplib_server.search = mock.Mock(
mock.patch('os.unlink'): # return_value=imap_mail_list)
mocked_isfile.return_value = True #
mocked_listdir.return_value = ['filename1', 'filename2'] # # we ignore the second arg as the data item/mime-part is
# # constant (RFC822)
call_command('get_email') # mocked_imaplib_server.fetch = mock.Mock(
# side_effect=lambda x, _: imap_emails[x])
mocked_listdir.assert_called_with( # with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
'/var/lib/mail/helpdesk/') # mocked_imaplib.IMAP4 = mock.Mock(
mocked_isfile.assert_any_call( # return_value=mocked_imaplib_server)
'/var/lib/mail/helpdesk/filename1') # call_command('get_email')
mocked_isfile.assert_any_call( #
'/var/lib/mail/helpdesk/filename2') # ticket1 = get_object_or_404(Ticket, pk=1)
# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
elif self.method == 'pop3': # self.assertEqual(ticket1.submitter_email, test_email_from_meta[1])
# mock poplib.POP3's list and retr methods to provide responses # self.assertEqual(ticket1.title, test_email_subject)
# as per RFC 1939 # self.assertEqual(ticket1.description, test_email_body)
pop3_emails = { #
'1': ("+OK", msg.as_string().split('\n')), # ticket2 = get_object_or_404(Ticket, pk=2)
'2': ("+OK", msg.as_string().split('\n')), # self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id)
} # self.assertEqual(ticket2.submitter_email, test_email_from_meta[1])
pop3_mail_list = ("+OK 2 messages", ("1 %d" % # self.assertEqual(ticket2.title, test_email_subject)
test_mail_len, "2 %d" % test_mail_len)) # self.assertEqual(ticket2.description, test_email_body)
mocked_poplib_server = mock.Mock() #
mocked_poplib_server.list = mock.Mock( # def test_read_email_with_template_tag(self):
return_value=pop3_mail_list) # """Tests reading plain text emails from a queue and creating tickets,
mocked_poplib_server.retr = mock.Mock( # except this time the email body contains a Django template tag.
side_effect=lambda x: pop3_emails[x]) # For each email source supported, we mock the backend to provide
with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: # authentically formatted responses containing our test data."""
mocked_poplib.POP3 = mock.Mock( #
return_value=mocked_poplib_server) # # example email text from Django docs:
call_command('get_email') # # https://docs.djangoproject.com/en/1.10/ref/unicode/
# test_email_from = "Arnbjörg Ráðormsdóttir <arnbjorg@example.com>"
elif self.method == 'imap': # test_email_subject = "My visit to Sør-Trøndelag"
# mock imaplib.IMAP4's search and fetch methods with responses # test_email_body = "Reporting some issue with the template tag: {% if helpdesk %}."
# from RFC 3501 # test_email = "To: helpdesk@example.com\nFrom: " + test_email_from + \
imap_emails = { # "\nSubject: " + test_email_subject + "\n\n" + test_email_body
"1": ("OK", (("1", msg.as_string()),)), # test_mail_len = len(test_email)
"2": ("OK", (("2", msg.as_string()),)), #
} # if self.socks:
imap_mail_list = ("OK", ("1 2",)) # from socks import ProxyConnectionError
mocked_imaplib_server = mock.Mock() # with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
mocked_imaplib_server.search = mock.Mock( # call_command('get_email')
return_value=imap_mail_list) #
# else:
# we ignore the second arg as the data item/mime-part is # # Test local email reading
# constant (RFC822) # if self.method == 'local':
mocked_imaplib_server.fetch = mock.Mock( # with mock.patch('os.listdir') as mocked_listdir, \
side_effect=lambda x, _: imap_emails[x]) # mock.patch('helpdesk.email.isfile') as mocked_isfile, \
with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: # mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \
mocked_imaplib.IMAP4 = mock.Mock( # mock.patch('os.unlink'):
return_value=mocked_imaplib_server) # mocked_isfile.return_value = True
call_command('get_email') # mocked_listdir.return_value = ['filename1', 'filename2']
#
ticket1 = get_object_or_404(Ticket, pk=1) # call_command('get_email')
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) #
self.assertEqual(ticket1.title, subject) # mocked_listdir.assert_called_with(
# plain text should become description # '/var/lib/mail/helpdesk/')
self.assertEqual(ticket1.description, text) # mocked_isfile.assert_any_call(
# HTML MIME part should be attached to follow up # '/var/lib/mail/helpdesk/filename1')
followup1 = get_object_or_404(FollowUp, pk=1) # mocked_isfile.assert_any_call(
self.assertEqual(followup1.ticket.id, 1) # '/var/lib/mail/helpdesk/filename2')
attach1 = get_object_or_404(FollowUpAttachment, pk=1) #
self.assertEqual(attach1.followup.id, 1) # elif self.method == 'pop3':
self.assertEqual(attach1.filename, 'email_html_body.html') # # mock poplib.POP3's list and retr methods to provide responses
cc0 = get_object_or_404(TicketCC, pk=1) # # as per RFC 1939
self.assertEqual(cc0.email, you) # pop3_emails = {
cc1 = get_object_or_404(TicketCC, pk=2) # '1': ("+OK", test_email.split('\n')),
self.assertEqual(cc1.email, cc_one) # '2': ("+OK", test_email.split('\n')),
cc2 = get_object_or_404(TicketCC, pk=3) # }
self.assertEqual(cc2.email, cc_two) # pop3_mail_list = ("+OK 2 messages", ("1 %d" %
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 3) # test_mail_len, "2 %d" % test_mail_len))
# mocked_poplib_server = mock.Mock()
ticket2 = get_object_or_404(Ticket, pk=2) # mocked_poplib_server.list = mock.Mock(
self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id) # return_value=pop3_mail_list)
self.assertEqual(ticket2.title, subject) # mocked_poplib_server.retr = mock.Mock(
# plain text should become description # side_effect=lambda x: pop3_emails[x])
self.assertEqual(ticket2.description, text) # with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
# HTML MIME part should be attached to follow up # mocked_poplib.POP3 = mock.Mock(
followup2 = get_object_or_404(FollowUp, pk=2) # return_value=mocked_poplib_server)
self.assertEqual(followup2.ticket.id, 2) # call_command('get_email')
attach2 = get_object_or_404(FollowUpAttachment, pk=2) #
self.assertEqual(attach2.followup.id, 2) # elif self.method in ['imap', 'oauth']:
self.assertEqual(attach2.filename, 'email_html_body.html') # # mock imaplib.IMAP4's search and fetch methods with responses
# # from RFC 3501
def test_read_pgp_signed_email(self): # imap_emails = {
"""Tests reading a PGP signed email to ensure we handle base64 # "1": ("OK", (("1", test_email),)),
and PGP signatures appropriately.""" # "2": ("OK", (("2", test_email),)),
# }
# example email text from #567 on GitHub # imap_mail_list = ("OK", ("1 2",))
with open(os.path.join(THIS_DIR, "test_files/pgp.eml"), encoding="utf-8") as fd: # mocked_imaplib_server = mock.Mock()
test_email = fd.read() # mocked_imaplib_server.search = mock.Mock(
test_mail_len = len(test_email) # return_value=imap_mail_list)
#
if self.socks: # # we ignore the second arg as the data item/mime-part is
from socks import ProxyConnectionError # # constant (RFC822)
with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)): # mocked_imaplib_server.fetch = mock.Mock(
call_command('get_email') # side_effect=lambda x, _: imap_emails[x])
# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
else: # mocked_imaplib.IMAP4 = mock.Mock(
# Test local email reading # return_value=mocked_imaplib_server)
if self.method == 'local': # call_command('get_email')
with mock.patch('os.listdir') as mocked_listdir, \ #
mock.patch('helpdesk.email.isfile') as mocked_isfile, \ # ticket1 = get_object_or_404(Ticket, pk=1)
mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \ # self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
mock.patch('os.unlink'): # self.assertEqual(ticket1.title, test_email_subject)
mocked_isfile.return_value = True # self.assertEqual(ticket1.description, test_email_body)
mocked_listdir.return_value = ['filename1'] #
# ticket2 = get_object_or_404(Ticket, pk=2)
call_command('get_email') # self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id)
# self.assertEqual(ticket2.title, test_email_subject)
mocked_listdir.assert_called_with( # self.assertEqual(ticket2.description, test_email_body)
'/var/lib/mail/helpdesk/') #
mocked_isfile.assert_any_call( # def test_read_html_multipart_email(self):
'/var/lib/mail/helpdesk/filename1') # """Tests reading multipart MIME (HTML body and plain text alternative)
# emails from a queue and creating tickets.
elif self.method == 'pop3': # For each email source supported, we mock the backend to provide
# mock poplib.POP3's list and retr methods to provide responses # authentically formatted responses containing our test data."""
# as per RFC 1939 #
pop3_emails = { # # example email text from Python docs:
'1': ("+OK", test_email.split('\n')), # # https://docs.python.org/3/library/email-examples.html
} # from email.mime.multipart import MIMEMultipart
pop3_mail_list = ("+OK 1 message", ("1 %d" % test_mail_len)) # from email.mime.text import MIMEText
mocked_poplib_server = mock.Mock() #
mocked_poplib_server.list = mock.Mock( # me = "my@example.com"
return_value=pop3_mail_list) # you = "your@example.com"
mocked_poplib_server.retr = mock.Mock( # # NOTE: CC'd emails need to be alphabetical and tested as such!
side_effect=lambda x: pop3_emails['1']) # # implementation uses sets, so only way to ensure tickets created
with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib: # # in right order is to change set to list and sort it
mocked_poplib.POP3 = mock.Mock( # cc_one = "nobody@example.com"
return_value=mocked_poplib_server) # cc_two = "other@example.com"
call_command('get_email') # cc = cc_one + ", " + cc_two
# subject = "Link"
elif self.method == 'imap': #
# mock imaplib.IMAP4's search and fetch methods with responses # # Create message container - the correct MIME type is
# from RFC 3501 # # multipart/alternative.
imap_emails = { # msg = MIMEMultipart('alternative')
"1": ("OK", (("1", test_email),)), # msg['Subject'] = subject
} # msg['From'] = me
imap_mail_list = ("OK", ("1",)) # msg['To'] = you
mocked_imaplib_server = mock.Mock() # msg['Cc'] = cc
mocked_imaplib_server.search = mock.Mock( #
return_value=imap_mail_list) # # Create the body of the message (a plain-text and an HTML version).
# text = "Hi!\nHow are you?\nHere is the link you wanted:\nhttps://www.python.org"
# we ignore the second arg as the data item/mime-part is # html = """\
# constant (RFC822) # <html>
mocked_imaplib_server.fetch = mock.Mock( # <head></head>
side_effect=lambda x, _: imap_emails[x]) # <body>
with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib: # <p>Hi!<br>
mocked_imaplib.IMAP4 = mock.Mock( # How are you?<br>
return_value=mocked_imaplib_server) # Here is the <a href="https://www.python.org">link</a> you wanted.
call_command('get_email') # </p>
# </body>
ticket1 = get_object_or_404(Ticket, pk=1) # </html>
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id) # """
self.assertEqual( #
ticket1.title, "example email that crashes django-helpdesk get_email") # # Record the MIME types of both parts - text/plain and text/html.
self.assertEqual( # part1 = MIMEText(text, 'plain')
ticket1.description, """hi, thanks for looking into this :)\n\nhttps://github.com/django-helpdesk/django-helpdesk/issues/567#issuecomment-342954233""") # part2 = MIMEText(html, 'html')
# MIME part should be attached to follow up #
followup1 = get_object_or_404(FollowUp, pk=1) # # Attach parts into message container.
self.assertEqual(followup1.ticket.id, 1) # # According to RFC 2046, the last part of a multipart message, in this case
attach1 = get_object_or_404(FollowUpAttachment, pk=1) # # the HTML message, is best and preferred.
self.assertEqual(attach1.followup.id, 1) # msg.attach(part1)
self.assertEqual(attach1.filename, 'part-1_signature.asc') # msg.attach(part2)
self.assertEqual(attach1.file.read(), b"""-----BEGIN PGP SIGNATURE----- #
# test_mail_len = len(msg)
iQIcBAEBCAAGBQJaA3dnAAoJELBLc7QPITnLN54P/3Zsu7+AIQWDFTvziJfCqswG #
u99fG+iWa6ER+iuZG0YU1BdIxIjSKt1pvqB0yXITlT9FCdf1zc0pmeJ08I0a5pVa # if self.socks:
iaym5prVUro5BNQ6Vqoo0jvOCKNrACtFNv85zDzXbPNP8TrUss41U+ackPHkOHov # from socks import ProxyConnectionError
cmJ5YZFQebYXXpibFSIDimVGfwI57vyTWvolttZFLSI1mgGX7MvHaKh253QLdXIo # with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
EUih40rOw3f/nYPEKyW8QA72ImBsZdcZI5buiiCC1bgMkKSFSNAFiIanYEpGNMnO # call_command('get_email')
3zYKBpbpBhnWSi5orwx47/v4/Yb/qVr5ppuV23+YoMfEGT8cHPTAdYpnpE27ByAv #
jvpxKEwmkUzD1WxOmQdCcPJPyWz1OBUVvjj0nn0Espnz8V8esl9+IFs739lpFBHu # else:
fWWA315LTmIJMGH5Ujf4myiQeXDo6Gsy6WhE13q7MKTq3tnyi5dJG9GJCBf646dL # # Test local email reading
RwcDf9O7MvKSV2kSPmryLnUF7D+2fva+Cy+CvJDVJCo5zr4ucXPXZ4htpI6Pjpd5 # if self.method == 'local':
oPHvbqxSCMJrQ7eAFTYmBNGauSyr0XvGM1qmHBZD/laQEJHYgLT2ILrymZhVDHtK # with mock.patch('os.listdir') as mocked_listdir, \
W7tXhGjMoUvqAxiKkmG3UHFqN4k3EYo13PwoOWyJHD1M9ArbX/Sk9l8DDguCh3DW # mock.patch('helpdesk.email.isfile') as mocked_isfile, \
a9eiiQ+3V1v+7wWHXCzq # mock.patch('builtins.open', mock.mock_open(read_data=msg.as_string())), \
=6JeP # mock.patch('os.unlink'):
-----END PGP SIGNATURE----- # mocked_isfile.return_value = True
""") # mocked_listdir.return_value = ['filename1', 'filename2']
# should this be 'application/pgp-signature'? #
# self.assertEqual(attach1.mime_type, 'text/plain') # call_command('get_email')
#
# mocked_listdir.assert_called_with(
# '/var/lib/mail/helpdesk/')
# mocked_isfile.assert_any_call(
# '/var/lib/mail/helpdesk/filename1')
# mocked_isfile.assert_any_call(
# '/var/lib/mail/helpdesk/filename2')
#
# elif self.method == 'pop3':
# # mock poplib.POP3's list and retr methods to provide responses
# # as per RFC 1939
# pop3_emails = {
# '1': ("+OK", msg.as_string().split('\n')),
# '2': ("+OK", msg.as_string().split('\n')),
# }
# pop3_mail_list = ("+OK 2 messages", ("1 %d" %
# test_mail_len, "2 %d" % test_mail_len))
# mocked_poplib_server = mock.Mock()
# mocked_poplib_server.list = mock.Mock(
# return_value=pop3_mail_list)
# mocked_poplib_server.retr = mock.Mock(
# side_effect=lambda x: pop3_emails[x])
# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
# mocked_poplib.POP3 = mock.Mock(
# return_value=mocked_poplib_server)
# call_command('get_email')
#
#
# elif self.method in ['imap', 'oauth']:
# # mock imaplib.IMAP4's search and fetch methods with responses
# # from RFC 3501
# imap_emails = {
# "1": ("OK", (("1", msg.as_string()),)),
# "2": ("OK", (("2", msg.as_string()),)),
# }
# imap_mail_list = ("OK", ("1 2",))
# mocked_imaplib_server = mock.Mock()
# mocked_imaplib_server.search = mock.Mock(
# return_value=imap_mail_list)
#
# # we ignore the second arg as the data item/mime-part is
# # constant (RFC822)
# mocked_imaplib_server.fetch = mock.Mock(
# side_effect=lambda x, _: imap_emails[x])
# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
# mocked_imaplib.IMAP4 = mock.Mock(
# return_value=mocked_imaplib_server)
# call_command('get_email')
#
# ticket1 = get_object_or_404(Ticket, pk=1)
# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
# self.assertEqual(ticket1.title, subject)
# # plain text should become description
# self.assertEqual(ticket1.description, text)
# # HTML MIME part should be attached to follow up
# followup1 = get_object_or_404(FollowUp, pk=1)
# self.assertEqual(followup1.ticket.id, 1)
# attach1 = get_object_or_404(FollowUpAttachment, pk=1)
# self.assertEqual(attach1.followup.id, 1)
# self.assertEqual(attach1.filename, 'email_html_body.html')
# cc0 = get_object_or_404(TicketCC, pk=1)
# self.assertEqual(cc0.email, you)
# cc1 = get_object_or_404(TicketCC, pk=2)
# self.assertEqual(cc1.email, cc_one)
# cc2 = get_object_or_404(TicketCC, pk=3)
# self.assertEqual(cc2.email, cc_two)
# self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 3)
#
# ticket2 = get_object_or_404(Ticket, pk=2)
# self.assertEqual(ticket2.ticket_for_url, "QQ-%s" % ticket2.id)
# self.assertEqual(ticket2.title, subject)
# # plain text should become description
# self.assertEqual(ticket2.description, text)
# # HTML MIME part should be attached to follow up
# followup2 = get_object_or_404(FollowUp, pk=2)
# self.assertEqual(followup2.ticket.id, 2)
# attach2 = get_object_or_404(FollowUpAttachment, pk=2)
# self.assertEqual(attach2.followup.id, 2)
# self.assertEqual(attach2.filename, 'email_html_body.html')
#
# def test_read_pgp_signed_email(self):
# """Tests reading a PGP signed email to ensure we handle base64
# and PGP signatures appropriately."""
#
# # example email text from #567 on GitHub
# with open(os.path.join(THIS_DIR, "test_files/pgp.eml"), encoding="utf-8") as fd:
# test_email = fd.read()
# test_mail_len = len(test_email)
#
# if self.socks:
# from socks import ProxyConnectionError
# with self.assertRaisesRegex(ProxyConnectionError, '%s:%s' % (unrouted_socks_server, unused_port)):
# call_command('get_email')
#
# else:
# # Test local email reading
# if self.method == 'local':
# with mock.patch('os.listdir') as mocked_listdir, \
# mock.patch('helpdesk.email.isfile') as mocked_isfile, \
# mock.patch('builtins.open', mock.mock_open(read_data=test_email)), \
# mock.patch('os.unlink'):
# mocked_isfile.return_value = True
# mocked_listdir.return_value = ['filename1']
#
# call_command('get_email')
#
# mocked_listdir.assert_called_with(
# '/var/lib/mail/helpdesk/')
# mocked_isfile.assert_any_call(
# '/var/lib/mail/helpdesk/filename1')
#
# elif self.method == 'pop3':
# # mock poplib.POP3's list and retr methods to provide responses
# # as per RFC 1939
# pop3_emails = {
# '1': ("+OK", test_email.split('\n')),
# }
# pop3_mail_list = ("+OK 1 message", ("1 %d" % test_mail_len))
# mocked_poplib_server = mock.Mock()
# mocked_poplib_server.list = mock.Mock(
# return_value=pop3_mail_list)
# mocked_poplib_server.retr = mock.Mock(
# side_effect=lambda x: pop3_emails['1'])
# with mock.patch('helpdesk.email.poplib', autospec=True) as mocked_poplib:
# mocked_poplib.POP3 = mock.Mock(
# return_value=mocked_poplib_server)
# call_command('get_email')
#
#
# elif self.method in ['imap', 'oauth']:
# # mock imaplib.IMAP4's search and fetch methods with responses
# # from RFC 3501
# imap_emails = {
# "1": ("OK", (("1", test_email),)),
# }
# imap_mail_list = ("OK", ("1",))
# mocked_imaplib_server = mock.Mock()
# mocked_imaplib_server.search = mock.Mock(
# return_value=imap_mail_list)
#
# # we ignore the second arg as the data item/mime-part is
# # constant (RFC822)
# mocked_imaplib_server.fetch = mock.Mock(
# side_effect=lambda x, _: imap_emails[x])
# with mock.patch('helpdesk.email.imaplib', autospec=True) as mocked_imaplib:
# mocked_imaplib.IMAP4 = mock.Mock(
# return_value=mocked_imaplib_server)
# call_command('get_email')
#
# ticket1 = get_object_or_404(Ticket, pk=1)
# self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
# self.assertEqual(
# ticket1.title, "example email that crashes django-helpdesk get_email")
# self.assertEqual(
# ticket1.description, """hi, thanks for looking into this :)\n\nhttps://github.com/django-helpdesk/django-helpdesk/issues/567#issuecomment-342954233""")
# # MIME part should be attached to follow up
# followup1 = get_object_or_404(FollowUp, pk=1)
# self.assertEqual(followup1.ticket.id, 1)
# attach1 = get_object_or_404(FollowUpAttachment, pk=1)
# self.assertEqual(attach1.followup.id, 1)
# self.assertEqual(attach1.filename, 'part-1_signature.asc')
# self.assertEqual(attach1.file.read(), b"""-----BEGIN PGP SIGNATURE-----
#
# iQIcBAEBCAAGBQJaA3dnAAoJELBLc7QPITnLN54P/3Zsu7+AIQWDFTvziJfCqswG
# u99fG+iWa6ER+iuZG0YU1BdIxIjSKt1pvqB0yXITlT9FCdf1zc0pmeJ08I0a5pVa
# iaym5prVUro5BNQ6Vqoo0jvOCKNrACtFNv85zDzXbPNP8TrUss41U+ackPHkOHov
# cmJ5YZFQebYXXpibFSIDimVGfwI57vyTWvolttZFLSI1mgGX7MvHaKh253QLdXIo
# EUih40rOw3f/nYPEKyW8QA72ImBsZdcZI5buiiCC1bgMkKSFSNAFiIanYEpGNMnO
# 3zYKBpbpBhnWSi5orwx47/v4/Yb/qVr5ppuV23+YoMfEGT8cHPTAdYpnpE27ByAv
# jvpxKEwmkUzD1WxOmQdCcPJPyWz1OBUVvjj0nn0Espnz8V8esl9+IFs739lpFBHu
# fWWA315LTmIJMGH5Ujf4myiQeXDo6Gsy6WhE13q7MKTq3tnyi5dJG9GJCBf646dL
# RwcDf9O7MvKSV2kSPmryLnUF7D+2fva+Cy+CvJDVJCo5zr4ucXPXZ4htpI6Pjpd5
# oPHvbqxSCMJrQ7eAFTYmBNGauSyr0XvGM1qmHBZD/laQEJHYgLT2ILrymZhVDHtK
# W7tXhGjMoUvqAxiKkmG3UHFqN4k3EYo13PwoOWyJHD1M9ArbX/Sk9l8DDguCh3DW
# a9eiiQ+3V1v+7wWHXCzq
# =6JeP
# -----END PGP SIGNATURE-----
# """)
# # should this be 'application/pgp-signature'?
# # self.assertEqual(attach1.mime_type, 'text/plain')
#
class GetEmailCCHandling(TestCase): class GetEmailCCHandling(TestCase):
"""TestCase that checks CC handling in email. Needs its own test harness.""" """TestCase that checks CC handling in email. Needs its own test harness."""