neptune/core/forms.py

from django import forms
from django.contrib.auth.forms import UserCreationForm
from django.core.exceptions import FieldDoesNotExist
from django.forms import ModelForm

from core.db.storage import db
from core.lib.parsing import QueryError
from core.lib.rules import NotificationRuleData, RuleParseError

from .models import NotificationRule, NotificationSettings, User

# flake8: noqa: E501


class RestrictedFormMixin:
    """
    This mixin is used to restrict the queryset of a form to the current user.
    The request object is passed from the view.
    Fieldargs is used to pass additional arguments to the queryset filter.
    """

    fieldargs = {}

    def __init__(self, *args, **kwargs):
        # self.fieldargs = {}
        self.request = kwargs.pop("request")
        super().__init__(*args, **kwargs)
        for field in self.fields:
            # Check it's not something like a CharField which has no queryset
            if not hasattr(self.fields[field], "queryset"):
                continue

            model = self.fields[field].queryset.model
            # Check if the model has a user field
            try:
                model._meta.get_field("user")
                # Add the user to the queryset filters
                self.fields[field].queryset = model.objects.filter(
                    user=self.request.user, **self.fieldargs.get(field, {})
                )
            except FieldDoesNotExist:
                pass


class NewUserForm(UserCreationForm):
    email = forms.EmailField(required=True)

    class Meta:
        model = User
        fields = (
            "username",
            "email",
            "first_name",
            "last_name",
            "password1",
            "password2",
        )

    def save(self, commit=True):
        user = super(NewUserForm, self).save(commit=False)
        user.email = self.cleaned_data["email"]
        if commit:
            user.save()
        return user


class CustomUserCreationForm(UserCreationForm):
    class Meta:
        model = User
        fields = "__all__"


class NotificationSettingsForm(RestrictedFormMixin, ModelForm):
    def __init__(self, *args, **kwargs):
        super(NotificationSettingsForm, self).__init__(*args, **kwargs)
        self.fields["url"].label = "URL"

    class Meta:
        model = NotificationSettings
        fields = (
            "topic",
            "url",
            "service",
        )
        help_texts = {
            "topic": "The topic to send notifications to.",
            "url": "Custom NTFY server/webhook destination. Leave blank to use the default server for NTFY. For webhooks this field is required.",
            "service": "The service to use for notifications.",
        }

    def clean(self):
        cleaned_data = super(NotificationSettingsForm, self).clean()
        if "service" in cleaned_data:
            if cleaned_data["service"] == "webhook":
                if not cleaned_data.get("url"):
                    self.add_error(
                        "url",
                        "You must set a URL for webhooks.",
                    )


class NotificationRuleForm(RestrictedFormMixin, ModelForm):
    def __init__(self, *args, **kwargs):
        super(NotificationRuleForm, self).__init__(*args, **kwargs)
        self.fields["url"].label = "URL"

    class Meta:
        model = NotificationRule
        fields = (
            "name",
            "data",
            "interval",
            "window",
            "amount",
            "priority",
            "topic",
            "url",
            "service",
            "policy",
            "enabled",
        )
        help_texts = {
            "name": "The name of the rule.",
            "priority": "The notification priority of the rule.",
            "url": "Custom NTFY server/webhook destination. Leave blank to use the default server for NTFY. For webhooks this field is required.",
            "service": "The service to use for notifications",
            "topic": "The topic to send notifications to. Leave blank for default.",
            "enabled": "Whether the rule is enabled.",
            "data": "The notification rule definition.",
            "interval": "How often to run the search. On demand evaluates messages as they are received, without running a scheduled search. The remaining options schedule a search of the database with the window below.",
            "window": "Time window to search: 1d, 1h, 1m, 1s, etc.",
            "amount": "Amount of matches to be returned for scheduled queries. Cannot be used with on-demand queries.",
            "policy": "When to trigger this policy.",
        }

    def clean(self):
        cleaned_data = super(NotificationRuleForm, self).clean()

        # TODO: should this be in rules.py?
        if "service" in cleaned_data:
            if cleaned_data["service"] == "webhook":
                if not cleaned_data.get("url"):
                    self.add_error(
                        "url",
                        "You must set a URL for webhooks.",
                    )
        try:
            # Passing db to avoid circular import
            parsed_data = NotificationRuleData(self.request.user, cleaned_data, db=db)
            if cleaned_data["enabled"]:
                parsed_data.test_schedule()
        except RuleParseError as e:
            self.add_error(e.field, f"Parsing error: {e}")
            return
        except QueryError as e:
            self.add_error("data", f"Query error: {e}")
            return

        # Write back the validated data
        # We need this to populate the index and source variable if
        # they are not set
        to_store = str(parsed_data)
        cleaned_data["data"] = to_store

        return cleaned_data