GlitchTip/users/views.py

from allauth.account.models import EmailAddress
from dj_rest_auth.registration.views import (
    SocialAccountDisconnectView as BaseSocialAccountDisconnectView,
)
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.http import Http404
from django.shortcuts import get_object_or_404
from rest_framework import exceptions, mixins, status, viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

from organizations_ext.models import Organization
from projects.models import UserProjectAlert
from users.utils import is_user_registration_open

from .models import User
from .serializers import (
    ConfirmEmailAddressSerializer,
    CurrentUserSerializer,
    EmailAddressSerializer,
    UserNotificationsSerializer,
    UserSerializer,
)


class UserViewSet(viewsets.ModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer

    def get_queryset(self):
        queryset = super().get_queryset()
        organization_slug = self.kwargs.get("organization_slug")
        if organization_slug:
            queryset = queryset.filter(
                organizations_ext_organization__slug=organization_slug,
                organizations_ext_organization__users=self.request.user,
            )
        else:
            queryset = queryset.filter(id=self.request.user.id)
        return queryset

    def get_object(self):
        if self.kwargs.get("pk") == "me":
            return self.request.user
        return super().get_object()

    def get_serializer_class(self):
        if self.kwargs.get("pk") == "me":
            return CurrentUserSerializer
        return super().get_serializer_class()

    def perform_create(self, serializer):
        organization_slug = self.kwargs.get("organization_slug")
        try:
            Organization.objects.get(slug=organization_slug)
        except Organization.DoesNotExist as err:
            raise exceptions.ValidationError("Organization does not exist") from err
        # TODO deal with organization and users who aren't set up yet
        if not is_user_registration_open() and not self.request.user.is_superuser:
            raise exceptions.PermissionDenied("Registration is not open")
        user = serializer.save()
        return user

    def destroy(self, request, *args, **kwargs):
        instance = self.get_object()
        if instance.organizations_ext_organizationuser.filter(
            organizationowner__isnull=False
        ):
            return Response(
                data={
                    "message": "User is organization owner. Delete organization or transfer ownership first."
                },
                status=status.HTTP_400_BAD_REQUEST,
            )

        self.perform_destroy(instance)
        return Response(status=status.HTTP_204_NO_CONTENT)

    @action(detail=True, methods=["get", "post", "put"])
    def notifications(self, request, pk=None):
        user = self.get_object()

        if request.method == "GET":
            serializer = UserNotificationsSerializer(user)
            return Response(serializer.data)

        serializer = UserNotificationsSerializer(user, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        else:
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    @action(
        detail=True, methods=["get", "post", "put"], url_path="notifications/alerts"
    )
    def alerts(self, request, pk=None):
        """
        Returns dictionary of project_id: status. Now project_id status means it's "default"

        To update, submit `{project_id: status}` where status is -1 (default), 0, or 1
        """
        user = self.get_object()
        alerts = user.userprojectalert_set.all()

        if request.method == "GET":
            data = {}
            for alert in alerts:
                data[alert.project_id] = alert.status
            return Response(data)

        data = request.data
        try:
            items = [x for x in data.items()]
        except AttributeError as err:
            raise exceptions.ValidationError(
                "Invalid alert format, expected dictionary"
            ) from err
        if len(data) != 1:
            raise exceptions.ValidationError("Invalid alert format, expected one value")
        project_id, alert_status = items[0]
        if alert_status not in [1, 0, -1]:
            raise exceptions.ValidationError("Invalid status, must be -1, 0, or 1")
        alert = alerts.filter(project_id=project_id).first()
        if alert and alert_status == -1:
            alert.delete()
        else:
            UserProjectAlert.objects.update_or_create(
                user=user, project_id=project_id, defaults={"status": alert_status}
            )
        return Response(status=204)


class EmailAddressViewSet(
    mixins.CreateModelMixin,
    mixins.ListModelMixin,
    viewsets.GenericViewSet,
):
    queryset = EmailAddress.objects.all()
    serializer_class = EmailAddressSerializer
    pagination_class = None

    def get_user(self, user_pk):
        if user_pk == "me":
            return self.request.user
        raise exceptions.ValidationError(
            "Can only change primary email address on own account"
        )

    def get_queryset(self):
        user = self.get_user(self.kwargs.get("user_pk"))
        queryset = super().get_queryset().filter(user=user)
        return queryset

    def put(self, request, user_pk, format=None):
        """
        Set a new primary email (must be verified) this will also set the email used when a user logs in.
        """
        user = self.get_user(user_pk)
        try:
            email_address = user.emailaddress_set.get(
                email=request.data.get("email"), verified=True
            )
            email_address.set_as_primary()
        except ObjectDoesNotExist as err:
            raise Http404 from err
        serializer = self.serializer_class(
            instance=email_address, data=request.data, context={"request": request}
        )
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

    def delete(self, request, user_pk, format=None):
        user = self.get_user(user_pk)
        try:
            email_address = user.emailaddress_set.get(
                email=request.data.get("email"), primary=False
            )
        except ObjectDoesNotExist as err:
            raise Http404 from err
        email_address.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

    @action(detail=False, methods=["post"])
    def confirm(self, request, user_pk):
        serializer = ConfirmEmailAddressSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        email_address = get_object_or_404(
            self.get_queryset(), email=serializer.validated_data.get("email")
        )
        email_address.send_confirmation(request)
        return Response(status=204)


class SocialAccountDisconnectView(BaseSocialAccountDisconnectView):
    def post(self, request, *args, **kwargs):
        try:
            return super().post(request, *args, **kwargs)
        except ValidationError as e:
            raise exceptions.ValidationError(e.message)