2455405819

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/HewlettPackard/squest/commit/<a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/<a class="double-link" href="https://git"><a class=hub.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29">8b89af42a">&lt;a href=&quot;https://github.com/HewlettPackard/squest/commit/</a><a class="double-link" href="https://github.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29">8b89af42a</a><a href="https://github.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29">&lt;a href=&quot;https://github.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;amp;lt;a class=&amp;amp;quot;double-link&amp;amp;quot; href=&amp;amp;quot;https://git&amp;quot;&amp;gt;&amp;amp;lt;a class=&amp;lt;/a&amp;gt;hub.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29&amp;quot;&amp;gt;8b89af42a&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5ff2694f29&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/HewlettPackard/squest/commit/8b89af42a518c59884f036fffbd47e5f&amp;lt;/a&amp;gt;f2694f29&amp;</a>quot;&gt;&amp;quot;&amp;gt;Merge &amp;lt;/a&amp;gt;&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://github.com/HewlettPackard/squest/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/HewlettPackard/squ... (continued)

Pull Request Pull Request #445: [Feature] Approval group

Run Details

645 of 645 new or added lines in 50 files covered. (100.0%)

7015 of 7486 relevant lines covered (93.71%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.56

/service_catalog/mail_utils.py

import base64

from django.conf import settings
from django.contrib.auth.models import User
from django.template.loader import get_template

from service_catalog import tasks
from service_catalog.models.request import RequestState

DEFAULT_FROM_EMAIL = f"squest@{settings.SQUEST_EMAIL_HOST}"
EMAIL_TITLE_PREFIX = "[Squest]"


def _add_user_in_user_list(user, user_list):
    if user.profile.notification_enabled and user.email and user.email not in user_list:
        user_list.append(user.email)
    return user_list


def _remove_user_in_user_list(user, user_list):
    if user.profile.notification_enabled and user.email and user.email in user_list:
        user_list.remove(user.email)
    return user_list


def _get_subject(target_object):
    return f"{EMAIL_TITLE_PREFIX} {target_object.__class__.__name__} #{target_object.id} - {target_object}"


def _get_headers(subject):
    email_id = str(base64.b64encode(subject.encode()))
    headers = dict()
    headers["Message-ID"] = email_id
    headers["In-Reply-To"] = email_id
    headers["References"] = email_id
    return headers


def _get_admin_emails(service):
    """
    Return a list of admin (is_staff) email if notification is enabled and target service subscribed
    :return:
    """
    admins = User.objects.filter(is_staff=True)
    # create a list of email
    email_admins = list()
    for admin in admins:
        if admin.profile.notification_enabled:
            if service in admin.profile.subscribed_services_notification.all():
                email_admins.append(admin.email)
    return email_admins


def _get_receivers_for_request_message(request_message):
    receiver_email_list = _get_admin_emails(service=request_message.request.instance.service)
    receiver_email_list = _add_user_in_user_list(request_message.request.user, receiver_email_list)
    receiver_email_list = _remove_user_in_user_list(request_message.sender, receiver_email_list)
    return receiver_email_list


def _get_receivers_for_support_message(support_message):
    receiver_email_list = _get_admin_emails(service=support_message.support.instance.service)
    receiver_email_list = _add_user_in_user_list(support_message.support.instance.spoc, receiver_email_list)
    receiver_email_list = _remove_user_in_user_list(support_message.sender, receiver_email_list)
    return receiver_email_list


def send_mail_request_update(target_request, user_applied_state=None, message=None, receiver_email_list=None,
                             plain_text=None):
    """
    Notify users that a request has been updated
    :param message: A message to add to the email
    :param user_applied_state: string email of the user who send the notification
    :type user_applied_state: User
    :param target_request:
    :type target_request: service_catalog.models.request.Request
    :param receiver_email_list: email diffusion list
    :type receiver_email_list: List
    :param plain_text: text displayed in the mail
    :type plain_text: String
    :return:
    """
    if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:
        return

    subject = _get_subject(target_request)
    template_name = "service_catalog/mails/request_state_update.html"

    context = {'request': target_request,
               'user_applied_state': user_applied_state,
               'current_site': settings.SQUEST_HOST,
               'message': message}
    if plain_text is None:
        plain_text = f"Request state update: {target_request.state}"
        if target_request.state == RequestState.SUBMITTED:
            plain_text = f"Request update for service: {target_request.instance.name}"
    html_template = get_template(template_name)
    html_content = html_template.render(context)
    if receiver_email_list is None:
        receiver_email_list = _get_admin_emails(service=target_request.instance.service)  # email sent to all admins
    if target_request.user.profile.notification_enabled:
        receiver_email_list.append(target_request.user.email)  # email sent to the requester
    if len(receiver_email_list) > 0:
        tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,
                               bcc=receiver_email_list,
                               headers=_get_headers(subject))


