Source code for tcms.signals

# pylint: disable=unused-argument, import-outside-toplevel
"""
Defines custom signals sent throughout Kiwi TCMS. You can connect your own
handlers if you'd like to augment some of the default behavior!

If you simply want to connect a signal handler add the following code to your
``local_settings.py``::

    from tcms.signals import *

    USER_REGISTERED_SIGNAL.connect(notify_admins)

In case you want to perform more complex signal handling we advise you to create
a new Django app and connect your handler function(s) to the desired signals
inside the
`AppConfig.ready
<https://docs.djangoproject.com/en/2.2/ref/applications/#django.apps.AppConfig.ready>`_
method. When you are done connect your Django app to the rest of Kiwi TCMS by
altering the following setting::

    INSTALLED_APPS += ['my_custom_app']
"""
from django.db.models import ObjectDoesNotExist
from django.dispatch import Signal
from django.utils.translation import gettext_lazy as _

__all__ = [
    "USER_REGISTERED_SIGNAL",
    "notify_admins",
    "pre_save_clean",
    "handle_attachments_pre_delete",
    "handle_attachments_post_save",
    "handle_comments_pre_delete",
    "handle_emails_post_case_save",
    "handle_emails_pre_case_delete",
    "handle_emails_post_plan_save",
    "handle_emails_post_run_save",
    "handle_emails_post_bug_save",
]


#: Sent when a new user is registered into Kiwi TCMS. This signal receives two
#: keyword parameters: ``request`` and ``user`` respectively!
USER_REGISTERED_SIGNAL = Signal()


[docs]def notify_admins(sender, **kwargs): """ Very simple signal handler which sends emails to site admins when a new user has been registered! .. warning:: This handler isn't connected to the ``USER_REGISTERED_SIGNAL`` by default! """ from django.conf import settings from django.contrib.auth import get_user_model from django.urls import reverse from tcms.core.utils import request_host_link from tcms.core.utils.mailto import mailto if kwargs.get("raw", False): return admin_emails = set() # super-users can approve others for super_user in get_user_model().objects.filter(is_superuser=True): admin_emails.add(super_user.email) # site admins should be able to do so as well for _name, email in settings.ADMINS: admin_emails.add(email) request = kwargs.get("request") user = kwargs.get("user") user_url = request_host_link(request) + reverse( "admin:auth_user_change", args=[user.pk] ) mailto( template_name="email/user_registered/notify_admins.txt", recipients=list(admin_emails), subject=str(_("New user awaiting approval")), context={ "username": user.username, "user_url": user_url, }, )
[docs]def handle_emails_post_case_save(sender, instance, created=False, **kwargs): """ Send email updates after a TestCase has been updated! """ if kwargs.get("raw", False): return if not created and instance.emailing.notify_on_case_update: from tcms.testcases.helpers import email email.email_case_update(instance)
[docs]def handle_emails_pre_case_delete(sender, **kwargs): """ Send email updates before a TestCase will be deleted! """ if kwargs.get("raw", False): return instance = kwargs["instance"] try: # note: using the `email_settings` related object instead of the # `emailing` property b/c it breaks with cascading deletes via admin. # if there are not settings created before hand they default to False # so email will not going to be sent and the exception is safe to ignore if instance.email_settings.notify_on_case_delete: from tcms.testcases.helpers import email email.email_case_deletion(instance) except ObjectDoesNotExist: pass
[docs]def pre_save_clean(sender, **kwargs): if kwargs.get("raw", False): return instance = kwargs["instance"] instance.clean()
[docs]def handle_emails_post_plan_save(sender, instance, created=False, **kwargs): """ Send email updates after a TestPlan has been updated! """ if kwargs.get("raw", False): return if not created and instance.emailing.notify_on_plan_update: from tcms.testplans.helpers import email email.email_plan_update(instance)
[docs]def handle_emails_post_run_save(sender, *_args, **kwargs): """ Send email updates after a TestRus has been created or updated! """ from tcms.core.history import history_email_for from tcms.core.utils.mailto import mailto if kwargs.get("raw", False): return instance = kwargs["instance"] if kwargs.get("created"): template_name = "email/post_run_save/email.txt" subject = _("NEW: TestRun #%(pk)d - %(summary)s") % { "pk": instance.pk, "summary": instance.summary, } context = {"test_run": instance} else: template_name = None subject, context = history_email_for(instance, instance.summary) mailto(template_name, subject, instance.get_notify_addrs(), context)
[docs]def handle_attachments_pre_delete(sender, **kwargs): """ Delete files attached to object which is about to be deleted b/c django-attachments' object_id is not a FK relationship and we can't rely on cascading delete! """ from attachments.models import Attachment from attachments.views import remove_file_from_disk if kwargs.get("raw", False): return instance = kwargs["instance"] attached_files = Attachment.objects.attachments_for_object(instance) for attachment in attached_files: remove_file_from_disk(attachment.attachment_file) attached_files.delete()
[docs]def handle_comments_pre_delete(sender, **kwargs): """ Delete comments attached to object which is about to be deleted b/c django-comments' object_pk is not a FK relationship and we can't rely on cascading delete! """ from tcms.core.helpers.comments import get_comments if kwargs.get("raw", False): return instance = kwargs["instance"] get_comments(instance).delete()
[docs]def handle_emails_post_bug_save(sender, instance, created=False, **kwargs): if not kwargs.get("called_from_add_comment"): return from tcms.core.helpers.comments import get_comments from tcms.core.utils.mailto import mailto comments = get_comments(instance) recipients = set(comments.values_list("user_email", flat=True)) recipients.add(instance.reporter.email) if instance.assignee: recipients.add(instance.assignee.email) if not recipients: return mailto( template_name="email/post_bug_save/email.txt", recipients=list(recipients), subject=_("Bug #%(pk)d - %(summary)s") % {"pk": instance.pk, "summary": instance.summary}, context={ "bug": instance, "comment": comments.last(), }, )
def _introspect_request(): """ Introspect the current thread b/c signals are executed synchronously after .save() and find out the `request` variable. """ import inspect for frame_record in inspect.stack(): if frame_record[3] == "get_response": return frame_record[0].f_locals["request"] return None
[docs]def handle_attachments_post_save(sender, instance, created=False, **kwargs): """ SimpleMDE image/file buttons will upload attachments under the currently logged-in user. This signal handler will re-attach these files under the document which is being saved! """ from attachments.models import Attachment if kwargs.get("raw", False): return request = _introspect_request() if not request: return for attachment in Attachment.objects.attachments_for_object(request.user): attachment.attach_to(instance)