6778659173

Committed 07 Nov 2023 01:14AM UTC coverage: 93.709% (+0.05%) from 93.66%

Build # 6778659173

Build Type

Pull #383

github

Committed by

web-flow

Commit Message

Merge b5553c900 into 5b93b8619

Pull Request Pull Request #383: Add offboarding sequences

Run Details

6271 of 6692 relevant lines covered (93.71%)

0.94 hits per line

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

97.67

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 organization.models import Notification

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


class AminTaskManager(models.Manager):
    def create_admin_task(
        self,
        new_hire,
        assigned_to,
        name,
        option,
        slack_user,
        email,
        date,
        priority,
        pending_admin_task,
        manual_integration,
        comment,
        send_notification,
    ):
        admin_task = AdminTask.objects.create(
            new_hire=new_hire,
            assigned_to=assigned_to,
            name=name,
            option=option,
            slack_user=slack_user,
            email=email,
            date=date,
            priority=priority,
            based_on=pending_admin_task,
            manual_integration=manual_integration,
        )
        AdminTaskComment.objects.create(
            content=comment,
            comment_by=admin_task.assigned_to,
            admin_task=admin_task,
        )
        if send_notification:
            admin_task.send_notification_new_assigned()

        admin_task.send_notification_third_party()

        Notification.objects.create(
            notification_type=Notification.Type.ADDED_ADMIN_TASK,
            extra_text=name,
            created_for=admin_task.assigned_to,
        )


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"),
    )
    manual_integration = models.ForeignKey(
        "integrations.Integration",
        null=True,
        on_delete=models.SET_NULL,
        help_text=_("Only set if generated based on a manual integration."),
    )

    objects = AminTaskManager()

    @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()

        # Check if we need to register the manual integration
        if self.manual_integration is not None:
            self.manual_integration.register_manual_integration_run(
                self.new_hire
            )

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

chiefonboarding / ChiefOnboarding / 6778659173

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