Coverage for sm/notifications.py: 89%

1from django.conf import settings 

2from django.contrib.auth import get_user_model 

3from django.core.mail import send_mail 

4from django.db.models.signals import post_save, pre_save 

5from django.dispatch import receiver 

6from django.template.loader import render_to_string 

7from django.utils.translation import gettext as _ 

8 

9from server.models import Model as Server 

10 

11 

12def _notification_enabled() -> bool: 

13 return getattr(settings, "SERVER_STATUS_NOTIFY", False) 

14 

15 

16def _recipients(server: Server) -> list: 

17 """Group members with an email address, plus superusers.""" 

18 users = set() 

19 if server.group_id: 

20 for user in server.group.user_set.all(): 

21 if user.email: 

22 users.add(user.email) 

23 for user in get_user_model().objects.filter(is_superuser=True): 

24 if user.email: 

25 users.add(user.email) 

26 return sorted(users) 

27 

28 

29def _send_status_change_email(instance: Server) -> None: 

30 recipients = _recipients(instance) 

31 if not recipients: 

32 return 

33 

34 subject = _("Server %(hostname)s status changed to %(status)s") % { 

35 "hostname": instance.hostname, 

36 "status": instance.status.name if instance.status_id else _("unknown"), 

37 } 

38 message = render_to_string( 

39 "notifications/server_status_change.txt", 

40 { 

41 "server": instance, 

42 "status": instance.status, 

43 "group": instance.group, 

44 }, 

45 ) 

46 send_mail( 

47 subject, 

48 message, 

49 getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@servermanager"), 

50 recipients, 

51 fail_silently=True, 

52 ) 

53 

54 

55@receiver(pre_save, sender=Server) 

56def remember_status_change(sender, instance, **kwargs): 

57 """Capture the previous status so we can detect changes in post_save.""" 

58 if instance.pk is None: 

59 instance._old_status_id = None 

60 return 

61 try: 

62 old = Server.objects.filter(pk=instance.pk).values("status_id").first() 

63 instance._old_status_id = old["status_id"] if old else None 

64 except Server.DoesNotExist: 

65 instance._old_status_id = None 

66 

67 

68@receiver(post_save, sender=Server) 

69def notify_on_status_change(sender, instance, created, **kwargs): 

70 """Send an email to the server's group members when its status changes.""" 

71 if created or not _notification_enabled(): 

72 return 

73 old_status_id = getattr(instance, "_old_status_id", None) 

74 if old_status_id == instance.status_id: 

75 return 

76 _send_status_change_email(instance)