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

Merge brunotikami/master for user and email threading, for #385 and #375

Browse Source
This commit is contained in:
Garret Wassermann 2018-12-28 10:53:28 -05:00
commit 5f0d22a692
9 changed files with 1163 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

287
helpdesk/email.py
View File

@ -1,16 +1,10 @@
#!/usr/bin/python
"""
Django Helpdesk - A Django powered ticket tracker for small enterprise.
(c) Copyright 2008 Jutda. Copyright 2018 Timothy Hobbs. All Rights Reserved.
See LICENSE for details.
scripts/get_email.py - Designed to be run from cron, this script checks the
POP and IMAP boxes, or a local mailbox directory,
defined for the queues within a
helpdesk, creating tickets from the new messages (or
adding to existing tickets if needed)
"""
from django.core.exceptions import ValidationError
from django.core.files.base import ContentFile
from django.core.files.uploadedfile import SimpleUploadedFile
from django.core.management.base import BaseCommand
@ -27,6 +21,8 @@ from datetime import timedelta
import base64
import binascii
import email
from email.header import decode_header
from email.utils import getaddresses, parseaddr, collapse_rfc2231_value
import imaplib
import mimetypes
from os import listdir, unlink
@ -37,6 +33,7 @@ import socket
import ssl
import sys
from time import ctime
from optparse import make_option
from bs4 import BeautifulSoup
@ -114,7 +111,7 @@ def pop3_sync(q, logger, server):
full_message = "\n".join([elm.decode('utf-8') for elm in raw_content])
else:
full_message = encoding.force_text("\n".join(raw_content), errors='replace')
ticket = ticket_from_message(message=full_message, queue=q, logger=logger)
ticket = object_from_message(message=full_message, queue=q, logger=logger)
if ticket:
server.dele(msgNum)
@ -153,7 +150,7 @@ def imap_sync(q, logger, server):
status, data = server.fetch(num, '(RFC822)')
full_message = encoding.force_text(data[0][1], errors='replace')
try:
ticket = ticket_from_message(message=full_message, queue=q, logger=logger)
ticket = object_from_message(message=full_message, queue=q, logger=logger)
except TypeError:
ticket = None # hotfix. Need to work out WHY.
if ticket:
@ -240,7 +237,7 @@ def process_queue(q, logger):
logger.info("Processing message %d" % i)
with open(m, 'r') as f:
full_message = encoding.force_text(f.read(), errors='replace')
ticket = ticket_from_message(message=full_message, queue=q, logger=logger)
ticket = object_from_message(message=full_message, queue=q, logger=logger)
if ticket:
logger.info("Successfully processed message %d, ticket/comment created." % i)
try:
@ -269,9 +266,157 @@ def decode_mail_headers(string):
return u' '.join([str(msg, encoding=charset, errors='replace') if charset else str(msg) for msg, charset in decoded])
def ticket_from_message(message, queue, logger):
def create_ticket_cc(ticket, cc_list):
if not cc_list:
return []
# Local import to deal with non-defined / circular reference problem
from helpdesk.views.staff import User, subscribe_to_ticket_updates
new_ticket_ccs = []
for cced_name, cced_email in cc_list:
cced_email = cced_email.strip()
if cced_email == ticket.queue.email_address:
continue
user = None
try:
user = User.objects.get(email=cced_email)
except User.DoesNotExist:
pass
try:
ticket_cc = subscribe_to_ticket_updates(ticket=ticket, user=user, email=cced_email)
new_ticket_ccs.append(ticket_cc)
except ValidationError as err:
pass
return new_ticket_ccs
def create_object_from_email_message(message, ticket_id, payload, files, logger):
ticket, previous_followup, new = None, None, False
now = timezone.now()
queue = payload['queue']
sender_email = payload['sender_email']
to_list = getaddresses(message.get_all('To', []))
cc_list = getaddresses(message.get_all('Cc', []))
message_id = message.get('Message-Id')
in_reply_to = message.get('In-Reply-To')
if in_reply_to is not None:
try:
queryset = FollowUp.objects.filter(message_id=in_reply_to).order_by('-date')
if queryset.count() > 0:
previous_followup = queryset.first()
ticket = previous_followup.ticket
except FollowUp.DoesNotExist:
pass #play along. The header may be wrong
if previous_followup is None and ticket_id is not None:
try:
ticket = Ticket.objects.get(id=ticket_id)
new = False
except Ticket.DoesNotExist:
ticket = None
# New issue, create a new <Ticket> instance
if ticket is None:
if not settings.QUEUE_EMAIL_BOX_UPDATE_ONLY:
ticket = Ticket.objects.create(
title = payload['subject'],
queue = queue,
submitter_email = sender_email,
created = now,
description = payload['body'],
priority = payload['priority'],
)
ticket.save()
logger.debug("Created new ticket %s-%s" % (ticket.queue.slug, ticket.id))
new = True
update = ''
# Old issue being re-opened
elif ticket.status == Ticket.CLOSED_STATUS:
ticket.status = Ticket.REOPENED_STATUS
ticket.save()
f = FollowUp(
ticket = ticket,
title = _('E-Mail Received from %(sender_email)s' % {'sender_email': sender_email}),
date = now,
public = True,
comment = payload['body'],
message_id = message_id,
)
if ticket.status == Ticket.REOPENED_STATUS:
f.new_status = Ticket.REOPENED_STATUS
f.title = _('Ticket Re-Opened by E-Mail Received from %(sender_email)s' % {'sender_email': sender_email})
f.save()
logger.debug("Created new FollowUp for Ticket")
logger.info("[%s-%s] %s" % (ticket.queue.slug, ticket.id, ticket.title,))
attached = process_attachments(f, files)
for att_file in attached:
logger.info("Attachment '%s' (with size %s) successfully added to ticket from email." % (att_file[0], att_file[1].size))
context = safe_template_context(ticket)
new_ticket_ccs = []
new_ticket_ccs.append(create_ticket_cc(ticket, to_list + cc_list))
notifications_to_be_sent = [sender_email,]
if queue.enable_notifications_on_email_events and len(notifications_to_be_sent):
ticket_cc_list = TicketCC.objects.filter(ticket=ticket).all().values_list('email', flat=True)
for email in ticket_cc_list :
notifications_to_be_sent.append(email)
# send mail to appropriate people now depending on what objects
# were created and who was CC'd
if new:
ticket.send(
{'submitter': ('newticket_submitter', context),
'new_ticket_cc': ('newticket_cc', context),
'ticket_cc': ('newticket_cc', context)},
fail_silently=True,
extra_headers={'In-Reply-To': message_id},
)
else:
context.update(comment=f.comment)
ticket.send(
{'submitter': ('newticket_submitter', context),
'assigned_to': ('updated_owner', context),},
fail_silently=True,
extra_headers={'In-Reply-To': message_id},
)
if queue.enable_notifications_on_email_events:
ticket.send(
{'ticket_cc': ('updated_cc', context),},
fail_silently=True,
extra_headers={'In-Reply-To': message_id},
)
return ticket
def object_from_message(message, queue, logger):
# 'message' must be an RFC822 formatted message.
message = email.message_from_string(message)
subject = message.get('subject', _('Comment from e-mail'))
subject = decode_mail_headers(decodeUnknown(message.get_charset(), subject))
for affix in STRIPPED_SUBJECT_STRINGS:
@ -281,7 +426,9 @@ def ticket_from_message(message, queue, logger):
sender = message.get('from', _('Unknown Sender'))
sender = decode_mail_headers(decodeUnknown(message.get_charset(), sender))
sender_email = email.utils.parseaddr(sender)[1]
body_plain, body_html = '', ''
cc = message.get_all('cc', None)
if cc:
# first, fixup the encoding if necessary
@ -370,113 +517,19 @@ def ticket_from_message(message, queue, logger):
if not body:
body = mail.text
if ticket:
try:
t = Ticket.objects.get(id=ticket)
except Ticket.DoesNotExist:
logger.info("Tracking ID %s-%s not associated with existing ticket. Creating new ticket." % (queue.slug, ticket))
ticket = None
else:
logger.info("Found existing ticket with Tracking ID %s-%s" % (t.queue.slug, t.id))
if t.status == Ticket.CLOSED_STATUS:
t.status = Ticket.REOPENED_STATUS
t.save()
new = False
smtp_priority = message.get('priority', '')
smtp_importance = message.get('importance', '')
high_priority_types = {'high', 'important', '1', 'urgent'}
priority = 2 if high_priority_types & {smtp_priority, smtp_importance} else 3
payload = {
'body': body,
'subject': subject,
'queue': queue,
'sender_email': sender_email,
'priority': priority,
'files': files,
}
return create_object_from_email_message(message, ticket, payload, files, logger=logger)
if ticket is None:
if settings.QUEUE_EMAIL_BOX_UPDATE_ONLY:
return None
new = True
t = Ticket.objects.create(
title=subject,
queue=queue,
submitter_email=sender_email,
created=timezone.now(),
description=body,
priority=priority,
)
logger.debug("Created new ticket %s-%s" % (t.queue.slug, t.id))
if cc:
# get list of currently CC'd emails
current_cc = TicketCC.objects.filter(ticket=ticket)
current_cc_emails = [x.email for x in current_cc if x.email]
# get emails of any Users CC'd to email, if defined
# (some Users may not have an associated email, e.g, when using LDAP)
current_cc_users = [x.user.email for x in current_cc if x.user and x.user.email]
# ensure submitter, assigned user, queue email not added
other_emails = [queue.email_address]
if t.submitter_email:
other_emails.append(t.submitter_email)
if t.assigned_to:
other_emails.append(t.assigned_to.email)
current_cc = set(current_cc_emails + current_cc_users + other_emails)
# first, add any User not previously CC'd (as identified by User's email)
all_users = User.objects.all()
all_user_emails = set([x.email for x in all_users])
users_not_currently_ccd = all_user_emails.difference(set(current_cc))
users_to_cc = cc.intersection(users_not_currently_ccd)
for user in users_to_cc:
tcc = TicketCC.objects.create(
ticket=t,
user=User.objects.get(email=user),
can_view=True,
can_update=False
)
tcc.save()
# then add remaining emails alphabetically, makes testing easy
new_cc = cc.difference(current_cc).difference(all_user_emails)
new_cc = sorted(list(new_cc))
for ccemail in new_cc:
tcc = TicketCC.objects.create(
ticket=t,
email=ccemail.replace('\n', ' ').replace('\r', ' '),
can_view=True,
can_update=False
)
tcc.save()
f = FollowUp(
ticket=t,
title=_('E-Mail Received from %(sender_email)s' % {'sender_email': sender_email}),
date=timezone.now(),
public=True,
comment=body,
)
if t.status == Ticket.REOPENED_STATUS:
f.new_status = Ticket.REOPENED_STATUS
f.title = _('Ticket Re-Opened by E-Mail Received from %(sender_email)s' % {'sender_email': sender_email})
f.save()
logger.debug("Created new FollowUp for Ticket")
logger.info("[%s-%s] %s" % (t.queue.slug, t.id, t.title,))
attached = process_attachments(f, files)
for att_file in attached:
logger.info("Attachment '%s' (with size %s) successfully added to ticket from email." % (att_file[0], att_file[1].size))
context = safe_template_context(t)
if new:
t.send(
{'submitter': ('newticket_submitter', context),
'new_ticket_cc': ('newticket_cc', context),
'ticket_cc': ('newticket_cc', context)},
fail_silently=True,
)
else:
context.update(comment=f.comment)
t.send(
{'assigned_to': ('updated_owner', context),
'ticket_cc': ('updated_cc', context)},
fail_silently=True,
)
return t