def send_mail_new_support_message(message):
    if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:
        return
    subject = _get_subject(message.support)
    template_name = "service_catalog/mails/support.html"
    plain_text = f"New support message received on Instance #{message.support.instance.id} (#{message.support.id})"
    context = {'message': message, 'current_site': settings.SQUEST_HOST}
    html_template = get_template(template_name)
    html_content = html_template.render(context)
    receiver_email_list = _get_receivers_for_support_message(message)
    if len(receiver_email_list) > 0:
        tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,
                               bcc=receiver_email_list,
                               headers=_get_headers(subject))


def send_mail_new_comment_on_request(message):
    if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:
        return
    subject = _get_subject(message.request)
    template_name = "service_catalog/mails/comment.html"
    plain_text = f"New comment received on Request #{message.request.id}"
    context = {'message': message, 'current_site': settings.SQUEST_HOST}
    html_template = get_template(template_name)
    html_content = html_template.render(context)
    receiver_email_list = _get_receivers_for_request_message(message)
    if len(receiver_email_list) > 0:
        tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,
                               bcc=receiver_email_list,
                               headers=_get_headers(subject))


def send_email_request_canceled(target_request, user_applied_state=None, request_owner_user=None):
    """
    :param target_request: Request model
    :type target_request: service_catalog.models.request.Request
    :param user_applied_state: user who called this method
    :type user_applied_state: User
    :param request_owner_user: user owner of the Request
    :type request_owner_user: User
    :return:
    """
    if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:
        return
    subject = _get_subject(target_request)
    plain_text = f"Request #{target_request.id} - CANCELLED"
    template_name = "service_catalog/mails/request_cancelled.html"
    context = {'request_id': target_request.id,
               'user_applied_state': user_applied_state,
               'current_site': settings.SQUEST_HOST}
    html_template = get_template(template_name)
    html_content = html_template.render(context)
    receiver_email_list = _get_admin_emails(service=target_request.instance.service)  # email sent to all admins
    if request_owner_user.profile.notification_enabled:
        receiver_email_list.append(request_owner_user.email)  # email sent to the requester
    if len(receiver_email_list) > 0:
        tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,
                               bcc=receiver_email_list,
                               headers=_get_headers(subject))

