diff --git a/changelog.d/+cleanup-destination-plugins.changed.md b/changelog.d/+cleanup-destination-plugins.changed.md
new file mode 100644
index 000000000..252ad3857
--- /dev/null
+++ b/changelog.d/+cleanup-destination-plugins.changed.md
@@ -0,0 +1,6 @@
+Moved a lot of common infrastructure from our NotificationMedium subclasses to
+the parent. This should make it easier to create more media of high quality.
+Also, it should make it easier to use the plugins for validating the
+settings-file elsewhere than just the API.
+
+This might break 3rd party notification plugins.
diff --git a/changelog.d/+destination-with-forms.changed.md b/changelog.d/+destination-with-forms.changed.md
new file mode 100644
index 000000000..9bd8ad11e
--- /dev/null
+++ b/changelog.d/+destination-with-forms.changed.md
@@ -0,0 +1 @@
+Htmx: Change the destinations page to validate with forms
diff --git a/docs/integrations/notifications/writing-notification-plugins.rst b/docs/integrations/notifications/writing-notification-plugins.rst
index eb4bacd40..fb1268105 100644
--- a/docs/integrations/notifications/writing-notification-plugins.rst
+++ b/docs/integrations/notifications/writing-notification-plugins.rst
@@ -28,22 +28,40 @@ implement the following:
Class constants
---------------
-You need to set the constants `MEDIA_SLUG`, `MEDIA_NAME` and
-`MEDIA_JSON_SCHEMA`.
-
-The media name is the name of the service you want to send notifications by.
-This is used only for display purposes so you might want to keep it short and
-sweet. So for example `Email`, `SMS` or `MS Teams`.
-
-The media slug is the slugified version of that, so the name simplified to only
-contain lowercase letters, numbers, underscores and hyphens. Always have it
-start with a letter, a-z. For example `email`, `sms` or `msteams`.
-
-The media `json schema `_ is a representation of how
-a destination that will be used by this notification plugin should look like.
-Such a destination should include all necessary information that is needed to
-send notifications with your notification plugin. In case of SMS that is a
-phone number or for MS Teams a webhook.
+You must set the constants ``MEDIA_SLUG`, `MEDIA_NAME`` and
+``MEDIA_JSON_SCHEMA``. If your plugin only takes or needs a single
+configuration flag you should also set ``MEDIA_SETTINGS_KEY``.
+
+MEDIA_NAME
+ The media name is the name of the service you want to send notifications by.
+ This is used only for display purposes so you might want to keep it short
+ and sweet. So for example ``"Email"``, ``"SMS"`` or ``"MS Teams"``.
+
+MEDIA_SLUG
+ The media slug is the slugified version of that, so the name simplified to
+ only contain lowercase letters, numbers, underscores and hyphens. Always
+ have it start with a letter, a-z. For example ``"email"``, ``"sms"`` or
+ ``"msteams"``.
+
+MEDIA_JSON_SCHEMA
+ The media `json schema `_ is a representation of
+ how a destination that will be used by this notification plugin should look
+ like, so that it is possible to autogenerate a form with javascript. It will
+ be accessible via the API. Such a destination should include all necessary
+ information that is needed to send notifications with your notification
+ plugin. In case of SMS that is a phone number or for MS Teams a webhook.
+
+MEDIA_SETTINGS_KEY
+ The media settings key is the name of the most important key in the settings
+ JSON field. It is used to cut down on the amount of code you need to write
+ if there is only one piece of config you need to send the notification.
+ Among other things, it is used to check for duplicate entries, so in a way
+ it acts as the primary key for your plugin. For that reason, it must be
+ required in the json schema. For example for an email plugin this would be
+ "email_address".
+
+Form
+ The ``forms.Form`` used to validate the settings-field.
Class methods for sending notifications
---------------------------------------
@@ -51,6 +69,8 @@ Class methods for sending notifications
.. autoclass:: argus.notificationprofile.media.base.NotificationMedium
:members: send
+This MUST be overridden.
+
The ``send`` method is the method that does the actual sending of the
notification. It gets the Argus event and a list of destinations as input and
returns a boolean indicating if the sending was successful.
@@ -62,35 +82,73 @@ The rest is very dependent on the notification medium and, if used, the Python
library. The given event can be used to extract relevant information that
should be included in the message that will be sent to each destination.
-Class methods for destinations
-------------------------------
+Helper class methods
+--------------------
.. autoclass:: argus.notificationprofile.media.base.NotificationMedium
:members: get_label, has_duplicate, raise_if_not_deletable, update, validate
:noindex:
-Your implementation of ``get_label`` should show a reasonable representation
-for a destination of that type that makes it easy to identify. For SMS that
-would simply be the phone number.
-
-The method ``has_duplicate`` will receive a QuerySet of destinations and a dict
-of settings for a possible destination and should return True if a destination
-with such settings exists in the given QuerySet.
-
-``raise_if_not_deletable`` should check if a given destination can be deleted.
-This is used in case some destinations are synced from an outside source and
-should not be able to be deleted by a user. If that is the case a
-``NotDeletableError`` should be raised. If not simply return None.
-
-The method ``update`` only has to be implemented if the regular update method
-of Django isn't sufficient. This can be the case if additional settings need to
-be updated.
-
-Finally the function ``validate`` makes sure that a destination with the given
-settings can be updated or created. The function ``has_duplicate`` can be used
-here to ensure that not two destinations with the same settings will be
-created. Additionally the settings themselves should also be validated. For
-example for SMS the given phone number will be checked. Django forms can be
-helpful for validation. A ``ValidationError`` should be raised if the given
-settings are invalid and the validated and cleaned data should be returned if
-not.
+With a little luck you might not need to override any of these.
+
+clean
+ This method will do any additional cleaning beyond what is defined by the
+ defined ``Form``. Expects a valid form instance and optional
+ DestinationConfig instance, and returns the updated valid form instance. If
+ you have fields that shouldn't be set by a user, or values that need extra
+ conversion, you can do that in this method. Use the passed in instance if
+ you need to fall back to defaults. This method should not be used to
+ validate anything and thus should never raise a validation Exception.
+
+get_label
+ Your implementation of ``get_label`` should show a reasonable representation
+ for a destination of that type that makes it easy to identify. For SMS that
+ would simply be the phone number. By default it shows the label stored in
+ the destination. If no label has been set, it uses MEDIA_SETTINGS_KEY to
+ look up the most important piece of information in the settings and uses
+ that directly. The included plugins need not override ``get_label`` for this
+ reason. If the label would be very long, for instance if the needed setting
+ is a very long url (40+ characters), you ought to write your own
+ ``get_label``.
+
+get_relevant_destination_settings
+ Used by ``send``. You should only need to override this if the key in
+ MEDIA_SETTINGS_KEY is insuffcient to look up the actual configuration of the
+ destinations of the type set by MEDIA_SLUG.
+
+has_duplicate
+ The method ``has_duplicate`` will receive a QuerySet of destinations and
+ a dict of settings for a possible destination and should return True if
+ a destination with such settings exists in the given QuerySet. By default it
+ will use MEDIA_SETTINGS_KEY to lookup the most important piece of
+ information in the settings.
+
+raise_if_not_deletable
+ ``raise_if_not_deletable`` should check if a given destination can be
+ deleted. This is necessary in case the destination is in use by a profile,
+ or some destinations are synced from an outside source or otherwise
+ auto-generated, and should not be able to be deleted by a user. If that is
+ the case a ``NotDeletableError`` should be raised. If not simply return
+ None.
+
+update
+ The method ``update`` only has to be implemented if the regular update
+ method is insufficient. This can be the case if there is more than one
+ key-value pair in settings that need to be updated.
+
+validate
+ The function ``validate`` makes sure that a destination with the given
+ settings can be updated or created. It uses the ``validate_settings`` method
+ to validate the settings-field, a form (CommonDestinationConfigForm) to
+ validate the media and label-fields, and an optional DestinationConfig
+ instance for the sake of the ``clean``-method. The validated form is
+ returned if ok, otherwise a ``ValidationError`` should be raised. It is
+ unlikely that you will ever need to override this method.
+
+validate_settings
+ This method validates the actual contents of the settings-field using the
+ ``Form`` that is defined and an optional DestinationConfig instance for the
+ sake of the ``clean``-method. The function ``has_duplicate`` can be used
+ here to ensure that no two destinations with the same settings will be
+ created. A ``ValidationError`` should be raised if the given settings are
+ invalid, and the validated and cleaned data should be returned if not.
diff --git a/src/argus/htmx/destination/forms.py b/src/argus/htmx/destination/forms.py
deleted file mode 100644
index fb50ac073..000000000
--- a/src/argus/htmx/destination/forms.py
+++ /dev/null
@@ -1,126 +0,0 @@
-from django import forms
-from django.forms import ModelForm
-
-from argus.notificationprofile.models import DestinationConfig, Media
-from argus.notificationprofile.serializers import RequestDestinationConfigSerializer
-from argus.notificationprofile.media import api_safely_get_medium_object
-
-
-class DestinationFormCreate(ModelForm):
- settings = forms.CharField(required=True)
-
- def __init__(self, *args, **kwargs):
- # Serializer request the request object
- self.request = kwargs.pop("request", None)
- super().__init__(*args, **kwargs)
-
- class Meta:
- model = DestinationConfig
- fields = ["label", "media", "settings"]
- labels = {
- "label": "Name",
- }
- widgets = {
- "media": forms.Select(attrs={"class": "select input-bordered w-full max-w-xs"}),
- }
-
- def clean(self):
- super().clean()
- settings_key = _get_settings_key_for_media(self.cleaned_data["media"])
- # Convert settings value (e.g. email address) to be compatible with JSONField
- self.cleaned_data["settings"] = {settings_key: self.cleaned_data["settings"]}
- self._init_serializer()
- return self._validate_serializer()
-
- def save(self):
- # self.serializer should be initiated and validated in clean() before save() is called
- self.serializer.save(user=self.request.user)
-
- def _init_serializer(self):
- serializer = RequestDestinationConfigSerializer(
- data={
- "media": self.cleaned_data["media"],
- "label": self.cleaned_data.get("label", ""),
- "settings": self.cleaned_data["settings"],
- },
- context={"request": self.request},
- )
- self.serializer = serializer
-
- def _validate_serializer(self):
- media = self.cleaned_data["media"]
- settings_key = _get_settings_key_for_media(media)
-
- # Add error messages from serializer to form
- if not self.serializer.is_valid():
- for error_name, error_detail in self.serializer.errors.items():
- if error_name in ["media", "label", settings_key]:
- if error_name == settings_key:
- error_name = "settings"
- self.add_error(error_name, error_detail)
- # Serializer might add more data to the JSON dict
- if settings := self.serializer.data.get("settings"):
- self.cleaned_data["settings"] = settings
- else:
- # Serializer might add more data to the JSON dict
- if settings := self.serializer.validated_data.get("settings"):
- self.cleaned_data["settings"] = settings
-
- if label := self.cleaned_data.get("label"):
- destination_filter = DestinationConfig.objects.filter(label=label, media=media)
- if self.instance:
- destination_filter = destination_filter.exclude(pk=self.instance.pk)
- if destination_filter.exists():
- self.add_error("label", "Name must be unique per media")
-
- return self.cleaned_data
-
-
-class DestinationFormUpdate(DestinationFormCreate):
- def __init__(self, *args, **kwargs):
- if instance := kwargs.get("instance"):
- settings_key = _get_settings_key_for_media(instance.media)
- # Extract settings value (email address etc.) from JSONField
- instance.settings = instance.settings.get(settings_key)
- super().__init__(*args, **kwargs)
-
- class Meta:
- model = DestinationConfig
- fields = ["label", "media", "settings"]
- labels = {
- "label": "Name",
- }
- widgets = {
- "media": forms.HiddenInput(),
- }
-
- def _init_serializer(self):
- # self.instance is modified in __init__,
- # so get unmodified version here for the serializer
- destination = DestinationConfig.objects.get(pk=self.instance.pk)
- settings_key = _get_settings_key_for_media(destination.media)
- data = {}
-
- if "label" in self.cleaned_data:
- label = self.cleaned_data["label"]
- if label != destination.label:
- data["label"] = label
-
- settings = self.cleaned_data["settings"]
- if settings.get(settings_key) != destination.settings.get(settings_key):
- data["settings"] = settings
-
- self.serializer = RequestDestinationConfigSerializer(
- destination,
- data=data,
- context={"request": self.request},
- partial=True,
- )
-
-
-def _get_settings_key_for_media(media: Media) -> str:
- """Returns the required settings key for the given media,
- e.g. "email_address", "phone_number"
- """
- medium = api_safely_get_medium_object(media.slug)
- return medium.MEDIA_JSON_SCHEMA["required"][0]
diff --git a/src/argus/htmx/destination/urls.py b/src/argus/htmx/destination/urls.py
index 7ffea3ffa..25d7f2b45 100644
--- a/src/argus/htmx/destination/urls.py
+++ b/src/argus/htmx/destination/urls.py
@@ -1,11 +1,11 @@
from django.urls import path
-from .views import destination_list, create_htmx, delete_htmx, update_htmx
+from .views import destination_list, create_destination, delete_destination, update_destination
-app_name = "htmx"
+app_name = "destination"
urlpatterns = [
path("", destination_list, name="destination-list"),
- path("htmx-create/", create_htmx, name="htmx-create"),
- path("/htmx-delete/", delete_htmx, name="htmx-delete"),
- path("/htmx-update/", update_htmx, name="htmx-update"),
+ path("/create/", create_destination, name="destination-create"),
+ path("/delete/", delete_destination, name="destination-delete"),
+ path("-/update/", update_destination, name="destination-update"),
]
diff --git a/src/argus/htmx/destination/views.py b/src/argus/htmx/destination/views.py
index df3c102c4..d8dd89c34 100644
--- a/src/argus/htmx/destination/views.py
+++ b/src/argus/htmx/destination/views.py
@@ -1,14 +1,92 @@
-from typing import Optional, Sequence
-from django.shortcuts import render, get_object_or_404
+from __future__ import annotations
-from django.views.decorators.http import require_http_methods
+from collections import namedtuple
+import logging
+from typing import Optional, TYPE_CHECKING, Tuple
+
+from django.http import QueryDict
+from django.contrib import messages
from django.http import HttpResponse
+from django.shortcuts import redirect, render, get_object_or_404
+from django.views.decorators.http import require_http_methods
from argus.notificationprofile.models import DestinationConfig, Media
-from argus.notificationprofile.media import api_safely_get_medium_object
-from argus.notificationprofile.media.base import NotificationMedium
+from argus.notificationprofile.media import get_medium_object
+from argus.notificationprofile.media.base import NotificationMedium, LabelForm
+
+if TYPE_CHECKING:
+ from django.forms import Form
+
+
+LOG = logging.getLogger(__name__)
+Forms = namedtuple("Forms", ["label_form", "settings_form"])
+
+
+# not a view
+def get_forms(request, media: str, instance: Optional[DestinationConfig] = None) -> Forms:
+ """Get and bind the two forms necessary to change a destination
+
+ - media is for selecting the right plugin
+ - label_form is for all the common fields of a destination
+ - settings_form is fetched from the destination plugin found via "media"
+ - instance holds the pre-existing values if updating
+ """
+ prefix = "destinationform"
+ medium = get_medium_object(media)
+
+ if instance and instance.pk:
+ prefix += f"-{instance.pk}-{media}"
+ label_name = medium.get_label(instance)
+ label_initial = {"label": label_name}
+ else:
+ prefix += f"-{media}"
+ label_initial = {}
-from .forms import DestinationFormCreate, DestinationFormUpdate
+ data = None
+ for key in request.POST:
+ if key.startswith(prefix):
+ data = request.POST
+ break
+ label_form = LabelForm(data=data, user=request.user, initial=label_initial, instance=instance, prefix=prefix)
+ settings_form = medium.validate_settings(data, request.user, instance=instance, prefix=prefix)
+
+ if data:
+ label_form.is_valid()
+ return Forms(label_form, settings_form)
+
+
+# not a view
+def save_forms(user, media: str, label_form: LabelForm, settings_form: Form) -> Tuple[DestinationConfig, bool]:
+ """Save the contents of the two forms necessary to change a destination
+
+ The two forms should first have been instantiated via ``get_forms``.
+
+ - media is for linking up the right plugin on create
+ - label_form is for all the common fields of a destination
+ - settings_form is fetched from the destination plugin found via "media"
+
+ if the forms are valid, return the (updated) DestinationConfig and a bool
+ of whether the object was changed or not. If the forms were not valid,
+ return (None, false)
+ """
+ changed = False
+ obj = label_form.instance
+ if obj.pk:
+ # ensure that media type cannot be changed when updating a destination
+ media = obj.media_id
+ if label_form.has_changed() or settings_form.has_changed():
+ changed = True
+ else:
+ return obj, changed
+ if not (label_form.is_valid() and settings_form.is_valid()):
+ obj = obj or None
+ return obj, changed
+ obj = label_form.save(commit=False)
+ obj.user = user
+ obj.media_id = media
+ obj.settings = settings_form.cleaned_data
+ obj.save()
+ return obj, changed
@require_http_methods(["GET"])
@@ -17,118 +95,122 @@ def destination_list(request):
@require_http_methods(["POST"])
-def create_htmx(request) -> HttpResponse:
- form = DestinationFormCreate(request.POST or None, request=request)
+def create_destination(request, media: str) -> HttpResponse:
+ medium = get_medium_object(media)
+ label_form, settings_form = get_forms(request, media)
template = "htmx/destination/_content.html"
- if form.is_valid():
- form.save()
- return _render_destination_list(request, template=template)
- return _render_destination_list(request, create_form=form, template=template)
+ context = {
+ "label_form": label_form,
+ "settings_form": settings_form,
+ "media": media,
+ }
+ obj, changed = save_forms(request.user, media, label_form, settings_form)
+ if changed:
+ if obj:
+ label = medium.get_label(obj)
+ message = f'Created new {medium.MEDIA_NAME} destination "{label}"'
+ messages.success(request, message)
+ LOG.info(message)
+ request.POST = QueryDict("")
+ return _render_destination_list(request, context=context, template=template)
+ error_msg = f"Could not create new {medium.MEDIA_NAME} destination"
+ messages.warning(request, error_msg)
+ LOG.warn(error_msg)
+ return _render_destination_list(request, context=context, template=template)
@require_http_methods(["POST"])
-def delete_htmx(request, pk: int) -> HttpResponse:
+def delete_destination(request, pk: int) -> HttpResponse:
destination = get_object_or_404(request.user.destinations.all(), pk=pk)
media = destination.media
error_msg = None
+ medium = get_medium_object(media.slug)
+ destination_label = medium.get_label(destination)
try:
- medium = api_safely_get_medium_object(destination.media.slug)
medium.raise_if_not_deletable(destination)
except NotificationMedium.NotDeletableError as e:
- error_msg = " ".join(e.args)
+ # template?
+ error_msg = ", ".join(e.args)
+ message = f'Failed to delete {medium.MEDIA_NAME} destination "{destination}": {error_msg}'
+ messages.warning(request, message)
+ LOG.warn(message)
else:
destination.delete()
+ message = f'Deleted {medium.MEDIA_NAME} destination "{destination_label}"'
+ messages.success(request, message)
+ LOG.info(message)
- forms = _get_update_forms(request.user, media=media)
-
- context = {
- "error_msg": error_msg,
- "forms": forms,
- "media": media,
- }
- return render(request, "htmx/destination/_collapse_with_forms.html", context=context)
+ return redirect("htmx:destination-list")
@require_http_methods(["POST"])
-def update_htmx(request, pk: int) -> HttpResponse:
- destination = DestinationConfig.objects.get(pk=pk)
- media = destination.media
- form = DestinationFormUpdate(request.POST or None, instance=destination, request=request)
- if is_valid := form.is_valid():
- form.save()
-
- update_forms = _get_update_forms(request.user, media=media)
-
- if not is_valid:
- update_forms = _replace_form_in_list(update_forms, form)
-
+def update_destination(request, pk: int, media: str) -> HttpResponse:
+ medium = get_medium_object(media)
+ template = "htmx/destination/_form_list.html"
+ destination = get_object_or_404(request.user.destinations.all(), pk=pk)
+ label = medium.get_label(destination)
+ forms = get_forms(request, media, instance=destination)
+ obj, changed = save_forms(request.user, media, *forms)
+ if changed:
+ if obj:
+ label = medium.get_label(obj)
+ message = f'Updated {medium.MEDIA_NAME} destination "{label}"'
+ messages.success(request, message)
+ LOG.info(message)
+ request.POST = QueryDict("")
+ return _render_destination_list(request, template=template)
+ else:
+ return _render_destination_list(request, template=template)
+ error_msg = f'Could not update {medium.MEDIA_NAME} destination "{label}"'
+ messages.warning(request, error_msg)
+ LOG.warn(request, error_msg)
+ all_forms = get_all_forms_grouped_by_media(request)
+ update_forms = get_all_update_forms_for_media(request, media)
+ for index, forms in enumerate(update_forms):
+ if forms.label_form.instance.pk == pk:
+ update_forms[index] = forms
+ break
+ all_forms[media] = update_forms
context = {
- "error_msg": None,
- "forms": update_forms,
+ "forms": all_forms,
+ "label_form": forms.label_form,
+ "settings_form": forms.settings_form,
"media": media,
}
- return render(request, "htmx/destination/_collapse_with_forms.html", context=context)
+ return _render_destination_list(request, context=context, template=template)
def _render_destination_list(
request,
- create_form: Optional[DestinationFormCreate] = None,
- update_forms: Optional[Sequence[DestinationFormUpdate]] = None,
+ context: Optional[dict] = None,
template: str = "htmx/destination/destination_list.html",
) -> HttpResponse:
- """Function to render the destinations page.
-
- :param create_form: this is used to display the form for creating a new destination
- with errors while retaining the user input. If you want a blank form, pass None.
- :param update_forms: list of update forms to display. Useful for rendering forms
- with error messages while retaining the user input.
- If this is None, the update forms will be generated from the user's destinations."""
-
- if create_form is None:
- create_form = DestinationFormCreate()
- if update_forms is None:
- update_forms = _get_update_forms(request.user)
- grouped_forms = _group_update_forms_by_media(update_forms)
- context = {
- "create_form": create_form,
- "grouped_forms": grouped_forms,
- "page_title": "Destinations",
- }
+ """Function to render the destinations page"""
+
+ if not context:
+ context = {}
+ if "forms" not in context:
+ context["forms"] = get_all_forms_grouped_by_media(request)
+ context["page_title"] = "Destinations"
return render(request, template, context=context)
-def _get_update_forms(user, media: Media = None) -> list[DestinationFormUpdate]:
+def get_all_update_forms_for_media(request, media: str) -> list[Forms]:
"""Get a list of update forms for the user's destinations.
- :param media: if provided, only return destinations for this media.
+ :param media: Only return destinations for this media.
"""
- if media:
- destinations = user.destinations.filter(media=media)
- else:
- destinations = user.destinations.all()
- # Sort by oldest first
- destinations = destinations.order_by("pk")
- return [DestinationFormUpdate(instance=destination) for destination in destinations]
+ destinations = request.user.destinations.filter(media_id=media).order_by("pk")
+ return [get_forms(request, media, instance=destination) for destination in destinations]
-def _group_update_forms_by_media(
- destination_forms: Sequence[DestinationFormUpdate],
-) -> dict[Media, list[DestinationFormUpdate]]:
- grouped_destinations = {}
- # Adding a media to the dict even if there are no destinations for it
- # is useful so that the template can render a section for that media
+def get_all_forms_grouped_by_media(request):
+ forms = {}
for media in Media.objects.all():
- grouped_destinations[media] = []
-
- for form in destination_forms:
- grouped_destinations[form.instance.media].append(form)
-
- return grouped_destinations
-
-
-def _replace_form_in_list(forms: list[DestinationFormUpdate], form: DestinationFormUpdate):
- for index, f in enumerate(forms):
- if f.instance.pk == form.instance.pk:
- forms[index] = form
- break
+ create_form = get_forms(request, media.slug)
+ update_forms = get_all_update_forms_for_media(request, media.slug)
+ forms[media] = {
+ "create_form": create_form,
+ "update_forms": update_forms,
+ }
return forms
diff --git a/src/argus/htmx/templates/htmx/destination/_collapse_with_forms.html b/src/argus/htmx/templates/htmx/destination/_collapse_with_forms.html
index 03f7610e2..f284ac6c9 100644
--- a/src/argus/htmx/templates/htmx/destination/_collapse_with_forms.html
+++ b/src/argus/htmx/templates/htmx/destination/_collapse_with_forms.html
@@ -1,8 +1,13 @@
- {{ media.name }} ({{ forms|length }})
-
- {% for media, forms in grouped_forms.items %}
+ {% for media, forms in forms.items %}
{% include "htmx/destination/_collapse_with_forms.html" with error_msg=None %}
{% endfor %}
diff --git a/src/argus/htmx/templates/htmx/destination/_non_field_form_errors.html b/src/argus/htmx/templates/htmx/destination/_non_field_form_errors.html
new file mode 100644
index 000000000..6288eed24
--- /dev/null
+++ b/src/argus/htmx/templates/htmx/destination/_non_field_form_errors.html
@@ -0,0 +1,7 @@
+{% if label_form.non_field_errors or settings_form.non_field_errors %}
+
+ {% if error_msg %}
{{ error }}
{% endif %}
+ {% for error in label_form.non_field_errors %}
{{ error }}
{% endfor %}
+ {% for error in settings_form.non_field_errors %}
{{ error }}
{% endfor %}
+
+{% endif %}
diff --git a/src/argus/htmx/templates/htmx/destination/destination_create.html b/src/argus/htmx/templates/htmx/destination/destination_create.html
new file mode 100644
index 000000000..8dcf97a3f
--- /dev/null
+++ b/src/argus/htmx/templates/htmx/destination/destination_create.html
@@ -0,0 +1,4 @@
+{% extends "htmx/base.html" %}
+{% block main %}
+ {% include "htmx/destination/_create_form.html" %}
+{% endblock main %}
diff --git a/src/argus/notificationprofile/media/__init__.py b/src/argus/notificationprofile/media/__init__.py
index 6e1773425..f7fdcc896 100644
--- a/src/argus/notificationprofile/media/__init__.py
+++ b/src/argus/notificationprofile/media/__init__.py
@@ -24,7 +24,6 @@
LOG = logging.getLogger(__name__)
-
__all__ = [
"api_safely_get_medium_object",
"send_notification",
@@ -42,6 +41,17 @@
MEDIA_CLASSES_DICT = {media_class.MEDIA_SLUG: media_class for media_class in _media_classes}
+class DestinationPluginError(Exception):
+ pass
+
+
+def get_medium_object(media_slug: str):
+ try:
+ return MEDIA_CLASSES_DICT[str(media_slug)]
+ except KeyError as e:
+ raise DestinationPluginError(f'Medium "{media_slug}" is not installed.') from e
+
+
def api_safely_get_medium_object(media_slug):
try:
obj = MEDIA_CLASSES_DICT[media_slug]
diff --git a/src/argus/notificationprofile/media/base.py b/src/argus/notificationprofile/media/base.py
index 04c5dc1e1..173d4f215 100644
--- a/src/argus/notificationprofile/media/base.py
+++ b/src/argus/notificationprofile/media/base.py
@@ -1,28 +1,83 @@
from __future__ import annotations
from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
+
+from django import forms
+from django.core.exceptions import ValidationError as DjangoValidationError
+from rest_framework.exceptions import ValidationError as DRFValidationError
+
+from ..models import DestinationConfig
if TYPE_CHECKING:
from collections.abc import Iterable
from types import NoneType
- from typing import Union
+ from typing import Union, Generator
from django.contrib.auth import get_user_model
from django.db.models.query import QuerySet
from argus.incident.models import Event
- from ..models import DestinationConfig
- from ..serializers import RequestDestinationConfigSerializer
User = get_user_model()
-__all__ = ["NotificationMedium"]
+__all__ = [
+ "LabelForm",
+ "CommonDestinationConfigForm",
+ "NotificationMedium",
+]
+
+
+class LabelForm(forms.ModelForm):
+ class Meta:
+ model = DestinationConfig
+ fields = ["label"]
+
+ def __init__(self, user, **kwargs):
+ self.user = user
+ super().__init__(**kwargs)
+
+
+class CommonDestinationConfigForm(forms.ModelForm):
+ class Meta:
+ model = DestinationConfig
+ fields = ["label", "media"]
+
+ def __init__(self, user, **kwargs):
+ self.user = user
+ super().__init__(**kwargs)
class NotificationMedium(ABC):
+ """
+ Must be defined by subclasses:
+
+ Class attributes:
+
+ - MEDIA_SLUG: short string id for the medium, lowercase
+ - MEDIA_NAME: human friendly id for the medium
+ - MEDIA_SETTINGS_KEY: the field in settings that is specific for this medium
+ - MEDIA_JSON_SCHEMA: A json-schema to describe the settings field to
+ javascript, used by the API
+ - Form: a django.forms.Form that validates the contents of the
+ settings-field. MEDIA_SETTINGS_KEY must be the field-name of a required
+ field in the form.
+
+ Class methods:
+
+ - send(event, destinations): How to send the given event to the given
+ destinations of type MEDIA_SLUG.
+ """
+
+ error_messages = {
+ "readonly_medium": "Media cannot be updated, only settings.",
+ "readonly_user": "User cannot be changed, only settings.",
+ "settings_type": "Settings has to be a dictionary.",
+ "empty_settings": '"settings" cannot be empty',
+ }
+
class NotDeletableError(Exception):
"""
Custom exception class that is raised when a destination cannot be
@@ -30,35 +85,114 @@ class NotDeletableError(Exception):
"""
@classmethod
- @abstractmethod
- def validate(cls, instance: RequestDestinationConfigSerializer, dict: dict, user: User) -> dict:
+ def validate(cls, data: dict, user: User, instance: DestinationConfig = None) -> CommonDestinationConfigForm:
+ if instance:
+ if data.get("media", "") != instance.media.slug:
+ code = "readonly_media"
+ message = cls.error_messages["readonly_media"]
+ raise DRFValidationError(detail={"media": message}, code=code)
+ if instance.user != user:
+ code = "readonly_user"
+ message = cls.error_messages["readonly_user"]
+ raise DRFValidationError(detail={"user": message}, code=code)
+ form = CommonDestinationConfigForm(data, instance=instance)
+ else:
+ form = CommonDestinationConfigForm(data)
+
+ if not form.is_valid():
+ code = "invalid"
+ detail = form.errors.get_json_data()
+ raise DRFValidationError(detail=detail, code=code)
+
+ settings = data["settings"]
+ if not isinstance(settings, dict):
+ code = "settings_type"
+ message = cls.error_messages["settings_type"]
+ raise DRFValidationError(detail={"settings": message}, code=code)
+ if not settings:
+ code = "empty_settings"
+ message = cls.error_messages["empty_settings"]
+ raise DRFValidationError(detail={"settings": message}, code=code)
+ settings_form = cls.validate_settings(data["settings"], user, instance=instance)
+ if not settings_form.is_valid():
+ code = "invalid"
+ detail = settings_form.errors.get_json_data()
+ raise DRFValidationError(detail=detail, code=code)
+ form.cleaned_data["settings"] = settings_form.cleaned_data
+ form.cleaned_data["user"] = user
+ return form
+
+ @classmethod
+ def validate_settings(
+ cls,
+ data: dict,
+ user: User,
+ instance: DestinationConfig = None,
+ prefix: str = "",
+ ) -> forms.Form:
"""
- Validates the settings of destination and returns a dict with
- validated and cleaned data
+ Validates the settings of destination and returns a bound, cleaned form
"""
- pass
+ initial = {}
+ if instance:
+ initial = instance.settings
+ form = cls.Form(data=data, initial=initial, prefix=prefix)
+ if not (form.has_changed() and form.is_valid()):
+ return form
+
+ form = cls.clean(form, instance)
+
+ if cls.has_duplicate(user.destinations, form.cleaned_data):
+ form.add_error(
+ None,
+ DjangoValidationError(
+ "A %(media)s destination with this %(key) already exists",
+ code="duplicate",
+ params={
+ "media": cls.MEDIA_SLUG,
+ "key": cls.MEDIA_SETTINGS_KEY,
+ },
+ ),
+ )
+ return form
+
+ @classmethod
+ def clean(cls, form: forms.Form, instance: DestinationConfig = None) -> forms.Form:
+ """Can change the cleaned data and errors list of a valid form"""
+ return form
@classmethod
- @abstractmethod
def has_duplicate(cls, queryset: QuerySet, settings: dict) -> bool:
"""
Returns True if a destination with the given settings already exists
in the given queryset
"""
- pass
+ key = f"settings__{cls.MEDIA_SETTINGS_KEY}"
+ value = settings[cls.MEDIA_SETTINGS_KEY]
+ return queryset.filter(media_id=cls.MEDIA_SLUG, **{key: value}).exists()
- @staticmethod
- @abstractmethod
- def get_label(destination: DestinationConfig) -> str:
+ @classmethod
+ def get_label(cls, destination: DestinationConfig) -> str:
"""
- Returns a descriptive label for this destination.
+ Returns the stored label or a descriptive label for this destination otherwise
"""
- pass
+ return destination.label if destination.label else destination.settings.get(cls.MEDIA_SETTINGS_KEY)
+
+ # No querysets beyond this point!
@classmethod
- def get_relevant_addresses(cls, destinations: Iterable[DestinationConfig]) -> set[DestinationConfig]:
- """Returns a set of addresses the message should be sent to"""
- pass
+ def _get_relevant_destinations(cls, destinations: Iterable[DestinationConfig]) -> Generator[DestinationConfig]:
+ return (destination for destination in destinations if destination.media_id == cls.MEDIA_SLUG)
+
+ @classmethod
+ def get_relevant_destination_settings(cls, destinations: Iterable[DestinationConfig]) -> set[str]:
+ """Returns a set of type specific destination values the message should be sent to"""
+ destinations = [
+ destination.settings[cls.MEDIA_SETTINGS_KEY] for destination in cls._get_relevant_destinations(destinations)
+ ]
+ return set(destinations)
+
+ get_relevant_addresses = get_relevant_destination_settings
@classmethod
@abstractmethod
@@ -67,23 +201,52 @@ def send(cls, event: Event, destinations: Iterable[DestinationConfig], **kwargs)
pass
@classmethod
- def raise_if_not_deletable(cls, destination: DestinationConfig) -> NoneType:
+ def is_not_deletable(cls, destination: DestinationConfig) -> dict[str, Any]:
"""
- Raises a NotDeletableError if the given destination is not able to be deleted
- (if it is in use by any notification profiles)
+ Flag if the destination cannot be deleted
+
+ Returns an empty dict (falsey) if the destination is deletable.
+ Returns a dict (truthy) if the destination is not deletable.
+
+ Structure of dict::
+
+ {label: error_message}
"""
connected_profiles = destination.notification_profiles.all()
if connected_profiles:
profiles = ", ".join([str(profile) for profile in connected_profiles])
- raise cls.NotDeletableError(
- f"Cannot delete this destination since it is in use in the notification profile(s): {profiles}."
- )
+ error_msg = f"Destination in use by profile(s): {profiles}."
+ return {"profiles": error_msg}
+ return {}
- @staticmethod
- def update(destination: DestinationConfig, validated_data: dict) -> Union[DestinationConfig, NoneType]:
+ @classmethod
+ def raise_if_not_deletable(cls, destination: DestinationConfig) -> NoneType:
+ """
+ Raises a NotDeletableError if the given destination is not able to be deleted
+ (if it is in use by any notification profiles)
+ """
+ errors = cls.is_not_deletable(destination)
+ if errors:
+ message = " ".join(errors.values())
+ raise cls.NotDeletableError(f"Cannot delete this destination: {message}")
+
+ @classmethod
+ def _update_destination(cls, destination: DestinationConfig, validated_data: dict) -> DestinationConfig:
+ # adapted from rest_framework.serializers.ModelSerializer.update
+ # DestinationConfig MUST NOT have any m2m-relations so this is safe
+
+ for attr, value in validated_data.items():
+ setattr(destination, attr, value)
+
+ return destination
+
+ @classmethod
+ def update(cls, destination: DestinationConfig, validated_data: dict) -> Union[DestinationConfig, NoneType]:
"""
- Updates a destination in case the normal update function is not
- sufficient and returns the updated destination in that case,
- returns None otherwise
+ Updates a destination
+
+ Override in case the normal update function is not sufficient
"""
- return None
+ instance = cls._update_destination(destination, validated_data)
+ instance.save()
+ return instance
diff --git a/src/argus/notificationprofile/media/email.py b/src/argus/notificationprofile/media/email.py
index 600490a99..11b66c63c 100644
--- a/src/argus/notificationprofile/media/email.py
+++ b/src/argus/notificationprofile/media/email.py
@@ -1,36 +1,33 @@
from __future__ import annotations
import logging
-from typing import TYPE_CHECKING
+from typing import Any, TYPE_CHECKING
from django import forms
from django.conf import settings
from django.core.mail import send_mail
from django.template.loader import render_to_string
-from rest_framework.exceptions import ValidationError
from argus.incident.models import Event
+from argus.util.datetime_utils import INFINITY, LOCAL_INFINITY
+
from .base import NotificationMedium
from ..models import DestinationConfig
-from argus.util.datetime_utils import INFINITY, LOCAL_INFINITY
if TYPE_CHECKING:
from collections.abc import Iterable
- from types import NoneType
- from typing import Union
-
from django.contrib.auth import get_user_model
- from django.db.models.query import QuerySet
-
- from ..serializers import RequestDestinationConfigSerializer
User = get_user_model()
+
LOG = logging.getLogger(__name__)
__all__ = [
+ "modelinstance_to_dict",
"send_email_safely",
+ "BaseEmailNotification",
"EmailNotification",
]
@@ -59,100 +56,26 @@ def send_email_safely(function, additional_error=None, *args, **kwargs) -> int:
# TODO: Store error as incident
-class EmailNotification(NotificationMedium):
+class BaseEmailNotification(NotificationMedium):
MEDIA_SLUG = "email"
MEDIA_NAME = "Email"
+ MEDIA_SETTINGS_KEY = "email_address"
MEDIA_JSON_SCHEMA = {
"title": "Email Settings",
"description": "Settings for a DestinationConfig using email.",
"type": "object",
- "required": ["email_address"],
- "properties": {"email_address": {"type": "string", "title": "Email address"}},
+ "required": [MEDIA_SETTINGS_KEY],
+ "properties": {
+ MEDIA_SETTINGS_KEY: {
+ "type": "string",
+ "title": "Email address",
+ },
+ },
}
class Form(forms.Form):
- synced = forms.BooleanField(disabled=True, required=False, initial=False)
email_address = forms.EmailField()
- @classmethod
- def validate(cls, instance: RequestDestinationConfigSerializer, email_dict: dict, user: User) -> dict:
- """
- Validates the settings of an email destination and returns a dict
- with validated and cleaned data
- """
- form = cls.Form(email_dict["settings"])
- if not form.is_valid():
- raise ValidationError(form.errors)
- if form.cleaned_data["email_address"] == instance.context["request"].user.email:
- raise ValidationError("This email address is already registered in another destination.")
- if user.destinations.filter(
- media_id="email", settings__email_address=form.cleaned_data["email_address"]
- ).exists():
- raise ValidationError({"email_address": "Email address already exists"})
-
- return form.cleaned_data
-
- @classmethod
- def raise_if_not_deletable(cls, destination: DestinationConfig) -> NoneType:
- """
- Raises a NotDeletableError if the given email destination is not able
- to be deleted (if it was defined by an outside source or is in use by
- any notification profiles)
- """
- super().raise_if_not_deletable(destination=destination)
-
- if destination.settings["synced"]:
- raise cls.NotDeletableError(
- "Cannot delete this email destination since it was defined by an outside source."
- )
-
- @staticmethod
- def update(destination: DestinationConfig, validated_data: dict) -> Union[DestinationConfig, NoneType]:
- """
- Updates the synced email destination by copying its contents to
- a new destination and updating the given destination with the given
- validated data and returning the updated destination
-
- This way the synced destination is not lost
- """
- if destination.settings["synced"]:
- new_synced_destination = DestinationConfig(
- user=destination.user,
- media_id=destination.media_id,
- settings=destination.settings,
- )
- destination.settings = validated_data["settings"]
- DestinationConfig.objects.bulk_update([destination], fields=["settings"])
- new_synced_destination.save()
- return destination
- return None
-
- @staticmethod
- def get_label(destination: DestinationConfig) -> str:
- """
- Returns the e-mail address represented by this destination
- """
- return destination.settings.get("email_address")
-
- @classmethod
- def has_duplicate(cls, queryset: QuerySet, settings: dict) -> bool:
- """
- Returns True if an email destination with the same email address
- already exists in the given queryset
- """
- return queryset.filter(settings__email_address=settings["email_address"]).exists()
-
- @classmethod
- def get_relevant_addresses(cls, destinations: Iterable[DestinationConfig]) -> set[DestinationConfig]:
- """Returns a list of email addresses the message should be sent to"""
- email_addresses = [
- destination.settings["email_address"]
- for destination in destinations
- if destination.media_id == cls.MEDIA_SLUG
- ]
-
- return set(email_addresses)
-
@staticmethod
def create_message_context(event: Event):
"""Creates the subject, message and html message for the email"""
@@ -183,7 +106,7 @@ def send(cls, event: Event, destinations: Iterable[DestinationConfig], **_) -> b
Returns False if no email destinations were given and
True if emails were sent
"""
- email_addresses = cls.get_relevant_addresses(destinations=destinations)
+ email_addresses = cls.get_relevant_destination_settings(destinations=destinations)
if not email_addresses:
return False
num_emails = len(email_addresses)
@@ -214,3 +137,71 @@ def send(cls, event: Event, destinations: Iterable[DestinationConfig], **_) -> b
)
LOG.debug("Email: Failed to send to:", " ".join(failed))
return True
+
+
+class EmailNotification(BaseEmailNotification):
+ class Form(forms.Form):
+ synced = forms.BooleanField(disabled=True, required=False, initial=False)
+ email_address = forms.EmailField()
+
+ @classmethod
+ def is_not_deletable(cls, destination: DestinationConfig) -> dict[str, Any]:
+ """
+ Flag if the destination is undeletable due to being defined by an outside source.
+ """
+ errors = super().is_not_deletable(destination)
+ if destination.settings.get("synced", None):
+ errors["synced"] = "Email address is read-only, it is defined by an outside source."
+ return errors
+
+ @classmethod
+ def _clone_if_changing_email_address(cls, destination: DestinationConfig, validated_data: dict):
+ """Clone synced destination"""
+ if not destination.settings.get("synced", None):
+ LOG.warn('Email destination %s does not have "synced" flag in its settings', destination.id)
+ return False
+
+ new_address = validated_data["settings"][cls.MEDIA_SETTINGS_KEY]
+ old_address = destination.settings[cls.MEDIA_SETTINGS_KEY]
+ if new_address == old_address: # address wasn't changed
+ return False
+
+ settings = {
+ "synced": True,
+ cls.MEDIA_SETTINGS_KEY: old_address,
+ }
+ DestinationConfig.objects.create(
+ user=destination.user,
+ media_id=destination.media_id,
+ settings=settings,
+ label=cls.get_label(destination),
+ )
+ LOG.info("Cloning synced email-address on update: %s", old_address)
+ return True
+
+ @classmethod
+ def clean(cls, form: forms.Form, instance: DestinationConfig = None) -> forms.Form:
+ synced = False
+ if instance:
+ # CYA. The admin currently do no validation of destinations created
+ synced = instance.settings.get("synced", False)
+ form.cleaned_data["synced"] = form.cleaned_data.get("synced", synced)
+ return form
+
+ @classmethod
+ def update(cls, destination: DestinationConfig, validated_data: dict) -> DestinationConfig:
+ """
+ Preserves a synced email destination by cloning its contents to
+ a new destination and updating the given destination with the given
+ validated data and returning the updated destination
+
+ This way the synced destination is not lost
+ """
+ cls._clone_if_changing_email_address(destination, validated_data)
+
+ # We cannot use super() here since this is not an instance method
+ instance = cls._update_destination(destination, validated_data)
+ instance.settings["synced"] = False
+
+ instance.save()
+ return instance
diff --git a/src/argus/notificationprofile/media/sms_as_email.py b/src/argus/notificationprofile/media/sms_as_email.py
index 57a5c31f2..c35682e38 100644
--- a/src/argus/notificationprofile/media/sms_as_email.py
+++ b/src/argus/notificationprofile/media/sms_as_email.py
@@ -13,7 +13,6 @@
from django.conf import settings
from django.core.mail import send_mail
from phonenumber_field.formfields import PhoneNumberField
-from rest_framework.exceptions import ValidationError
from ...incident.models import Event
from .base import NotificationMedium
@@ -23,10 +22,8 @@
from collections.abc import Iterable
from django.contrib.auth import get_user_model
- from django.db.models.query import QuerySet
from ..models import DestinationConfig
- from ..serializers import RequestDestinationConfigSerializer
User = get_user_model()
@@ -36,65 +33,28 @@
class SMSNotification(NotificationMedium):
MEDIA_SLUG = "sms"
MEDIA_NAME = "SMS"
+ MEDIA_SETTINGS_KEY = "phone_number"
MEDIA_JSON_SCHEMA = {
"title": "SMS Settings",
"description": "Settings for a DestinationConfig using SMS.",
"type": "object",
- "required": ["phone_number"],
+ "required": [MEDIA_SETTINGS_KEY],
"properties": {
- "phone_number": {
+ MEDIA_SETTINGS_KEY: {
"type": "string",
"title": "Phone number",
"description": "The phone number is validated and the country code needs to be given.",
- }
+ },
},
}
- class PhoneNumberForm(forms.Form):
+ class Form(forms.Form):
phone_number = PhoneNumberField()
@classmethod
- def validate(cls, instance: RequestDestinationConfigSerializer, sms_dict: dict, user: User) -> dict:
- """
- Validates the settings of an SMS destination and returns a dict
- with validated and cleaned data
- """
- form = cls.PhoneNumberForm(sms_dict["settings"])
- if not form.is_valid():
- raise ValidationError(form.errors)
-
- form.cleaned_data["phone_number"] = form.cleaned_data["phone_number"].as_e164
-
- if user.destinations.filter(media_id="sms", settings__phone_number=form.cleaned_data["phone_number"]).exists():
- raise ValidationError({"phone_number": "Phone number already exists"})
-
- return form.cleaned_data
-
- @staticmethod
- def get_label(destination: DestinationConfig) -> str:
- """
- Returns the phone number represented by this SMS destination
- """
- return destination.settings.get("phone_number")
-
- @classmethod
- def has_duplicate(cls, queryset: QuerySet, settings: dict) -> bool:
- """
- Returns True if a sms destination with the same phone number
- already exists in the given queryset
- """
- return queryset.filter(settings__phone_number=settings["phone_number"]).exists()
-
- @classmethod
- def get_relevant_addresses(cls, destinations: Iterable[DestinationConfig]) -> set[DestinationConfig]:
- """Returns a list of phone numbers the message should be sent to"""
- phone_numbers = [
- destination.settings["phone_number"]
- for destination in destinations
- if destination.media_id == cls.MEDIA_SLUG
- ]
-
- return set(phone_numbers)
+ def clean(cls, form: Form, instance: DestinationConfig = None) -> Form:
+ form.cleaned_data[cls.MEDIA_SETTINGS_KEY] = form.cleaned_data[cls.MEDIA_SETTINGS_KEY].as_e164
+ return form
@classmethod
def send(cls, event: Event, destinations: Iterable[DestinationConfig], **_) -> bool:
@@ -108,7 +68,7 @@ def send(cls, event: Event, destinations: Iterable[DestinationConfig], **_) -> b
LOG.error("SMS_GATEWAY_ADDRESS is not set, cannot dispatch SMS notifications using this plugin")
return
- phone_numbers = cls.get_relevant_addresses(destinations=destinations)
+ phone_numbers = cls.get_relevant_destination_settings(destinations=destinations)
if not phone_numbers:
return False
diff --git a/src/argus/notificationprofile/serializers.py b/src/argus/notificationprofile/serializers.py
index af465aed1..11d0700d3 100644
--- a/src/argus/notificationprofile/serializers.py
+++ b/src/argus/notificationprofile/serializers.py
@@ -137,27 +137,42 @@ class Meta:
]
def validate(self, attrs: dict):
- if self.instance and "media" in attrs.keys() and not attrs["media"].slug == self.instance.media.slug:
- raise serializers.ValidationError("Media cannot be updated, only settings.")
- if "settings" in attrs.keys():
- if not isinstance(attrs["settings"], dict):
- raise serializers.ValidationError("Settings has to be a dictionary.")
- if self.instance:
- medium = api_safely_get_medium_object(self.instance.media.slug)
- else:
- medium = api_safely_get_medium_object(attrs["media"].slug)
- attrs["settings"] = medium.validate(self, attrs, self.context["request"].user)
+ settings = attrs.get("settings", None)
+ if not settings:
+ raise serializers.ValidationError("Settings cannot be empty")
+
+ if self.instance:
+ medium = api_safely_get_medium_object(self.instance.media.slug)
+ # A PATCH need not contain the media key
+ if attrs.get("media", self.instance.media) != self.instance.media:
+ raise serializers.ValidationError(medium.error_messages["readonly_medium"])
+ user = self.instance.user
+ if user != self.context["request"].user:
+ raise serializers.ValidationError(medium.error_messages["readonly_user"])
+ else:
+ # new, not a PATCH
+ medium = api_safely_get_medium_object(attrs["media"].slug)
+ user = self.context["request"].user
+
+ if not isinstance(attrs["settings"], dict):
+ raise serializers.ValidationError(medium.error_messages["settings_type"])
+
+ settings_form = medium.validate_settings(
+ attrs["settings"],
+ user,
+ instance=self.instance,
+ )
+ if not settings_form.is_valid():
+ raise serializers.ValidationError(settings_form.errors.get_json_data())
+
+ attrs["settings"] = settings_form.cleaned_data
return attrs
def update(self, destination: DestinationConfig, validated_data: dict):
medium = api_safely_get_medium_object(destination.media.slug)
updated_destination = medium.update(destination, validated_data)
-
- if updated_destination:
- return updated_destination
-
- return super().update(destination, validated_data)
+ return updated_destination
class DuplicateDestinationSerializer(serializers.Serializer):
diff --git a/tests/notificationprofile/destinations/test_base.py b/tests/notificationprofile/destinations/test_base.py
new file mode 100644
index 000000000..8c610e662
--- /dev/null
+++ b/tests/notificationprofile/destinations/test_base.py
@@ -0,0 +1,62 @@
+from unittest import skip
+
+from django import forms
+from django.test import TestCase, tag
+from rest_framework.exceptions import ValidationError
+
+from argus.auth.factories import PersonUserFactory
+from argus.notificationprofile.media.base import NotificationMedium
+
+
+class DummyNotification(NotificationMedium):
+ MEDIA_SETTINGS_KEY = "foo"
+ MEDIA_NAME = "Dummy"
+ MEDIA_SLUG = "dummy"
+
+ class Form(forms.Form):
+ foo = forms.IntegerField()
+
+
+@tag("unit")
+class NotificationMediumValidateSettingsTests(TestCase):
+ def setUp(self):
+ self.user = PersonUserFactory()
+
+ def test_is_not_valid_on_wrong_settings(self):
+ data = {"bar": False}
+ form = DummyNotification.validate_settings(data, self.user)
+ self.assertFalse(form.is_valid())
+ errors = form.errors.get_json_data()
+ self.assertEqual(errors["foo"][0]["message"], "This field is required.")
+
+
+@tag("unit")
+@skip("unfinished, needs instance of media type")
+class NotificationMediumValidateTests(TestCase):
+ def setUp(self):
+ self.user = PersonUserFactory()
+
+ def test_croaks_on_non_dict_settings(self):
+ data = {"settings": None}
+ with self.assertRaises(ValidationError) as e:
+ DummyNotification.validate(data, self.user)
+ foo_error = e.exception
+ self.assertEqual(foo_error.message, DummyNotification.error_messages["settings_type"])
+
+ def test_croaks_on_empty_settings(self):
+ data = {"settings": {}}
+ with self.assertRaises(ValidationError) as e:
+ DummyNotification.validate(data, self.user)
+ foo_error = e.exception
+ self.assertEqual(foo_error.message, DummyNotification.error_messages["empty_settings"])
+
+ def test_croaks_if_plugin_not_loaded(self):
+ data = {
+ "media": DummyNotification.MEDIA_SLUG,
+ "settings": {"foo": 1},
+ }
+ with self.assertRaises(ValidationError) as e:
+ DummyNotification.validate(data, self.user)
+ self.assertIn("media", e.exception.error_dict)
+ media_error = e.exception.error_dict["media"][0]
+ self.assertEqual(media_error.message, "Select a valid choice. That choice is not one of the available choices.")