6700616467

Committed 31 Oct 2023 01:00AM UTC coverage: 92.452% (-1.2%) from 93.66%

Build # 6700616467

Build Type

Pull #383

github

Committed by

web-flow

Commit Message

Merge 300d02535 into fb838f71e

Pull Request Pull Request #383: Add offboarding sequences

Run Details

6161 of 6664 relevant lines covered (92.45%)

0.92 hits per line

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

97.33

back/admin/admin_tasks/models.py

from django.conf import settings
from django.db import models
from django.template.loader import render_to_string
from django.utils.translation import gettext as _

from slack_bot.utils import Slack, actions, button, paragraph

from .emails import (
    send_email_new_assigned_admin,
    send_email_new_comment,
    send_email_notification_to_external_person,
)


class AdminTask(models.Model):
    class Priority(models.IntegerChoices):
        EMAIL = 1, _("Low")
        MEDIUM = 2, _("Medium")
        HIGH = 3, _("High")

    class Notification(models.IntegerChoices):
        NO = 0, _("No")
        EMAIL = 1, _("Email")
        SLACK = 2, _("Slack")

    new_hire = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name=_("New hire"),
        on_delete=models.CASCADE,
        related_name="new_hire_tasks",
    )
    assigned_to = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name=_("Assigned to"),
        on_delete=models.CASCADE,
        related_name="owner",
        blank=True,
        null=True,
    )
    name = models.CharField(verbose_name=_("Name"), max_length=500)
    option = models.IntegerField(
        verbose_name=_("Send email or text to extra user?"),
        choices=Notification.choices,
    )
    slack_user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name=_("Slack user"),
        on_delete=models.SET_NULL,
        related_name="slack_user",
        blank=True,
        null=True,
    )
    email = models.EmailField(
        verbose_name=_("Email"), max_length=12500, default="", blank=True
    )
    completed = models.BooleanField(verbose_name=_("Completed"), default=False)
    date = models.DateField(verbose_name=_("Date"), blank=True, null=True)
    priority = models.IntegerField(
        verbose_name=_("Priority"), choices=Priority.choices, default=Priority.MEDIUM
    )
    based_on = models.ForeignKey(
        "sequences.PendingAdminTask",
        null=True,
        on_delete=models.SET_NULL,
        help_text=_("If generated through a sequence, then this will be filled"),
    )
    integration = models.ForeignKey(
        "integrations.Integration",
        null=True,
        on_delete=models.SET_NULL,
        help_text=_("Only set if generated based on a manual integration."),
    )
    create_integration = models.BooleanField(
        default=False,
        help_text=_(
            "Specifies if integration has been created or removed by completing the "
            "admin task."
        ),
    )

    @property
    def get_icon_template(self):
        return render_to_string("_admin_task_icon.html")

    def send_notification_third_party(self):
        if self.assigned_to is None:
            # Only happens when a sequence adds this with "manager" or "buddy" is
            # choosen option
            return
        if self.option == AdminTask.Notification.EMAIL:
            send_email_notification_to_external_person(self)
        elif self.option == AdminTask.Notification.SLACK:
            blocks = [
                paragraph(
                    _(
                        "%(name_assigned_to)s needs your help with this task:\n*"
                        "%(task_name)s*\n_%(comment)s_"
                    )
                    % {
                        "name_assigned_to": self.assigned_to.full_name,
                        "task_name": self.name,
                        "comment": self.comment.last().content,
                    }
                ),
                actions(
                    [
                        button(
                            text=_("I have completed this"),
                            style="primary",
                            value=str(self.pk),
                            action_id="admin_task:complete",
                        )
                    ]
                ),
            ]
            Slack().send_message(blocks, self.slack_user.slack_user_id)

    def send_notification_new_assigned(self):
        if self.assigned_to is None:
            # Only happens when a sequence adds this with "manager" or "buddy" is
            # choosen option
            return
        if self.assigned_to.has_slack_account:
            comment = ""
            if self.comment.all().exists():
                comment = _("_%(comment)s_\n by _%(name)s_") % {
                    "comment": self.comment.last().content,
                    "name": self.comment.last().comment_by.full_name,
                }

            text = _("You have just been assigned to *%(title)s* for *%(name)s*\n") % {
                "title": self.name,
                "name": self.new_hire.full_name,
            }
            text += comment
            blocks = [
                paragraph(text),
                actions(
                    [
                        button(
                            text=_("I have completed this"),
                            style="primary",
                            value=str(self.pk),
                            action_id="admin_task:complete",
                        )
                    ]
                ),
            ]
            Slack().send_message(
                blocks=blocks,
                channel=self.assigned_to.slack_user_id,
                text=_("New assigned task!"),
            )
        else:
            send_email_new_assigned_admin(self)

    def mark_completed(self):
        from admin.sequences.tasks import process_condition

        self.completed = True
        self.save()

        # Get conditions with this to do item as (part of the) condition
        conditions = self.new_hire.conditions.filter(
            condition_admin_tasks=self.based_on
        )

        for condition in conditions:
            condition_admin_tasks_id = condition.condition_admin_tasks.values_list(
                "id", flat=True
            )

            # Check if all admin to do items have been added to new hire and are
            # completed. If not, then we know it should not be triggered yet
            completed_tasks = AdminTask.objects.filter(
                based_on_id__in=condition_admin_tasks_id,
                new_hire=self.new_hire,
                completed=True,
            )

            # If the amount matches, then we should process it
            if completed_tasks.count() == len(condition_admin_tasks_id):
                # Send notification only if user has a slack account
                process_condition(
                    condition.id, self.new_hire.id, self.new_hire.has_slack_account
                )

    class Meta:
        ordering = ["completed", "date"]


