agendas: add email notifications for events (#44158)

This commit is contained in:
Valentin Deniaud 2020-07-16 15:12:47 +02:00
parent f8f324a945
commit e2fdbb8f58
15 changed files with 533 additions and 7 deletions

View File

@ -0,0 +1,65 @@
# chrono - agendas system
# Copyright (C) 2020 Entr'ouvert
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU Affero General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from urllib.parse import urljoin
from django.conf import settings
from django.core.mail import send_mail
from django.core.management.base import BaseCommand
from django.db.transaction import atomic
from django.template.loader import render_to_string
from django.utils import timezone
from django.utils.translation import ugettext_lazy as _
from chrono.agendas.models import Agenda
class Command(BaseCommand):
EMAIL_SUBJECTS = {
'almost_full': _('Alert: event "%s" is almost full (90%%)'),
'full': _('Alert: event "%s" is full'),
'cancelled': _('Alert: event "%s" is cancelled'),
}
help = 'Send email notifications'
def handle(self, **options):
agendas = Agenda.objects.filter(notifications_settings__isnull=False).select_related(
'notifications_settings'
)
for agenda in agendas:
for notification_type in agenda.notifications_settings.get_notification_types():
recipients = notification_type.get_recipients()
if not recipients:
continue
status = notification_type.related_field
filter_kwargs = {status: True, status + '_notification_timestamp__isnull': True}
events = agenda.event_set.filter(**filter_kwargs)
for event in events:
self.send_notification(event, status, recipients)
def send_notification(self, event, status, recipients):
subject = self.EMAIL_SUBJECTS[status] % event
ctx = {'event': event, 'event_url': urljoin(settings.SITE_BASE_URL, event.get_absolute_view_url())}
ctx.update(settings.TEMPLATE_VARS)
body = render_to_string('agendas/event_notification_body.txt', ctx)
html_body = render_to_string('agendas/event_notification_body.html', ctx)
timestamp = timezone.now()
with atomic():
setattr(event, status + '_notification_timestamp', timestamp)
event.save()
send_mail(subject, body, settings.DEFAULT_FROM_EMAIL, recipients, html_message=html_body)

View File

@ -0,0 +1,45 @@
# -*- coding: utf-8 -*-
# Generated by Django 1.11.18 on 2020-09-03 08:41
from __future__ import unicode_literals
import django.contrib.postgres.fields
from django.db import migrations, models
import django.db.models.deletion
class Migration(migrations.Migration):
dependencies = [
('agendas', '0059_event_almost_full'),
]
operations = [
migrations.CreateModel(
name='AgendaNotificationsSettings',
fields=[
('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('almost_full_event', models.CharField(blank=True, choices=[('edit-role', 'Edit Role'), ('view-role', 'View Role'), ('use-email-field', 'Specify email addresses manually')], max_length=16, verbose_name='Almost full event (90%)')),
('almost_full_event_emails', django.contrib.postgres.fields.ArrayField(base_field=models.EmailField(max_length=254), blank=True, null=True, size=None)),
('full_event', models.CharField(blank=True, choices=[('edit-role', 'Edit Role'), ('view-role', 'View Role'), ('use-email-field', 'Specify email addresses manually')], max_length=16, verbose_name='Full event')),
('full_event_emails', django.contrib.postgres.fields.ArrayField(base_field=models.EmailField(max_length=254), blank=True, null=True, size=None)),
('cancelled_event', models.CharField(blank=True, choices=[('edit-role', 'Edit Role'), ('view-role', 'View Role'), ('use-email-field', 'Specify email addresses manually')], max_length=16, verbose_name='Cancelled event')),
('cancelled_event_emails', django.contrib.postgres.fields.ArrayField(base_field=models.EmailField(max_length=254), blank=True, null=True, size=None)),
('agenda', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, related_name='notifications_settings', to='agendas.Agenda')),
],
),
migrations.AddField(
model_name='event',
name='almost_full_notification_timestamp',
field=models.DateTimeField(blank=True, null=True),
),
migrations.AddField(
model_name='event',
name='cancelled_notification_timestamp',
field=models.DateTimeField(blank=True, null=True),
),
migrations.AddField(
model_name='event',
name='full_notification_timestamp',
field=models.DateTimeField(blank=True, null=True),
),
]

View File

@ -29,6 +29,7 @@ import vobject
import django
from django.conf import settings
from django.contrib.auth.models import Group
from django.contrib.postgres.fields import ArrayField
from django.core.exceptions import FieldDoesNotExist
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator
@ -272,6 +273,8 @@ class Agenda(models.Model):
}
if self.kind == 'events':
agenda['events'] = [x.export_json() for x in self.event_set.all()]
if hasattr(self, 'notifications_settings'):
agenda['notifications_settings'] = self.notifications_settings.export_json()
elif self.kind == 'meetings':
agenda['meetingtypes'] = [x.export_json() for x in self.meetingtype_set.all()]
agenda['desks'] = [desk.export_json() for desk in self.desk_set.all()]
@ -286,6 +289,7 @@ class Agenda(models.Model):
permissions = data.pop('permissions') or {}
if data['kind'] == 'events':
events = data.pop('events')
notifications_settings = data.pop('notifications_settings', None)
elif data['kind'] == 'meetings':
meetingtypes = data.pop('meetingtypes')
desks = data.pop('desks')
@ -313,9 +317,13 @@ class Agenda(models.Model):
if data['kind'] == 'events':
if overwrite:
Event.objects.filter(agenda=agenda).delete()
AgendaNotificationsSettings.objects.filter(agenda=agenda).delete()
for event_data in events:
event_data['agenda'] = agenda
Event.import_json(event_data).save()
if notifications_settings:
notifications_settings['agenda'] = agenda
AgendaNotificationsSettings.import_json(notifications_settings).save()
elif data['kind'] == 'meetings':
if overwrite:
MeetingType.objects.filter(agenda=agenda).delete()
@ -784,6 +792,10 @@ class Event(models.Model):
desk = models.ForeignKey('Desk', null=True, on_delete=models.CASCADE)
resources = models.ManyToManyField('Resource')
almost_full_notification_timestamp = models.DateTimeField(null=True, blank=True)
full_notification_timestamp = models.DateTimeField(null=True, blank=True)
cancelled_notification_timestamp = models.DateTimeField(null=True, blank=True)
class Meta:
ordering = ['agenda', 'start_datetime', 'duration', 'label']
unique_together = ('agenda', 'slug')
@ -909,6 +921,9 @@ class Event(models.Model):
def get_absolute_url(self):
return reverse('chrono-manager-event-edit', kwargs={'pk': self.agenda.id, 'event_pk': self.id})
def get_absolute_view_url(self):
return reverse('chrono-manager-event-view', kwargs={'pk': self.agenda.id, 'event_pk': self.id})
@classmethod
def import_json(cls, data):
data['start_datetime'] = make_aware(
@ -1533,3 +1548,101 @@ class EventCancellationReport(models.Model):
class Meta:
ordering = ['-timestamp']
class NotificationType:
def __init__(self, name, related_field, settings):
self.name = name
self.related_field = related_field
self.settings = settings
def get_recipients(self):
choice = getattr(self.settings, self.name)
if not choice:
return []
if choice == self.settings.EMAIL_FIELD:
return getattr(self.settings, self.name + '_emails')
role = self.settings.get_role_from_choice(choice)
if not role or not hasattr(role, 'role'):
return []
emails = role.role.emails
if role.role.emails_to_members:
emails.extend(role.user_set.values_list('email', flat=True))
return emails
@property
def display_value(self):
choice = getattr(self.settings, self.name)
if not choice:
return ''
if choice == self.settings.EMAIL_FIELD:
emails = getattr(self.settings, self.name + '_emails')
return ', '.join(emails)
role = self.settings.get_role_from_choice(choice)
if role:
display_name = getattr(self.settings, 'get_%s_display' % self.name)()
return '%s (%s)' % (display_name, role)
@property
def label(self):
return self.settings._meta.get_field(self.name).verbose_name
class AgendaNotificationsSettings(models.Model):
EMAIL_FIELD = 'use-email-field'
VIEW_ROLE = 'view-role'
EDIT_ROLE = 'edit-role'
CHOICES = [
(EDIT_ROLE, _('Edit Role')),
(VIEW_ROLE, _('View Role')),
(EMAIL_FIELD, _('Specify email addresses manually')),
]
agenda = models.OneToOneField(Agenda, on_delete=models.CASCADE, related_name='notifications_settings')
almost_full_event = models.CharField(
max_length=16, blank=True, choices=CHOICES, verbose_name=_('Almost full event (90%)')
)
almost_full_event_emails = ArrayField(models.EmailField(), blank=True, null=True)
full_event = models.CharField(max_length=16, blank=True, choices=CHOICES, verbose_name=_('Full event'))
full_event_emails = ArrayField(models.EmailField(), blank=True, null=True)
cancelled_event = models.CharField(
max_length=16, blank=True, choices=CHOICES, verbose_name=_('Cancelled event')
)
cancelled_event_emails = ArrayField(models.EmailField(), blank=True, null=True)
@classmethod
def get_email_field_names(cls):
return [field.name for field in cls._meta.get_fields() if isinstance(field, ArrayField)]
def get_notification_types(self):
for field in ['almost_full_event', 'full_event', 'cancelled_event']:
yield NotificationType(name=field, related_field=field.replace('_event', ''), settings=self)
def get_role_from_choice(self, choice):
if choice == self.EDIT_ROLE:
return self.agenda.edit_role
elif choice == self.VIEW_ROLE:
return self.agenda.view_role
@classmethod
def import_json(cls, data):
data = clean_import_data(cls, data)
return cls(**data)
def export_json(self):
return {
'almost_full_event': self.almost_full_event,
'almost_full_event_emails': self.almost_full_event_emails,
'full_event': self.full_event,
'full_event_emails': self.full_event_emails,
'cancelled_event': self.cancelled_event,
'cancelled_event_emails': self.cancelled_event_emails,
}

View File

@ -0,0 +1,7 @@
{% load i18n %}
{% blocktrans %}
Hi,
You have been notified because the status of event "{{ event }}" has changed.
{% endblocktrans %}
<a href="{{ event_url }}">{% trans "View event" %}</a>.

View File

@ -0,0 +1,9 @@
{% load i18n %}
{% autoescape off %}
{% blocktrans %}
Hi,
You have been notified because the status of event "{{ event }}" has changed.
You can view it here: {{ event_url }}.
{% endblocktrans %}
{% endautoescape %}

View File

@ -40,6 +40,7 @@ from chrono.agendas.models import (
VirtualMember,
Resource,
Category,
AgendaNotificationsSettings,
WEEKDAYS_LIST,
)
@ -508,3 +509,20 @@ class EventCancelForm(forms.ModelForm):
class Meta:
model = Event
fields = []
class AgendaNotificationsForm(forms.ModelForm):
class Meta:
model = AgendaNotificationsSettings
fields = '__all__'
widgets = {
'agenda': forms.HiddenInput(),
}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
for email_field in AgendaNotificationsSettings.get_email_field_names():
self.fields[email_field].widget.attrs['size'] = 80
self.fields[email_field].label = ''
self.fields[email_field].help_text = _('Enter a comma separated list of email addresses.')

View File

@ -0,0 +1,34 @@
{% extends "chrono/manager_agenda_view.html" %}
{% load i18n %}
{% block breadcrumb %}
{{ block.super }}
<a href="">{% trans "Notification settings" %}</a>
{% endblock %}
{% block appbar %}
<h2>{% trans "Notification settings" %}</h2>
{% endblock %}
{% block content %}
<form method="post" enctype="multipart/form-data">
{% csrf_token %}
{{ form.as_p }}
<div class="buttons">
<button class="submit-button">{% trans "Save" %}</button>
<a class="cancel" href="{% url 'chrono-manager-agenda-settings' pk=agenda.id %}">{% trans 'Cancel' %}</a>
</div>
<script>
$('select').change(function(){
role_field_id = $(this).attr('id')
email_field_id = '#' + role_field_id + '_emails'
if ($(this).val() == 'use-email-field')
$(email_field_id).parent('p').show();
else
$(email_field_id).parent('p').hide();
});
$('select').trigger('change');
</script>
</form>
{% endblock %}

View File

@ -30,4 +30,26 @@
</div>
</div>
<div class="section">
<h3>{% trans "Notifications" %}</h3>
<div>
<ul>
{% for notification_type in object.notifications_settings.get_notification_types %}
{% with display_value=notification_type.display_value label=notification_type.label %}
{% if display_value %}
<li>
{% blocktrans %}
{{ label }}: {{ display_value }} will be notified.
{% endblocktrans %}
</li>
{% endif %}
{% endwith %}
{% empty %}
{% trans "Notifications are disabled for this agenda." %}
{% endfor %}
</ul>
<a rel="popup" href="{% url 'chrono-manager-agenda-notifications-settings' pk=object.id %}">{% trans 'Configure' %}</a>
</div>
</div>
{% endblock %}

View File

@ -73,6 +73,11 @@ urlpatterns = [
views.agenda_import_events,
name='chrono-manager-agenda-import-events',
),
url(
r'^agendas/(?P<pk>\d+)/notifications$',
views.agenda_notifications_settings,
name='chrono-manager-agenda-notifications-settings',
),
url(
r'^agendas/(?P<pk>\d+)/events/(?P<event_pk>\d+)/$',
views.event_view,

View File

@ -23,7 +23,7 @@ import uuid
from django.contrib import messages
from django.core.exceptions import PermissionDenied
from django.db.models import Q
from django.db.models import Q, F
from django.db.models import Min, Max
from django.http import Http404, HttpResponse, HttpResponseRedirect
from django.shortcuts import get_object_or_404
@ -63,6 +63,7 @@ from chrono.agendas.models import (
Resource,
Category,
EventCancellationReport,
AgendaNotificationsSettings,
)
from .forms import (
@ -90,6 +91,7 @@ from .forms import (
CategoryEditForm,
BookingCancelForm,
EventCancelForm,
AgendaNotificationsForm,
)
from .utils import import_site
@ -1345,6 +1347,28 @@ class AgendaImportEventsView(ManagedAgendaMixin, FormView):
agenda_import_events = AgendaImportEventsView.as_view()
class AgendaNotificationsSettingsView(ManagedAgendaMixin, UpdateView):
template_name = 'chrono/manager_agenda_notifications_form.html'
model = AgendaNotificationsSettings
form_class = AgendaNotificationsForm
def get_object(self):
try:
return self.agenda.notifications_settings
except AgendaNotificationsSettings.DoesNotExist:
# prevent old events from sending notifications
statuses = ('almost_full', 'full', 'cancelled')
timestamp = now()
for status in statuses:
filter_kwargs = {status: True}
update_kwargs = {status + '_notification_timestamp': timestamp}
self.agenda.event_set.filter(**filter_kwargs).update(**update_kwargs)
return AgendaNotificationsSettings.objects.create(agenda=self.agenda)
agenda_notifications_settings = AgendaNotificationsSettingsView.as_view()
class EventDetailView(ViewableAgendaMixin, DetailView):
model = Event
pk_url_kwarg = 'event_pk'

View File

@ -175,6 +175,8 @@ EXCEPTIONS_SOURCES = {
'holidays': {'class': 'workalendar.europe.France', 'label': _('Holidays')},
}
TEMPLATE_VARS = {}
local_settings_file = os.environ.get(
'CHRONO_SETTINGS_FILE', os.path.join(os.path.dirname(__file__), 'local_settings.py')
)

View File

@ -27,3 +27,5 @@ KNOWN_SERVICES = {
}
EXCEPTIONS_SOURCES = {}
SITE_BASE_URL = 'https://example.com'

View File

@ -4,7 +4,7 @@ import mock
import requests
from django.contrib.auth.models import Group
from django.contrib.auth.models import Group, User
from django.core.files.base import ContentFile
from django.core.management import call_command
from django.test import override_settings
@ -24,6 +24,7 @@ from chrono.agendas.models import (
TimePeriodExceptionSource,
VirtualMember,
EventCancellationReport,
AgendaNotificationsSettings,
)
pytestmark = pytest.mark.django_db
@ -1145,3 +1146,93 @@ def test_agendas_cancel_events_command_network_error(freezer):
freezer.move_to('2020-03-01')
call_command('cancel_events')
assert not EventCancellationReport.objects.exists()
@mock.patch('django.contrib.auth.models.Group.role', create=True)
@pytest.mark.parametrize(
'emails_to_members,emails',
[(False, []), (False, ['test@entrouvert.com']), (True, []), (True, ['test@entrouvert.com']),],
)
def test_agenda_notifications_role_email(mocked_role, emails_to_members, emails, mailoutbox):
group = Group.objects.create(name='group')
user = User.objects.create(username='user', email='user@entrouvert.com')
user.groups.add(group)
mocked_role.emails_to_members = emails_to_members
mocked_role.emails = emails
expected_recipients = emails
if emails_to_members:
expected_recipients.append(user.email)
expected_email_count = 1 if emails else 0
agenda = Agenda.objects.create(label='Foo bar', kind='event', edit_role=group)
event = Event.objects.create(agenda=agenda, places=10, start_datetime=now(), label='Hop')
settings = AgendaNotificationsSettings.objects.create(agenda=agenda)
settings.almost_full_event = AgendaNotificationsSettings.EDIT_ROLE
settings.save()
# book 9/10 places to reach almost full state
for i in range(9):
Booking.objects.create(event=event)
event.refresh_from_db()
assert event.almost_full
call_command('send_email_notifications')
assert len(mailoutbox) == expected_email_count
if mailoutbox:
assert mailoutbox[0].recipients() == expected_recipients
assert mailoutbox[0].subject == 'Alert: event "Hop" is almost full (90%)'
# no new email on subsequent run
call_command('send_email_notifications')
assert len(mailoutbox) == expected_email_count
def test_agenda_notifications_email_list(mailoutbox):
agenda = Agenda.objects.create(label='Foo bar', kind='event')
event = Event.objects.create(agenda=agenda, places=10, start_datetime=now(), label='Hop')
settings = AgendaNotificationsSettings.objects.create(agenda=agenda)
settings.full_event = AgendaNotificationsSettings.EMAIL_FIELD
settings.full_event_emails = recipients = ['hop@entrouvert.com', 'top@entrouvert.com']
settings.save()
for i in range(10):
Booking.objects.create(event=event)
event.refresh_from_db()
assert event.full
call_command('send_email_notifications')
assert len(mailoutbox) == 1
assert mailoutbox[0].recipients() == recipients
assert mailoutbox[0].subject == 'Alert: event "Hop" is full'
assert (
'view it here: https://example.com/manage/agendas/%s/events/%s/' % (agenda.pk, event.pk,)
in mailoutbox[0].body
)
# no new email on subsequent run
call_command('send_email_notifications')
assert len(mailoutbox) == 1
def test_agenda_notifications_cancelled(mailoutbox):
agenda = Agenda.objects.create(label='Foo bar', kind='event')
event = Event.objects.create(agenda=agenda, places=10, start_datetime=now(), label='Hop')
settings = AgendaNotificationsSettings.objects.create(agenda=agenda)
settings.cancelled_event = AgendaNotificationsSettings.EMAIL_FIELD
settings.cancelled_event_emails = recipients = ['hop@entrouvert.com', 'top@entrouvert.com']
settings.save()
event.cancelled = True
event.save()
call_command('send_email_notifications')
assert len(mailoutbox) == 1
assert mailoutbox[0].recipients() == recipients
assert mailoutbox[0].subject == 'Alert: event "Hop" is cancelled'
# no new email on subsequent run
call_command('send_email_notifications')
assert len(mailoutbox) == 1

View File

@ -28,6 +28,7 @@ from chrono.agendas.models import (
AgendaImportError,
MeetingType,
VirtualMember,
AgendaNotificationsSettings,
)
from chrono.manager.utils import import_site
@ -456,3 +457,29 @@ def test_import_export_slug_fields(app):
with pytest.raises(AgendaImportError) as excinfo:
import_site(payload)
assert str(excinfo.value) == 'Bad slug format "meeting-type&"'
def test_import_export_notification_settings():
agenda = Agenda.objects.create(label='Foo bar', kind='events')
settings = AgendaNotificationsSettings.objects.create(
agenda=agenda,
almost_full_event=AgendaNotificationsSettings.EDIT_ROLE,
full_event=AgendaNotificationsSettings.VIEW_ROLE,
cancelled_event=AgendaNotificationsSettings.EMAIL_FIELD,
cancelled_event_emails=['hop@entrouvert.com', 'top@entrouvert.com'],
)
output = get_output_of_command('export_site')
payload = json.loads(output)
agenda.delete()
assert not AgendaNotificationsSettings.objects.exists()
import_site(payload)
agenda = Agenda.objects.first()
AgendaNotificationsSettings.objects.get(
agenda=agenda,
almost_full_event=AgendaNotificationsSettings.EDIT_ROLE,
full_event=AgendaNotificationsSettings.VIEW_ROLE,
cancelled_event=AgendaNotificationsSettings.EMAIL_FIELD,
cancelled_event_emails=['hop@entrouvert.com', 'top@entrouvert.com'],
)

View File

@ -52,15 +52,18 @@ def simple_user():
@pytest.fixture
def manager_user():
def managers_group():
group, _ = Group.objects.get_or_create(name='Managers')
return group
@pytest.fixture
def manager_user(managers_group):
try:
user = User.objects.get(username='manager')
except User.DoesNotExist:
user = User.objects.create_user('manager', password='manager')
group, created = Group.objects.get_or_create(name='Managers')
if created:
group.save()
user.groups.set([group])
user.groups.set([managers_group])
return user
@ -3892,3 +3895,62 @@ def test_event_cancellation_forbidden(app, admin_user):
resp = resp.click('Cancel', href='/cancel')
assert 'event has bookings with no callback url configured' in resp.text
assert 'Proceed with cancellation' not in resp.text
def test_agenda_notifications(app, admin_user, managers_group):
agenda = Agenda.objects.create(label='Events', kind='events')
login(app)
resp = app.get('/manage/agendas/%s/settings' % agenda.id)
assert 'Notifications' in resp.text
assert 'Notifications are disabled' in resp.text
resp = resp.click('Configure')
resp.form['almost_full_event'] = 'edit-role'
resp.form['full_event'] = 'view-role'
resp.form['cancelled_event'] = 'use-email-field'
resp.form['cancelled_event_emails'] = 'hop@entrouvert.com, top@entrouvert.com'
resp = resp.form.submit().follow()
settings = agenda.notifications_settings
assert settings.almost_full_event == 'edit-role'
assert settings.full_event == 'view-role'
assert settings.cancelled_event == 'use-email-field'
assert settings.cancelled_event_emails == ['hop@entrouvert.com', 'top@entrouvert.com']
assert 'Cancelled event: hop@entrouvert.com, top@entrouvert.com will be notified' in resp.text
assert not 'Full event:' in resp.text
assert not 'Almost full event (90%):' in resp.text
agenda.view_role = managers_group
agenda.edit_role = Group.objects.create(name='hop')
agenda.save()
resp = app.get('/manage/agendas/%s/settings' % agenda.id)
assert 'Almost full event (90%): Edit Role (hop) will be notified' in resp.text
assert 'Full event: View Role (Managers) will be notified' in resp.text
def test_agenda_notifications_no_old_events(app, admin_user, mailoutbox):
agenda = Agenda.objects.create(label='Events', kind='events')
event = Event.objects.create(agenda=agenda, start_datetime=now(), places=10, label='Old event')
event.cancelled = True
event.save()
login(app)
resp = app.get('/manage/agendas/%s/settings' % agenda.id)
resp = resp.click('Configure')
resp.form['cancelled_event'] = 'use-email-field'
resp.form['cancelled_event_emails'] = 'hop@entrouvert.com'
resp.form.submit()
event = Event.objects.create(agenda=agenda, start_datetime=now(), places=10, label='New event')
event.cancelled = True
event.save()
call_command('send_email_notifications')
# no notification is sent for old event
assert len(mailoutbox) == 1
assert 'New event' in mailoutbox[0].subject