1	import base64	1✔
2
3	from django.conf import settings	1✔
4	from django.contrib.auth.models import User	1✔
5	from django.template.loader import get_template	1✔
6
7	from service_catalog import tasks	1✔
8	from service_catalog.models.request import RequestState	1✔
9
10	DEFAULT_FROM_EMAIL = f"squest@{settings.SQUEST_EMAIL_HOST}"	1✔
11	EMAIL_TITLE_PREFIX = "[Squest]"	1✔
12
13
14	def _add_user_in_user_list(user, user_list):	1✔
15	if user.profile.notification_enabled and user.email and user.email not in user_list:	1✔
16	user_list.append(user.email)	1✔
17	return user_list	1✔
18
19
20	def _remove_user_in_user_list(user, user_list):	1✔
21	if user.profile.notification_enabled and user.email and user.email in user_list:	1✔
22	user_list.remove(user.email)	1✔
23	return user_list	1✔
24
25
26	def _get_subject(target_object):	1✔
27	return f"{EMAIL_TITLE_PREFIX} {target_object.__class__.__name__} #{target_object.id} - {target_object}"	1✔
28
29
30	def _get_headers(subject):	1✔
31	email_id = str(base64.b64encode(subject.encode()))	1✔
32	headers = dict()	1✔
33	headers["Message-ID"] = email_id	1✔
34	headers["In-Reply-To"] = email_id	1✔
35	headers["References"] = email_id	1✔
36	return headers	1✔
37
38
39	def _get_admin_emails(service):	1✔
40	"""
41	Return a list of admin (is_staff) email if notification is enabled and target service subscribed
42	:return:
43	"""
44	admins = User.objects.filter(is_staff=True)	1✔
45	# create a list of email
46	email_admins = list()	1✔
47	for admin in admins:	1✔
48	if admin.profile.notification_enabled:	1✔
49	if service in admin.profile.subscribed_services_notification.all():	1✔
50	email_admins.append(admin.email)	1✔
51	return email_admins	1✔
52
53
54	def _get_receivers_for_request_message(request_message):	1✔
55	receiver_email_list = _get_admin_emails(service=request_message.request.instance.service)	1✔
56	receiver_email_list = _add_user_in_user_list(request_message.request.user, receiver_email_list)	1✔
57	receiver_email_list = _remove_user_in_user_list(request_message.sender, receiver_email_list)	1✔
58	return receiver_email_list	1✔
59
60
61	def _get_receivers_for_support_message(support_message):	1✔
62	receiver_email_list = _get_admin_emails(service=support_message.support.instance.service)	1✔
63	receiver_email_list = _add_user_in_user_list(support_message.support.instance.spoc, receiver_email_list)	1✔
64	receiver_email_list = _remove_user_in_user_list(support_message.sender, receiver_email_list)	1✔
65	return receiver_email_list	1✔
66
67
68	def send_mail_request_update(target_request, user_applied_state=None, message=None, receiver_email_list=None,	1✔
69	plain_text=None):
70	"""
71	Notify users that a request has been updated
72	:param message: A message to add to the email
73	:param user_applied_state: string email of the user who send the notification
74	:type user_applied_state: User
75	:param target_request:
76	:type target_request: service_catalog.models.request.Request
77	:param receiver_email_list: email diffusion list
78	:type receiver_email_list: List
79	:param plain_text: text displayed in the mail
80	:type plain_text: String
81	:return:
82	"""
83	if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:	1✔
84	return	1✔
85
86	subject = _get_subject(target_request)	×
87	template_name = "service_catalog/mails/request_state_update.html"	×
88
89	context = {'request': target_request,	×
90	'user_applied_state': user_applied_state,
91	'current_site': settings.SQUEST_HOST,
92	'message': message}
93	if plain_text is None:	×
94	plain_text = f"Request state update: {target_request.state}"	×
95	if target_request.state == RequestState.SUBMITTED:	×
96	plain_text = f"Request update for service: {target_request.instance.name}"	×
97	html_template = get_template(template_name)	×
98	html_content = html_template.render(context)	×
99	if receiver_email_list is None:	×
100	receiver_email_list = _get_admin_emails(service=target_request.instance.service) # email sent to all admins	×
101	if target_request.user.profile.notification_enabled:	×
102	receiver_email_list.append(target_request.user.email) # email sent to the requester	×
103	if len(receiver_email_list) > 0:	×
104	tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,	×
105	bcc=receiver_email_list,
106	headers=_get_headers(subject))
107
108
109	def send_mail_new_support_message(message):	1✔
110	if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:	1✔
111	return	1✔
112	subject = _get_subject(message.support)	×
113	template_name = "service_catalog/mails/support.html"	×
114	plain_text = f"New support message received on Instance #{message.support.instance.id} (#{message.support.id})"	×
115	context = {'message': message, 'current_site': settings.SQUEST_HOST}	×
116	html_template = get_template(template_name)	×
117	html_content = html_template.render(context)	×
118	receiver_email_list = _get_receivers_for_support_message(message)	×
119	if len(receiver_email_list) > 0:	×
120	tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,	×
121	bcc=receiver_email_list,
122	headers=_get_headers(subject))
123
124
125	def send_mail_new_comment_on_request(message):	1✔
126	if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:	1✔
127	return	1✔
128	subject = _get_subject(message.request)	×
129	template_name = "service_catalog/mails/comment.html"	×
130	plain_text = f"New comment received on Request #{message.request.id}"	×
131	context = {'message': message, 'current_site': settings.SQUEST_HOST}	×
132	html_template = get_template(template_name)	×
133	html_content = html_template.render(context)	×
134	receiver_email_list = _get_receivers_for_request_message(message)	×
135	if len(receiver_email_list) > 0:	×
136	tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,	×
137	bcc=receiver_email_list,
138	headers=_get_headers(subject))
139
140
141	def send_email_request_canceled(target_request, user_applied_state=None, request_owner_user=None):	1✔
142	"""
143	:param target_request: Request model
144	:type target_request: service_catalog.models.request.Request
145	:param user_applied_state: user who called this method
146	:type user_applied_state: User
147	:param request_owner_user: user owner of the Request
148	:type request_owner_user: User
149	:return:
150	"""
151	if not settings.SQUEST_EMAIL_NOTIFICATION_ENABLED:	1✔
152	return	1✔
153	subject = _get_subject(target_request)	×
154	plain_text = f"Request #{target_request.id} - CANCELLED"	×
155	template_name = "service_catalog/mails/request_cancelled.html"	×
156	context = {'request_id': target_request.id,	×
157	'user_applied_state': user_applied_state,
158	'current_site': settings.SQUEST_HOST}
159	html_template = get_template(template_name)	×
160	html_content = html_template.render(context)	×
161	receiver_email_list = _get_admin_emails(service=target_request.instance.service) # email sent to all admins	×
162	if request_owner_user.profile.notification_enabled:	×
163	receiver_email_list.append(request_owner_user.email) # email sent to the requester	×
164	if len(receiver_email_list) > 0:	×
165	tasks.send_email.delay(subject, plain_text, html_content, DEFAULT_FROM_EMAIL,	×
166	bcc=receiver_email_list,
167	headers=_get_headers(subject))

HewlettPackard / squest / 2455405819

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous