18481185920

Committed 13 Oct 2025 11:53PM UTC coverage: 89.613% (-0.006%) from 89.619%

Build # 18481185920

Build Type

Pull #572

github

Committed by

web-flow

Commit Message

Merge f5b8970dc into f01e5ecbf

Pull Request Pull Request #572: Add permissions based on department

Run Details

7057 of 7875 relevant lines covered (89.61%)

0.9 hits per line

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

96.94

back/admin/admin_tasks/models.py

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

from organization.models import Notification
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 FilteredForManagerQuerySet(models.QuerySet):
    def for_user(self, user):
        if user.is_manager:
            return self.filter(
                Q(new_hire__departments__isnull=True)
                | Q(new_hire__departments__in=user.departments.all())
            ).discint()
        return self


class AdminTaskManager(models.Manager):
    def for_user(self, user):
        return self.get_queryset().for_user(user)

    def get_queryset(self):
        return FilteredForManagerQuerySet(self.model, using=self._db)

    def create_admin_task(
        self,
        new_hire,
        assigned_to,
        name,
        option=0,
        slack_user=None,
        email="",
        date=None,
        priority=2,
        pending_admin_task=None,
        hardware=None,
        manual_integration=None,
        comment="-",
        send_notification=True,
    ):
        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,
            hardware=hardware,
            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."),
    )
    hardware = models.ForeignKey(
        "hardware.Hardware",
        null=True,
        on_delete=models.SET_NULL,
        help_text=_("Only set if generated based on hardware."),
    )

    objects = AdminTaskManager()

    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)

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

chiefonboarding / ChiefOnboarding / 18481185920

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