forked from extern/django-helpdesk
Merge pull request #1088 from brucegibbins/oauth
Add IMAP OAUTH Mailbox Type
This commit is contained in:
commit
585f513f05
@ -28,10 +28,12 @@ from helpdesk.models import FollowUp, IgnoreEmail, Queue, Ticket
|
|||||||
import imaplib
|
import imaplib
|
||||||
import logging
|
import logging
|
||||||
import mimetypes
|
import mimetypes
|
||||||
|
import oauthlib.oauth2 as oauth2lib
|
||||||
import os
|
import os
|
||||||
from os.path import isfile, join
|
from os.path import isfile, join
|
||||||
import poplib
|
import poplib
|
||||||
import re
|
import re
|
||||||
|
import requests_oauthlib
|
||||||
import socket
|
import socket
|
||||||
import ssl
|
import ssl
|
||||||
import sys
|
import sys
|
||||||
@ -43,7 +45,6 @@ from typing import List, Tuple
|
|||||||
# import User model, which may be a custom model
|
# import User model, which may be a custom model
|
||||||
User = get_user_model()
|
User = get_user_model()
|
||||||
|
|
||||||
|
|
||||||
STRIPPED_SUBJECT_STRINGS = [
|
STRIPPED_SUBJECT_STRINGS = [
|
||||||
"Re: ",
|
"Re: ",
|
||||||
"Fw: ",
|
"Fw: ",
|
||||||
@ -225,6 +226,99 @@ def imap_sync(q, logger, server):
|
|||||||
server.logout()
|
server.logout()
|
||||||
|
|
||||||
|
|
||||||
|
def imap_oauth_sync(q, logger, server):
|
||||||
|
"""
|
||||||
|
IMAP eMail server with OAUTH authentication.
|
||||||
|
Only tested against O365 implementation
|
||||||
|
|
||||||
|
Uses HELPDESK OAUTH Dict in Settings.
|
||||||
|
|
||||||
|
"""
|
||||||
|
|
||||||
|
try:
|
||||||
|
logger.debug("Start Mailbox polling via IMAP OAUTH")
|
||||||
|
|
||||||
|
client = oauth2lib.BackendApplicationClient(
|
||||||
|
client_id=settings.HELPDESK_OAUTH["client_id"],
|
||||||
|
scope=settings.HELPDESK_OAUTH["scope"],
|
||||||
|
)
|
||||||
|
|
||||||
|
oauth = requests_oauthlib.OAuth2Session(client=client)
|
||||||
|
token = oauth.fetch_token(
|
||||||
|
token_url=settings.HELPDESK_OAUTH["token_url"],
|
||||||
|
client_id=settings.HELPDESK_OAUTH["client_id"],
|
||||||
|
client_secret=settings.HELPDESK_OAUTH["secret"],
|
||||||
|
include_client_id=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
server.debug = settings.HELPDESK_IMAP_DEBUG_LEVEL
|
||||||
|
|
||||||
|
# TODO: Perhaps store the authentication string template externally? Settings? Queue Table?
|
||||||
|
server.authenticate(
|
||||||
|
"XOAUTH2",
|
||||||
|
lambda x: f"user={q.email_box_user}\x01auth=Bearer {token['access_token']}\x01\x01".encode(),
|
||||||
|
)
|
||||||
|
|
||||||
|
# Select the Inbound Mailbox folder
|
||||||
|
server.select(q.email_box_imap_folder)
|
||||||
|
|
||||||
|
except imaplib.IMAP4.abort as e1:
|
||||||
|
logger.error(f"IMAP authentication failed in OAUTH: {e1}", exc_info=True)
|
||||||
|
server.logout()
|
||||||
|
sys.exit()
|
||||||
|
|
||||||
|
except ssl.SSLError as e2:
|
||||||
|
logger.error(
|
||||||
|
f"IMAP login failed due to SSL error. (This is often due to a timeout): {e2}", exc_info=True
|
||||||
|
)
|
||||||
|
server.logout()
|
||||||
|
sys.exit()
|
||||||
|
|
||||||
|
try:
|
||||||
|
data = server.search(None, 'NOT', 'DELETED')[1]
|
||||||
|
if data:
|
||||||
|
msgnums = data[0].split()
|
||||||
|
logger.info(f"Found {len(msgnums)} message(s) on IMAP server")
|
||||||
|
for num in msgnums:
|
||||||
|
logger.info(f"Processing message {num}")
|
||||||
|
data = server.fetch(num, '(RFC822)')[1]
|
||||||
|
full_message = encoding.force_str(data[0][1], errors='replace')
|
||||||
|
|
||||||
|
try:
|
||||||
|
ticket = object_from_message(message=full_message, queue=q, logger=logger)
|
||||||
|
|
||||||
|
except IgnoreTicketException as itex:
|
||||||
|
logger.warn(f"Message {num} was ignored. {itex}")
|
||||||
|
|
||||||
|
except DeleteIgnoredTicketException:
|
||||||
|
server.store(num, '+FLAGS', '\\Deleted')
|
||||||
|
logger.warn("Message %s was ignored and deleted from IMAP server" % num)
|
||||||
|
|
||||||
|
except TypeError as te:
|
||||||
|
# Log the error with stacktrace to help identify what went wrong
|
||||||
|
logger.error(f"Unexpected error processing message: {te}", exc_info=True)
|
||||||
|
|
||||||
|
else:
|
||||||
|
if ticket:
|
||||||
|
server.store(num, '+FLAGS', '\\Deleted')
|
||||||
|
logger.info(
|
||||||
|
"Successfully processed message %s, deleted from IMAP server" % num)
|
||||||
|
else:
|
||||||
|
logger.warn(
|
||||||
|
"Message %s was not successfully processed, and will be left on IMAP server" % num)
|
||||||
|
|
||||||
|
except imaplib.IMAP4.error:
|
||||||
|
logger.error(
|
||||||
|
"IMAP retrieve failed. Is the folder '%s' spelled correctly, and does it exist on the server?",
|
||||||
|
q.email_box_imap_folder
|
||||||
|
)
|
||||||
|
|
||||||
|
# Purged Flagged Messages & Logout
|
||||||
|
server.expunge()
|
||||||
|
server.close()
|
||||||
|
server.logout()
|
||||||
|
|
||||||
|
|
||||||
def process_queue(q, logger):
|
def process_queue(q, logger):
|
||||||
logger.info("***** %s: Begin processing mail for django-helpdesk" % ctime())
|
logger.info("***** %s: Begin processing mail for django-helpdesk" % ctime())
|
||||||
|
|
||||||
@ -272,7 +366,18 @@ def process_queue(q, logger):
|
|||||||
'init': imaplib.IMAP4,
|
'init': imaplib.IMAP4,
|
||||||
},
|
},
|
||||||
'sync': imap_sync
|
'sync': imap_sync
|
||||||
}
|
},
|
||||||
|
'oauth': {
|
||||||
|
'ssl': {
|
||||||
|
'port': 993,
|
||||||
|
'init': imaplib.IMAP4_SSL,
|
||||||
|
},
|
||||||
|
'insecure': {
|
||||||
|
'port': 143,
|
||||||
|
'init': imaplib.IMAP4,
|
||||||
|
},
|
||||||
|
'sync': imap_oauth_sync
|
||||||
|
},
|
||||||
}
|
}
|
||||||
if email_box_type in mail_defaults:
|
if email_box_type in mail_defaults:
|
||||||
encryption = 'insecure'
|
encryption = 'insecure'
|
||||||
@ -361,7 +466,6 @@ def is_autoreply(message):
|
|||||||
|
|
||||||
|
|
||||||
def create_ticket_cc(ticket, cc_list):
|
def create_ticket_cc(ticket, cc_list):
|
||||||
|
|
||||||
if not cc_list:
|
if not cc_list:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
@ -393,7 +497,6 @@ def create_ticket_cc(ticket, cc_list):
|
|||||||
|
|
||||||
|
|
||||||
def create_object_from_email_message(message, ticket_id, payload, files, logger):
|
def create_object_from_email_message(message, ticket_id, payload, files, logger):
|
||||||
|
|
||||||
ticket, previous_followup, new = None, None, False
|
ticket, previous_followup, new = None, None, False
|
||||||
now = timezone.now()
|
now = timezone.now()
|
||||||
|
|
||||||
@ -538,9 +641,9 @@ def send_info_email(message_id: str, f: FollowUp, ticket: Ticket, context: dict,
|
|||||||
|
|
||||||
|
|
||||||
def get_ticket_id_from_subject_slug(
|
def get_ticket_id_from_subject_slug(
|
||||||
queue_slug: str,
|
queue_slug: str,
|
||||||
subject: str,
|
subject: str,
|
||||||
logger: logging.Logger
|
logger: logging.Logger
|
||||||
) -> typing.Optional[int]:
|
) -> typing.Optional[int]:
|
||||||
"""Get a ticket id from the subject string
|
"""Get a ticket id from the subject string
|
||||||
|
|
||||||
@ -559,8 +662,8 @@ def get_ticket_id_from_subject_slug(
|
|||||||
|
|
||||||
|
|
||||||
def add_file_if_always_save_incoming_email_message(
|
def add_file_if_always_save_incoming_email_message(
|
||||||
files_,
|
files_,
|
||||||
message: str
|
message: str
|
||||||
) -> None:
|
) -> None:
|
||||||
"""When `settings.HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE` is `True`
|
"""When `settings.HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE` is `True`
|
||||||
add a file to the files_ list"""
|
add a file to the files_ list"""
|
||||||
|
18
helpdesk/migrations/0037_alter_queue_email_box_type.py
Normal file
18
helpdesk/migrations/0037_alter_queue_email_box_type.py
Normal file
@ -0,0 +1,18 @@
|
|||||||
|
# Generated by Django 3.2.16 on 2023-03-25 15:45
|
||||||
|
|
||||||
|
from django.db import migrations, models
|
||||||
|
|
||||||
|
|
||||||
|
class Migration(migrations.Migration):
|
||||||
|
|
||||||
|
dependencies = [
|
||||||
|
('helpdesk', '0036_add_attachment_validator'),
|
||||||
|
]
|
||||||
|
|
||||||
|
operations = [
|
||||||
|
migrations.AlterField(
|
||||||
|
model_name='queue',
|
||||||
|
name='email_box_type',
|
||||||
|
field=models.CharField(blank=True, choices=[('pop3', 'POP 3'), ('imap', 'IMAP'), ('oauth', 'IMAP OAUTH'), ('local', 'Local Directory')], help_text='E-Mail server type for creating tickets automatically from a mailbox - both POP3 and IMAP are supported, as well as reading from a local directory.', max_length=5, null=True, verbose_name='E-Mail Box Type'),
|
||||||
|
),
|
||||||
|
]
|
@ -175,7 +175,9 @@ class Queue(models.Model):
|
|||||||
email_box_type = models.CharField(
|
email_box_type = models.CharField(
|
||||||
_('E-Mail Box Type'),
|
_('E-Mail Box Type'),
|
||||||
max_length=5,
|
max_length=5,
|
||||||
choices=(('pop3', _('POP 3')), ('imap', _('IMAP')),
|
choices=(('pop3', _('POP 3')),
|
||||||
|
('imap', _('IMAP')),
|
||||||
|
('oauth', _('IMAP OAUTH')),
|
||||||
('local', _('Local Directory'))),
|
('local', _('Local Directory'))),
|
||||||
blank=True,
|
blank=True,
|
||||||
null=True,
|
null=True,
|
||||||
|
@ -249,3 +249,19 @@ HELPDESK_FULL_FIRST_MESSAGE_FROM_EMAIL = getattr(
|
|||||||
# (which gets stripped/corrupted otherwise)
|
# (which gets stripped/corrupted otherwise)
|
||||||
HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE = getattr(
|
HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE = getattr(
|
||||||
settings, "HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE", False)
|
settings, "HELPDESK_ALWAYS_SAVE_INCOMING_EMAIL_MESSAGE", False)
|
||||||
|
|
||||||
|
#######################
|
||||||
|
# email OAUTH #
|
||||||
|
#######################
|
||||||
|
|
||||||
|
HELPDESK_OAUTH = getattr(
|
||||||
|
settings, 'HELPDESK_OAUTH', {
|
||||||
|
"token_url": "",
|
||||||
|
"client_id": "",
|
||||||
|
"secret": "",
|
||||||
|
"scope": [""]
|
||||||
|
}
|
||||||
|
)
|
||||||
|
|
||||||
|
# Set Debug Logging Level for IMAP Services. Default to '0' for No Debugging
|
||||||
|
HELPDESK_IMAP_DEBUG_LEVEL = getattr(settings, 'HELPDESK_IMAP_DEBUG_LEVEL', 0)
|
||||||
|
@ -1,6 +1,5 @@
|
|||||||
# -*- coding: utf-8 -*-
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
|
||||||
from django.contrib.auth.hashers import make_password
|
from django.contrib.auth.hashers import make_password
|
||||||
from django.contrib.auth.models import User
|
from django.contrib.auth.models import User
|
||||||
from django.core import mail
|
from django.core import mail
|
||||||
@ -16,29 +15,31 @@ from helpdesk.models import Attachment, FollowUp, FollowUpAttachment, IgnoreEmai
|
|||||||
from helpdesk.tests import utils
|
from helpdesk.tests import utils
|
||||||
import itertools
|
import itertools
|
||||||
import logging
|
import logging
|
||||||
|
from oauthlib.oauth2 import BackendApplicationClient
|
||||||
import os
|
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
|
||||||
|
|
||||||
|
|
||||||
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
THIS_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||||
|
|
||||||
# class A addresses can't have first octet of 0
|
# class A addresses can't have first octet of 0
|
||||||
unrouted_socks_server = "0.0.0.1"
|
unrouted_socks_server = "0.0.0.1"
|
||||||
unrouted_email_server = "0.0.0.1"
|
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):
|
||||||
|
|
||||||
def setUp(self):
|
def setUp(self):
|
||||||
self.queue_public = Queue.objects.create(title='Test', slug='test')
|
self.queue_public = Queue.objects.create(title='Test', slug='test')
|
||||||
self.logger = logging.getLogger('helpdesk')
|
self.logger = logging.getLogger('helpdesk')
|
||||||
|
|
||||||
# tests correct syntax for command line option
|
# tests correct syntax for command line option
|
||||||
def test_get_email_quiet_option(self):
|
def test_get_email_quiet_option(self):
|
||||||
"""Test quiet option is properly propagated"""
|
"""Test quiet option is properly propagated"""
|
||||||
@ -60,7 +61,6 @@ class GetEmailCommonTests(TestCase):
|
|||||||
test_email = fd.read()
|
test_email = fd.read()
|
||||||
ticket = helpdesk.email.object_from_message(
|
ticket = helpdesk.email.object_from_message(
|
||||||
test_email, self.queue_public, self.logger)
|
test_email, self.queue_public, self.logger)
|
||||||
|
|
||||||
# title got truncated because of max_lengh of the model.title field
|
# title got truncated because of max_lengh of the model.title field
|
||||||
assert ticket.title == (
|
assert ticket.title == (
|
||||||
"Attachment without body - and a loooooooooooooooooooooooooooooooooo"
|
"Attachment without body - and a loooooooooooooooooooooooooooooooooo"
|
||||||
@ -283,6 +283,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)
|
||||||
@ -291,7 +302,6 @@ class GetEmailParametricTemplate(object):
|
|||||||
"""Tests reading plain text emails from a queue and creating tickets.
|
"""Tests reading plain text emails from a queue and creating tickets.
|
||||||
For each email source supported, we mock the backend to provide
|
For each email source supported, we mock the backend to provide
|
||||||
authentically formatted responses containing our test data."""
|
authentically formatted responses containing our test data."""
|
||||||
|
|
||||||
# example email text from Django docs:
|
# example email text from Django docs:
|
||||||
# https://docs.djangoproject.com/en/1.10/ref/unicode/
|
# https://docs.djangoproject.com/en/1.10/ref/unicode/
|
||||||
test_email_from = "Arnbjörg Ráðormsdóttir <arnbjorg@example.com>"
|
test_email_from = "Arnbjörg Ráðormsdóttir <arnbjorg@example.com>"
|
||||||
@ -355,7 +365,6 @@ class GetEmailParametricTemplate(object):
|
|||||||
mocked_imaplib_server = mock.Mock()
|
mocked_imaplib_server = mock.Mock()
|
||||||
mocked_imaplib_server.search = mock.Mock(
|
mocked_imaplib_server.search = mock.Mock(
|
||||||
return_value=imap_mail_list)
|
return_value=imap_mail_list)
|
||||||
|
|
||||||
# we ignore the second arg as the data item/mime-part is
|
# we ignore the second arg as the data item/mime-part is
|
||||||
# constant (RFC822)
|
# constant (RFC822)
|
||||||
mocked_imaplib_server.fetch = mock.Mock(
|
mocked_imaplib_server.fetch = mock.Mock(
|
||||||
@ -365,6 +374,43 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_imaplib_server)
|
return_value=mocked_imaplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
elif self.method == 'oauth':
|
||||||
|
# mock the oauthlib session and requests oauth backendclient
|
||||||
|
# then 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])
|
||||||
|
|
||||||
|
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:
|
||||||
|
mocked_imaplib.IMAP4 = mock.Mock(
|
||||||
|
return_value=mocked_imaplib_server)
|
||||||
|
|
||||||
|
call_command('get_email')
|
||||||
|
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
||||||
self.assertEqual(ticket1.title, test_email_subject)
|
self.assertEqual(ticket1.title, test_email_subject)
|
||||||
@ -452,6 +498,44 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_imaplib_server)
|
return_value=mocked_imaplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
elif self.method == 'oauth':
|
||||||
|
# mock the oauthlib session and requests oauth backendclient
|
||||||
|
# then 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])
|
||||||
|
|
||||||
|
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:
|
||||||
|
mocked_imaplib.IMAP4 = mock.Mock(
|
||||||
|
return_value=mocked_imaplib_server)
|
||||||
|
|
||||||
|
call_command('get_email')
|
||||||
|
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
||||||
self.assertEqual(ticket1.submitter_email, test_email_from_meta[1])
|
self.assertEqual(ticket1.submitter_email, test_email_from_meta[1])
|
||||||
@ -543,6 +627,44 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_imaplib_server)
|
return_value=mocked_imaplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
elif self.method == 'oauth':
|
||||||
|
# mock the oauthlib session and requests oauth backendclient
|
||||||
|
# then 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])
|
||||||
|
|
||||||
|
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:
|
||||||
|
mocked_imaplib.IMAP4 = mock.Mock(
|
||||||
|
return_value=mocked_imaplib_server)
|
||||||
|
|
||||||
|
call_command('get_email')
|
||||||
|
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
||||||
self.assertEqual(ticket1.title, test_email_subject)
|
self.assertEqual(ticket1.title, test_email_subject)
|
||||||
@ -651,6 +773,7 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_poplib_server)
|
return_value=mocked_poplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
|
||||||
elif self.method == 'imap':
|
elif self.method == 'imap':
|
||||||
# mock imaplib.IMAP4's search and fetch methods with responses
|
# mock imaplib.IMAP4's search and fetch methods with responses
|
||||||
# from RFC 3501
|
# from RFC 3501
|
||||||
@ -672,6 +795,44 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_imaplib_server)
|
return_value=mocked_imaplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
elif self.method == 'oauth':
|
||||||
|
# mock the oauthlib session and requests oauth backendclient
|
||||||
|
# then 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])
|
||||||
|
|
||||||
|
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:
|
||||||
|
mocked_imaplib.IMAP4 = mock.Mock(
|
||||||
|
return_value=mocked_imaplib_server)
|
||||||
|
|
||||||
|
call_command('get_email')
|
||||||
|
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
||||||
self.assertEqual(ticket1.title, subject)
|
self.assertEqual(ticket1.title, subject)
|
||||||
@ -751,6 +912,7 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_poplib_server)
|
return_value=mocked_poplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
|
||||||
elif self.method == 'imap':
|
elif self.method == 'imap':
|
||||||
# mock imaplib.IMAP4's search and fetch methods with responses
|
# mock imaplib.IMAP4's search and fetch methods with responses
|
||||||
# from RFC 3501
|
# from RFC 3501
|
||||||
@ -771,6 +933,43 @@ class GetEmailParametricTemplate(object):
|
|||||||
return_value=mocked_imaplib_server)
|
return_value=mocked_imaplib_server)
|
||||||
call_command('get_email')
|
call_command('get_email')
|
||||||
|
|
||||||
|
elif self.method == 'oauth':
|
||||||
|
# mock the oauthlib session and requests oauth backendclient
|
||||||
|
# then 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])
|
||||||
|
|
||||||
|
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:
|
||||||
|
mocked_imaplib.IMAP4 = mock.Mock(
|
||||||
|
return_value=mocked_imaplib_server)
|
||||||
|
|
||||||
|
call_command('get_email')
|
||||||
|
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
self.assertEqual(ticket1.ticket_for_url, "QQ-%s" % ticket1.id)
|
||||||
self.assertEqual(
|
self.assertEqual(
|
||||||
@ -876,7 +1075,6 @@ class GetEmailCCHandling(TestCase):
|
|||||||
def test_read_email_cc(self):
|
def test_read_email_cc(self):
|
||||||
"""Tests reading plain text emails from a queue and adding to a ticket,
|
"""Tests reading plain text emails from a queue and adding to a ticket,
|
||||||
particularly to test appropriate handling of CC'd emails."""
|
particularly to test appropriate handling of CC'd emails."""
|
||||||
|
|
||||||
# first, check that test ticket exists
|
# first, check that test ticket exists
|
||||||
ticket1 = get_object_or_404(Ticket, pk=1)
|
ticket1 = get_object_or_404(Ticket, pk=1)
|
||||||
self.assertEqual(ticket1.ticket_for_url, "CC-1")
|
self.assertEqual(ticket1.ticket_for_url, "CC-1")
|
||||||
@ -887,7 +1085,6 @@ class GetEmailCCHandling(TestCase):
|
|||||||
self.assertEqual(ccstaff.user, User.objects.get(username='staff'))
|
self.assertEqual(ccstaff.user, User.objects.get(username='staff'))
|
||||||
self.assertEqual(ticket1.assigned_to,
|
self.assertEqual(ticket1.assigned_to,
|
||||||
User.objects.get(username='assigned'))
|
User.objects.get(username='assigned'))
|
||||||
|
|
||||||
# example email text from Django docs:
|
# example email text from Django docs:
|
||||||
# https://docs.djangoproject.com/en/1.10/ref/unicode/
|
# https://docs.djangoproject.com/en/1.10/ref/unicode/
|
||||||
test_email_from = "submitter@example.com"
|
test_email_from = "submitter@example.com"
|
||||||
@ -918,7 +1115,6 @@ class GetEmailCCHandling(TestCase):
|
|||||||
|
|
||||||
mocked_listdir.assert_called_with('/var/lib/mail/helpdesk/')
|
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/filename1')
|
||||||
|
|
||||||
# 9 unique email addresses are CC'd when all is done
|
# 9 unique email addresses are CC'd when all is done
|
||||||
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 9)
|
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 9)
|
||||||
# next we make sure no duplicates were added, and the
|
# next we make sure no duplicates were added, and the
|
||||||
@ -942,16 +1138,12 @@ class GetEmailCCHandling(TestCase):
|
|||||||
cc9 = get_object_or_404(TicketCC, pk=9)
|
cc9 = get_object_or_404(TicketCC, pk=9)
|
||||||
self.assertEqual(cc9.user, User.objects.get(username='observer'))
|
self.assertEqual(cc9.user, User.objects.get(username='observer'))
|
||||||
self.assertEqual(cc9.email, "observer@example.com")
|
self.assertEqual(cc9.email, "observer@example.com")
|
||||||
|
|
||||||
|
|
||||||
# build matrix of test cases
|
# build matrix of test cases
|
||||||
case_methods = [c[0] for c in Queue._meta.get_field('email_box_type').choices]
|
case_methods = [c[0] for c in Queue._meta.get_field('email_box_type').choices]
|
||||||
|
|
||||||
# uncomment if you want to run tests with socks - which is much slover
|
# uncomment if you want to run tests with socks - which is much slover
|
||||||
# case_socks = [False] + [c[0] for c in Queue._meta.get_field('socks_proxy_type').choices]
|
# case_socks = [False] + [c[0] for c in Queue._meta.get_field('socks_proxy_type').choices]
|
||||||
case_socks = [False]
|
case_socks = [False]
|
||||||
case_matrix = list(itertools.product(case_methods, case_socks))
|
case_matrix = list(itertools.product(case_methods, case_socks))
|
||||||
|
|
||||||
# Populate TestCases from the matrix of parameters
|
# Populate TestCases from the matrix of parameters
|
||||||
thismodule = sys.modules[__name__]
|
thismodule = sys.modules[__name__]
|
||||||
for method, socks in case_matrix:
|
for method, socks in case_matrix:
|
||||||
|
@ -109,7 +109,9 @@ class QuickDjangoTest:
|
|||||||
HELPDESK_TEAMS_MIGRATION_DEPENDENCIES=[],
|
HELPDESK_TEAMS_MIGRATION_DEPENDENCIES=[],
|
||||||
HELPDESK_KBITEM_TEAM_GETTER=lambda _: None,
|
HELPDESK_KBITEM_TEAM_GETTER=lambda _: None,
|
||||||
# test the API
|
# test the API
|
||||||
HELPDESK_ACTIVATE_API_ENDPOINT=True
|
HELPDESK_ACTIVATE_API_ENDPOINT=True,
|
||||||
|
# Set IMAP Server Debug Verbosity
|
||||||
|
HELPDESK_IMAP_DEBUG_LEVEL=int(os.environ.get("HELPDESK_IMAP_DEBUG_LEVEL", "0")),
|
||||||
)
|
)
|
||||||
|
|
||||||
from django.test.runner import DiscoverRunner
|
from django.test.runner import DiscoverRunner
|
||||||
|
@ -1,6 +1,5 @@
|
|||||||
pysocks
|
pysocks
|
||||||
pycodestyle
|
pycodestyle
|
||||||
codecov
|
|
||||||
coverage
|
coverage
|
||||||
argparse
|
argparse
|
||||||
pbr
|
pbr
|
||||||
|
@ -12,3 +12,5 @@ pinax_teams
|
|||||||
djangorestframework
|
djangorestframework
|
||||||
django-model-utils
|
django-model-utils
|
||||||
django-cleanup
|
django-cleanup
|
||||||
|
oauthlib
|
||||||
|
requests_oauthlib
|
||||||
|
Loading…
Reference in New Issue
Block a user