2
helpdesk/lib.py
View File

@ -12,7 +12,7 @@ import os
from django.conf import settings
from django.db.models import Q
from django.utils.encoding import smart_text
from django.utils.encoding import smart_text, smart_str
from django.utils.safestring import mark_safe
from helpdesk.models import Attachment, EmailTemplate

18
helpdesk/migrations/0022_add_submitter_email_id_field_to_ticket.py Normal file
View File

@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.9.1 on 2016-02-07 19:51
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('helpdesk', '0021_voting_tracker'),
]
operations = [
migrations.AddField(
model_name='followup',
name='message_id',
field=models.CharField(blank=True, editable=False, help_text="The Message ID of the submitter's email.", max_length=256, null=True, verbose_name='E-Mail ID'),
),
]

18
helpdesk/migrations/0023_add_enable_notifications_on_email_events_to_ticket.py Normal file
View File

@ -0,0 +1,18 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.9.1 on 2016-03-01 19:43
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('helpdesk', '0022_add_submitter_email_id_field_to_ticket'),
]
operations = [
migrations.AddField(
model_name='queue',
name='enable_notifications_on_email_events',
field=models.BooleanField(default=False, help_text='When an email arrives to either create a ticket or to interact with an existing discussion. Should email notifications be sent ? Note: the new_ticket_cc and updated_ticket_cc work independently of this feature', verbose_name='Notify contacts when email updates arrive'),
),
]