class AdminTaskComment(models.Model):
    admin_task = models.ForeignKey(
        "AdminTask", on_delete=models.CASCADE, related_name="comment"
    )
    content = models.CharField(max_length=12500)
    date = models.DateTimeField(auto_now_add=True)
    comment_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)

    class Meta:
        ordering = ["-date"]

    def send_notification_new_message(self):
        if self.admin_task.assigned_to != self.comment_by:
            if self.admin_task.assigned_to.has_slack_account:
                blocks = [
                    paragraph(
                        _(
                            "%(name)s added a message to your task:\n*"
                            "%(task_title)s*\n_%(comment)s_"
                        )
                        % {
                            "name": self.comment_by.full_name,
                            "task_title": self.admin_task.name,
                            "comment": self.content,
                        }
                    ),
                    actions(
                        [
                            button(
                                text=_("I have completed this"),
                                style="primary",
                                value=str(self.pk),
                                action_id="admin_task:complete",
                            )
                        ]
                    ),
                ]
                Slack().send_message(blocks, self.admin_task.assigned_to.slack_user_id)
            else:
                send_email_new_comment(self)

1	from django.conf import settings	1✔
2	from django.db import models	1✔
3	from django.template.loader import render_to_string	1✔
4	from django.utils.translation import gettext as _	1✔
5
6	from slack_bot.utils import Slack, actions, button, paragraph	1✔
7
8	from .emails import (	1✔
9	send_email_new_assigned_admin,
10	send_email_new_comment,
11	send_email_notification_to_external_person,
12	)
13
14
15	class AdminTask(models.Model):	1✔
16	class Priority(models.IntegerChoices):	1✔
17	EMAIL = 1, _("Low")	1✔
18	MEDIUM = 2, _("Medium")	1✔
19	HIGH = 3, _("High")	1✔
20
21	class Notification(models.IntegerChoices):	1✔
22	NO = 0, _("No")	1✔
23	EMAIL = 1, _("Email")	1✔
24	SLACK = 2, _("Slack")	1✔
25
26	new_hire = models.ForeignKey(	1✔
27	settings.AUTH_USER_MODEL,
28	verbose_name=_("New hire"),
29	on_delete=models.CASCADE,
30	related_name="new_hire_tasks",
31	)
32	assigned_to = models.ForeignKey(	1✔
33	settings.AUTH_USER_MODEL,
34	verbose_name=_("Assigned to"),
35	on_delete=models.CASCADE,
36	related_name="owner",
37	blank=True,
38	null=True,
39	)
40	name = models.CharField(verbose_name=_("Name"), max_length=500)	1✔
41	option = models.IntegerField(	1✔
42	verbose_name=_("Send email or text to extra user?"),
43	choices=Notification.choices,
44	)
45	slack_user = models.ForeignKey(	1✔
46	settings.AUTH_USER_MODEL,
47	verbose_name=_("Slack user"),
48	on_delete=models.SET_NULL,
49	related_name="slack_user",
50	blank=True,
51	null=True,
52	)
53	email = models.EmailField(	1✔
54	verbose_name=_("Email"), max_length=12500, default="", blank=True
55	)
56	completed = models.BooleanField(verbose_name=_("Completed"), default=False)	1✔
57	date = models.DateField(verbose_name=_("Date"), blank=True, null=True)	1✔
58	priority = models.IntegerField(	1✔
59	verbose_name=_("Priority"), choices=Priority.choices, default=Priority.MEDIUM
60	)
61	based_on = models.ForeignKey(	1✔
62	"sequences.PendingAdminTask",
63	null=True,
64	on_delete=models.SET_NULL,
65	help_text=_("If generated through a sequence, then this will be filled"),
66	)
67	integration = models.ForeignKey(	1✔
68	"integrations.Integration",
69	null=True,
70	on_delete=models.SET_NULL,
71	help_text=_("Only set if generated based on a manual integration."),
72	)
73	create_integration = models.BooleanField(	1✔
74	default=False,
75	help_text=_(
76	"Specifies if integration has been created or removed by completing the "
77	"admin task."
78	),
79	)
80
81	@property	1✔
82	def get_icon_template(self):	1✔
83	return render_to_string("_admin_task_icon.html")	1✔
84
85	def send_notification_third_party(self):	1✔
86	if self.assigned_to is None:	1✔
87	# Only happens when a sequence adds this with "manager" or "buddy" is
88	# choosen option
89	return	×
90	if self.option == AdminTask.Notification.EMAIL:	1✔
91	send_email_notification_to_external_person(self)	1✔
92	elif self.option == AdminTask.Notification.SLACK:	1✔
93	blocks = [	1✔
94	paragraph(
95	_(
96	"%(name_assigned_to)s needs your help with this task:\n*"
97	"%(task_name)s*\n_%(comment)s_"
98	)
99	% {
100	"name_assigned_to": self.assigned_to.full_name,
101	"task_name": self.name,
102	"comment": self.comment.last().content,
103	}
104	),
105	actions(
106	[
107	button(
108	text=_("I have completed this"),
109	style="primary",
110	value=str(self.pk),
111	action_id="admin_task:complete",
112	)
113	]
114	),
115	]
116	Slack().send_message(blocks, self.slack_user.slack_user_id)	1✔
117
118	def send_notification_new_assigned(self):	1✔
119	if self.assigned_to is None:	1✔
120	# Only happens when a sequence adds this with "manager" or "buddy" is
121	# choosen option
122	return	×
123	if self.assigned_to.has_slack_account:	1✔
124	comment = ""	1✔
125	if self.comment.all().exists():	1✔
126	comment = _("_%(comment)s_\n by _%(name)s_") % {	1✔
127	"comment": self.comment.last().content,
128	"name": self.comment.last().comment_by.full_name,
129	}
130
131	text = _("You have just been assigned to %(title)s for %(name)s\n") % {	1✔
132	"title": self.name,
133	"name": self.new_hire.full_name,
134	}
135	text += comment	1✔
136	blocks = [	1✔
137	paragraph(text),
138	actions(
139	[
140	button(
141	text=_("I have completed this"),
142	style="primary",
143	value=str(self.pk),
144	action_id="admin_task:complete",
145	)
146	]
147	),
148	]
149	Slack().send_message(	1✔
150	blocks=blocks,
151	channel=self.assigned_to.slack_user_id,
152	text=_("New assigned task!"),
153	)
154	else:
155	send_email_new_assigned_admin(self)	1✔
156
157	def mark_completed(self):	1✔
158	from admin.sequences.tasks import process_condition	1✔
159
160	self.completed = True	1✔
161	self.save()	1✔
162
163	# Get conditions with this to do item as (part of the) condition
164	conditions = self.new_hire.conditions.filter(	1✔
165	condition_admin_tasks=self.based_on
166	)
167
168	for condition in conditions:	1✔
169	condition_admin_tasks_id = condition.condition_admin_tasks.values_list(	1✔
170	"id", flat=True
171	)
172
173	# Check if all admin to do items have been added to new hire and are
174	# completed. If not, then we know it should not be triggered yet
175	completed_tasks = AdminTask.objects.filter(	1✔
176	based_on_id__in=condition_admin_tasks_id,
177	new_hire=self.new_hire,
178	completed=True,
179	)
180
181	# If the amount matches, then we should process it
182	if completed_tasks.count() == len(condition_admin_tasks_id):	1✔
183	# Send notification only if user has a slack account
184	process_condition(	1✔
185	condition.id, self.new_hire.id, self.new_hire.has_slack_account
186	)
187
188	class Meta:	1✔
189	ordering = ["completed", "date"]	1✔
190
191
192	class AdminTaskComment(models.Model):	1✔
193	admin_task = models.ForeignKey(	1✔
194	"AdminTask", on_delete=models.CASCADE, related_name="comment"
195	)
196	content = models.CharField(max_length=12500)	1✔
197	date = models.DateTimeField(auto_now_add=True)	1✔
198	comment_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)	1✔
199
200	class Meta:	1✔
201	ordering = ["-date"]	1✔
202
203	def send_notification_new_message(self):	1✔
204	if self.admin_task.assigned_to != self.comment_by:	1✔
205	if self.admin_task.assigned_to.has_slack_account:	1✔
206	blocks = [	1✔
207	paragraph(
208	_(
209	"%(name)s added a message to your task:\n*"
210	"%(task_title)s*\n_%(comment)s_"
211	)
212	% {
213	"name": self.comment_by.full_name,
214	"task_title": self.admin_task.name,
215	"comment": self.content,
216	}
217	),
218	actions(
219	[
220	button(
221	text=_("I have completed this"),
222	style="primary",
223	value=str(self.pk),
224	action_id="admin_task:complete",
225	)
226	]
227	),
228	]
229	Slack().send_message(blocks, self.admin_task.assigned_to.slack_user_id)	1✔
230	else:
231	send_email_new_comment(self)	1✔

chiefonboarding / ChiefOnboarding / 6700616467

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