27
helpdesk/models.py
View File

@ -108,6 +108,15 @@ class Queue(models.Model):
'multiple addresses with a comma.'),
)
enable_notifications_on_email_events = models.BooleanField(
_('Notify contacts when email updates arrive'),
blank=True,
default=False,
help_text=_('When an email arrives to either create a ticket or to '
'interact with an existing discussion. Should email notifications be sent ? '
'Note: the new_ticket_cc and updated_ticket_cc work independently of this feature'),
)
email_box_type = models.CharField(
_('E-Mail Box Type'),
max_length=5,
@ -493,7 +502,7 @@ class Ticket(models.Model):
Send notifications to everyone interested in this ticket.
The the roles argument is a dictionary mapping from roles to (template, context) pairs.
If a role is not present in the dictionary, users of that type will not recieve the notification.
If a role is not present in the dictionary, users of that type will not receive the notification.
The following roles exist:
@ -528,12 +537,13 @@ class Ticket(models.Model):
send_templated_mail(template, context, recipient, sender=self.queue.from_address, **kwargs)
recipients.add(recipient)
send('submitter', self.submitter_email)
send('ticket_cc', self.queue.updated_ticket_cc)
send('new_ticket_cc', self.queue.new_ticket_cc)
if self.assigned_to:
send('assigned_to', self.assigned_to.email)
send('ticket_cc', self.queue.updated_ticket_cc)
for cc in self.ticketcc_set.all():
send('ticket_cc', cc.email_address)
if self.queue.enable_notifications_on_email_events:
for cc in self.ticketcc_set.all():
send('ticket_cc', cc.email_address)
return recipients
def _get_assigned_to(self):
@ -750,6 +760,15 @@ class FollowUp(models.Model):
help_text=_('If the status was changed, what was it changed to?'),
)
message_id = models.CharField(
_('E-Mail ID'),
max_length=256,
blank=True,
null=True,
help_text=_("The Message ID of the submitter's email."),
editable=False,
)
objects = FollowUpManager()
class Meta:

6
helpdesk/templated_email.py
View File

@ -14,7 +14,8 @@ def send_templated_mail(template_name,
sender=None,
bcc=None,
fail_silently=False,
files=None):
files=None,
extra_headers={}):
"""
send_templated_mail() is a wrapper around Django's e-mail routines that
allows us to easily send multipart (text/plain & text/html) e-mails using
@ -39,6 +40,9 @@ def send_templated_mail(template_name,
files can be a list of tuples. Each tuple should be a filename to attach,
along with the File objects to be read. files can be blank.
extra_headers is a dictionary of extra email headers, needed to process
email replies and keep proper threading.
"""
from django.core.mail import EmailMultiAlternatives

46
helpdesk/tests/test_get_email.py
View File

@ -328,11 +328,13 @@ class GetEmailParametricTemplate(object):
attach1 = get_object_or_404(Attachment, pk=1)
self.assertEqual(attach1.followup.id, 1)
self.assertEqual(attach1.filename, 'email_html_body.html')
cc1 = get_object_or_404(TicketCC, pk=1)
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=2)
cc2 = get_object_or_404(TicketCC, pk=3)
self.assertEqual(cc2.email, cc_two)
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 2)
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)
@ -704,25 +706,29 @@ class GetEmailCCHandling(TestCase):
mocked_listdir.assert_called_with('/var/lib/mail/helpdesk/')
mocked_isfile.assert_any_call('/var/lib/mail/helpdesk/filename1')
# ensure these 4 CCs (test_email_cc one thru four) are the only ones
# created and added to the existing staff_user that was CC'd,
# and the observer user that gets CC'd to new email.,
# and that submitter and assignee are not added as CC either
# (in other words, even though everyone was CC'd to this email,
# we should come out with only 6 CCs after filtering)
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 6)
# 9 unique email addresses are CC'd when all is done
self.assertEqual(len(TicketCC.objects.filter(ticket=1)), 9)
# next we make sure no duplicates were added, and the
# staff users nor submitter were not re-added as email TicketCCs
cc0 = get_object_or_404(TicketCC, pk=2)
self.assertEqual(cc0.user, User.objects.get(username='observer'))
cc1 = get_object_or_404(TicketCC, pk=3)
self.assertEqual(cc1.email, test_email_cc_one)
cc2 = get_object_or_404(TicketCC, pk=4)
self.assertEqual(cc2.email, test_email_cc_two)
cc3 = get_object_or_404(TicketCC, pk=5)
self.assertEqual(cc3.email, test_email_cc_three)
cc4 = get_object_or_404(TicketCC, pk=6)
self.assertEqual(cc4.email, test_email_cc_four)
cc1 = get_object_or_404(TicketCC, pk=1)
self.assertEqual(cc1.user, User.objects.get(username='staff'))
cc2 = get_object_or_404(TicketCC, pk=2)
self.assertEqual(cc2.email, "alice@example.com")
cc3 = get_object_or_404(TicketCC, pk=3)
self.assertEqual(cc3.email, test_email_cc_two)
cc4 = get_object_or_404(TicketCC, pk=4)
self.assertEqual(cc4.email, test_email_cc_three)
cc5 = get_object_or_404(TicketCC, pk=5)
self.assertEqual(cc5.email, test_email_cc_four)
cc6 = get_object_or_404(TicketCC, pk=6)
self.assertEqual(cc6.email, "assigned@example.com")
cc7 = get_object_or_404(TicketCC, pk=7)
self.assertEqual(cc7.email, "staff@example.com")
cc8 = get_object_or_404(TicketCC, pk=8)
self.assertEqual(cc8.email, "submitter@example.com")
cc9 = get_object_or_404(TicketCC, pk=9)
self.assertEqual(cc9.user, User.objects.get(username='observer'))
self.assertEqual(cc9.email, "observer@example.com")
# build matrix of test cases

879
helpdesk/tests/test_ticket_submission.py
View File

@ -1,13 +1,23 @@
from helpdesk.models import Queue, CustomField, Ticket
import email
import uuid
from helpdesk.models import Queue, CustomField, FollowUp, Ticket, TicketCC
from django.test import TestCase
from django.core import mail
from django.core.exceptions import ObjectDoesNotExist
from django.forms import ValidationError
from django.test.client import Client
from django.urls import reverse
try: # python 3
from urllib.parse import urlparse
except ImportError: # python 2
from urlparse import urlparse
from helpdesk.email import object_from_message, create_ticket_cc
from urllib.parse import urlparse
import logging
logger = logging.getLogger('helpdesk')
class TicketBasicsTestCase(TestCase):
@ -34,7 +44,12 @@ class TicketBasicsTestCase(TestCase):
self.client = Client()
def test_create_ticket_direct(self):
def test_create_ticket_instance_from_payload(self):
"""
Ensure that a <Ticket> instance is created whenever an email is sent to a public queue.
"""
email_count = len(mail.outbox)
ticket_data = dict(queue=self.queue_public, **self.ticket_data)
ticket = Ticket.objects.create(**ticket_data)
@ -123,3 +138,855 @@ class TicketBasicsTestCase(TestCase):
# Ensure only two e-mails were sent - submitter & updated.
self.assertEqual(email_count + 2, len(mail.outbox))
class EmailInteractionsTestCase(TestCase):
fixtures = ['emailtemplate.json']
def setUp(self):
self.queue_public = Queue.objects.create(title='Mail Queue 1',
slug='mq1',
email_address='queue-1@example.com',
allow_public_submission=True,
new_ticket_cc='new.public.with.notifications@example.com',
updated_ticket_cc='update.public.with.notifications@example.com',
enable_notifications_on_email_events=True,
)
self.queue_public_with_notifications_disabled = Queue.objects.create(title='Mail Queue 2',
slug='mq2',
email_address='queue-2@example.com',
allow_public_submission=True,
new_ticket_cc='new.public.without.notifications@example.com',
updated_ticket_cc='update.public.without.notifications@example.com',
enable_notifications_on_email_events=False,
)
self.ticket_data = {
'title': 'Test Ticket',
'description': 'Some Test Ticket',
}
def test_create_ticket_from_email_with_message_id(self):
"""
Ensure that a <Ticket> instance is created whenever an email is sent to a public queue.
Also, make sure that the RFC 2822 field "message-id" is stored on the <Ticket.submitter_email_id>
field.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify the sender (+1)
# and the new and update queues (+2)
self.assertEqual(email_count + 1 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
def test_create_ticket_from_email_without_message_id(self):
"""
Ensure that a <Ticket> instance is created whenever an email is sent to a public queue.
Also, make sure that the RFC 2822 field "message-id" is stored on the <Ticket.submitter_email_id>
field.
"""
msg = email.message.Message()
submitter_email = 'foo@bar.py'
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
ticket = Ticket.objects.get(title=self.ticket_data['title'], queue=self.queue_public, submitter_email=submitter_email)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify the sender (+1)
# and the new and update queues (+2)
self.assertEqual(email_count + 1 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
def test_create_ticket_from_email_with_carbon_copy(self):
"""
Ensure that an instance of <TicketCC> is created for every valid element of the
"rfc_2822_cc" field when creating a <Ticket> instance.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
self.assertEqual(email_count + 1 + 2 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
# Ensure that <TicketCC> exists
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_ticket_from_email_to_multiple_emails(self):
"""
Ensure that an instance of <TicketCC> is created for every valid element of the
"rfc_2822_cc" field when creating a <Ticket> instance.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
to_list = [self.queue_public.email_address]
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', ','.join(to_list + cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
self.assertEqual(email_count + 1 + 2 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
# Ensure that the queue's email was not subscribed to the event notifications.
self.assertRaises(TicketCC.DoesNotExist, TicketCC.objects.get, ticket=ticket, email=to_list[0])
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
# Ensure that <TicketCC> exists
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_ticket_from_email_with_invalid_carbon_copy(self):
"""
Ensure that no <TicketCC> instance is created if an invalid element of the
"rfc_2822_cc" field is provided when creating a <Ticket> instance.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['null@example', 'invalid@foobar']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the submitter (+1)
# contacts on the cc_list (+2),
# the new and update queues (+2)
self.assertEqual(email_count + 1 + 2 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email, mail.outbox[0].to)
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
# Ensure that <TicketCC> exists. Even if it's an invalid email.
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_followup_from_email_with_valid_message_id_with_no_initial_cc_list(self):
"""
Ensure that if a message is received with an valid In-Reply-To ID,
the expected <TicketCC> instances are created even if the there were
no <TicketCC>s so far.
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
# As we have created an Ticket from an email, we notify the sender
# and contacts on the cc_list (+1 as it's treated as a list),
# the new and update queues (+2)
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
# As we have created an Ticket from an email, we notify the sender (+1)
# and the new and update queues (+2)
expected_email_count = 1 + 2
self.assertEqual(expected_email_count, len(mail.outbox))
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', message_id)
reply.__setitem__('Subject', self.ticket_data['title'])
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public.email_address)
reply.__setitem__('Cc', ','.join(cc_list))
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
object_from_message(str(reply), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Even after 2 messages with the same cc_list, <get> MUST return only
# one object
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
# As an update was made, we increase the expected_email_count with:
# submitter: +1
# cc_list: +2
# public_update_queue: +1
expected_email_count += 1 + 2 + 1
self.assertEqual(expected_email_count, len(mail.outbox))
# As we have created a FollowUp from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
# Ensure that the submitter is notified
#self.assertIn(submitter_email, mail.outbox[expected_email_count - 3].to)
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#for cc_email in cc_list:
#self.assertIn(cc_email, mail.outbox[expected_email_count - 1].to)
def test_create_followup_from_email_with_valid_message_id_with_original_cc_list_included(self):
"""
Ensure that if a message is received with an valid In-Reply-To ID,
the expected <TicketCC> instances are created but if there's any
overlap with the previous Cc list, no duplicates are created.
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
self.assertTrue(ticket_cc.can_view, True)
# As we have created a Ticket from an email, we notify the sender
# and contacts on the cc_list (+1 as it's treated as a list),
# the new and update queues (+2)
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
# As we have created an Ticket from an email, we notify the sender (+1)
# and the new and update queues (+2)
expected_email_count = 1 + 2
self.assertEqual(expected_email_count, len(mail.outbox))
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', message_id)
reply.__setitem__('Subject', self.ticket_data['title'])
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public.email_address)
reply.__setitem__('Cc', ','.join(cc_list))
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
object_from_message(str(reply), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As an update was made, we increase the expected_email_count with:
# public_update_queue: +1
expected_email_count += 1
self.assertEqual(expected_email_count, len(mail.outbox))
# As we have created a FollowUp from an email, we notify the sender
# and contacts on the cc_list (+1 as it's treated as a list),
# the new and update queues (+2)
# Ensure that the submitter is notified
self.assertIn(submitter_email, mail.outbox[expected_email_count - 1].to)
# Ensure that contacts on cc_list will be notified on the same email (index 0)
for cc_email in cc_list:
self.assertIn(cc_email, mail.outbox[expected_email_count - 1].to)
# Even after 2 messages with the same cc_list,
# <get> MUST return only one object
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_followup_from_email_with_invalid_message_id(self):
"""
Ensure that if a message is received with an invalid In-Reply-To ID and we
can infer the original Ticket ID by the message's subject, the expected
<TicketCC> instances are created
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
self.assertTrue(ticket_cc.can_view, True)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
expected_email_count = 1 + 2 + 2
self.assertEqual(expected_email_count, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email,mail.outbox[0].to)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
invalid_message_id = 'INVALID'
reply_subject = 'Re: ' + self.ticket_data['title']
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', invalid_message_id)
reply.__setitem__('Subject', reply_subject)
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public.email_address)
reply.__setitem__('Cc', ','.join(cc_list))
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(reply), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Even after 2 messages with the same cc_list, <get> MUST return only
# one object
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
self.assertEqual(email_count + 1 + 2 + 2, len(mail.outbox))
def test_create_ticket_from_email_to_a_notification_enabled_queue(self):
"""
Ensure that when an email is sent to a Queue with notifications_enabled turned ON,
and a <Ticket> is created, all contacts n the TicketCC list are notified.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
self.assertEqual(email_count + 1 + 2 + 2, len(mail.outbox))
# Ensure that the submitter is notified
self.assertIn(submitter_email, mail.outbox[0].to)
# Ensure that <TicketCC> exist
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_ticket_from_email_to_a_notification_disabled_queue(self):
"""
Ensure that when an email is sent to a Queue with notifications_enabled turned OFF, only the
new_ticket_cc and updated_ticket_cc contacts (if they are set) are notified. No contact
from the TicketCC list should be notified.
"""
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public_with_notifications_disabled.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public_with_notifications_disabled, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq2-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# the new and update queues (+2),
# and that's it because we've disabled queue notifications
self.assertEqual(email_count + 1 + 2, len(mail.outbox))
# Ensure that <TicketCC> is created even if the Queue notifications are disabled
# so when staff members interact with the <Ticket>, they get notified
for cc_email in cc_list:
# Ensure that contacts on the cc_list are not notified
self.assertNotIn(cc_email, mail.outbox[0].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_followup_from_email_to_a_notification_enabled_queue(self):
"""
Ensure that when an email is sent to a Queue with notifications_enabled turned ON,
and a <FollowUp> is created, all contacts n the TicketCC list are notified.
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# contacts on the cc_list (+2),
# the new and update queues (+2)
expected_email_count = email_count + 1 + 2 + 2
self.assertEqual(expected_email_count, len(mail.outbox))
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[0].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'bravo@example.net'
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', message_id)
reply.__setitem__('Subject', self.ticket_data['title'])
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public.email_address)
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
object_from_message(str(reply), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# As an update was made, we increase the expected_email_count with:
# submitter: +1
# a new email to all TicketCC subscribers : +2
# public_update_queue: +1
expected_email_count += 1 + 2 + 1
self.assertEqual(expected_email_count, len(mail.outbox))
# Ensure that <TicketCC> exist
for cc_email in cc_list:
# Ensure that contacts on cc_list will be notified on the same email (index 0)
#self.assertIn(cc_email, mail.outbox[expected_email_count - 1].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
def test_create_followup_from_email_to_a_notification_disabled_queue(self):
"""
Ensure that when an email is sent to a Queue with notifications_enabled
turned OFF, and a <FollowUp> is created, TicketCC is NOT notified.
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
cc_list = ['bravo@example.net', 'charlie@foobar.com']
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public_with_notifications_disabled.email_address)
msg.__setitem__('Cc', ','.join(cc_list))
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public_with_notifications_disabled, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq2-%s" % ticket.id)
# As we have created an Ticket from an email, we notify:
# the sender (+1),
# the new and update queues (+2)
expected_email_count = email_count + 1 + 2
self.assertEqual(expected_email_count, len(mail.outbox))
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Ensure that contacts on cc_list will not be notified
self.assertNotIn(cc_email, mail.outbox[0].to)
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'bravo@example.net'
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', message_id)
reply.__setitem__('Subject', self.ticket_data['title'])
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public_with_notifications_disabled.email_address)
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
object_from_message(str(reply), self.queue_public_with_notifications_disabled, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq2-%s" % ticket.id)
# As an update was made, we increase the expected_email_count with:
# public_update_queue: +1
expected_email_count += 1
self.assertEqual(expected_email_count, len(mail.outbox))
def test_create_followup_from_email_with_valid_message_id_with_original_cc_list_included(self):
"""
Ensure that if a message is received with an valid In-Reply-To ID,
the expected <TicketCC> instances are created even if the there were
no <TicketCC>s so far.
"""
### Ticket and TicketCCs creation ###
msg = email.message.Message()
message_id = uuid.uuid4().hex
submitter_email = 'foo@bar.py'
msg.__setitem__('Message-ID', message_id)
msg.__setitem__('Subject', self.ticket_data['title'])
msg.__setitem__('From', submitter_email)
msg.__setitem__('To', self.queue_public.email_address)
msg.__setitem__('Content-Type', 'text/plain;')
msg.set_payload(self.ticket_data['description'])
email_count = len(mail.outbox)
object_from_message(str(msg), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
### end of the Ticket and TicketCCs creation ###
# Reply message
reply = email.message.Message()
reply_message_id = uuid.uuid4().hex
submitter_email = 'bravo@example.net'
cc_list = ['foo@bar.py', 'charlie@foobar.com']
reply.__setitem__('Message-ID', reply_message_id)
reply.__setitem__('In-Reply-To', message_id)
reply.__setitem__('Subject', self.ticket_data['title'])
reply.__setitem__('From', submitter_email)
reply.__setitem__('To', self.queue_public.email_address)
reply.__setitem__('Cc', ','.join(cc_list))
reply.__setitem__('Content-Type', 'text/plain;')
reply.set_payload(self.ticket_data['description'])
object_from_message(str(reply), self.queue_public, logger=logger)
followup = FollowUp.objects.get(message_id=message_id)
ticket = Ticket.objects.get(id=followup.ticket.id)
self.assertEqual(ticket.ticket_for_url, "mq1-%s" % ticket.id)
# Ensure that <TicketCC> is created
for cc_email in cc_list:
# Even after 2 messages with the same cc_list, <get> MUST return only
# one object
ticket_cc = TicketCC.objects.get(ticket=ticket, email=cc_email)
self.assertTrue(ticket_cc.ticket, ticket)
self.assertTrue(ticket_cc.email, cc_email)
# As we have created an Ticket from an email, we notify the sender (+1)
# and the new and update queues (+2)
expected_email_count = 1 + 2
# As an update was made, we increase the expected_email_count with:
# submitter: +1
# cc_list: +2
# public_update_queue: +1
expected_email_count += 1 + 2 + 1
self.assertEqual(expected_email_count, len(mail.outbox))

38
helpdesk/views/staff.py
View File

@ -388,17 +388,37 @@ def return_ticketccstring_and_show_subscribe(user, ticket):
return ticketcc_string, show_subscribe
def subscribe_to_ticket_updates(ticket, user=None, email=None, can_view=True, can_update=False):
def subscribe_staff_member_to_ticket(ticket, user):
if ticket is not None:
queryset = TicketCC.objects.filter(ticket=ticket, user=user, email=email)
# Don't create duplicate entries for subscribers
if queryset.count() > 0:
return queryset.first()
if user is None and len(email) < 5:
raise ValidationError(
_('When you add somebody on Cc, you must provide either a User or a valid email. Email: %s' %email)
)
ticketcc = TicketCC(
ticket=ticket,
user=user,
email=email,
can_view=can_view,
can_update=can_update
)
ticketcc.save()
return ticketcc
def subscribe_staff_member_to_ticket(ticket, user, email=''):
"""used in view_ticket() and update_ticket()"""
ticketcc = TicketCC(
ticket=ticket,
user=user,
can_view=True,
can_update=True,
)
ticketcc.save()
return subscribe_to_ticket_updates(ticket=ticket, user=user, email=email, can_view=can_view, can_update=can_update)
def update_ticket(request, ticket_id, public=False):
if not (public or (