Coverage for website/pushnotifications/api/v2/views.py: 57.81%

1from oauth2_provider.contrib.rest_framework import IsAuthenticatedOrTokenHasScope 

2from rest_framework.filters import OrderingFilter 

3from rest_framework.generics import ( 

4 CreateAPIView, 

5 ListAPIView, 

6 RetrieveAPIView, 

7 UpdateAPIView, 

8) 

9 

10from pushnotifications.api.v2.filters import CategoryFilter 

11from pushnotifications.api.v2.permissions import IsAuthenticatedOwnerOrReadOnly 

12from pushnotifications.api.v2.serializers import ( 

13 CategorySerializer, 

14 DeviceSerializer, 

15 MessageSerializer, 

16) 

17from pushnotifications.models import Category, Device, Message 

18from thaliawebsite.api.v2.permissions import IsAuthenticatedOrTokenHasScopeForMethod 

19 

20 

21class DeviceListView(ListAPIView, CreateAPIView): 

22 """Returns an overview of all devices that are owner by the user.""" 

23 

24 permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod] 

25 serializer_class = DeviceSerializer 

26 queryset = Device.objects.all() 

27 required_scopes_per_method = { 

28 "GET": ["pushnotifications:read"], 

29 "POST": ["pushnotifications:write"], 

30 } 

31 

32 def get_queryset(self): 

33 if self.request.user: 

34 return Device.objects.filter(user=self.request.user) 

35 return super().get_queryset() 

36 

37 def perform_create(self, serializer): 

38 try: 

39 serializer.instance = Device.objects.get( 

40 user=self.request.user, 

41 registration_id=serializer.validated_data["registration_id"], 

42 ) 

43 except Device.DoesNotExist: 

44 pass 

45 

46 data = serializer.validated_data 

47 categories = [c.pk for c in Category.objects.all()] 

48 if "receive_category" in data and len(data["receive_category"]) > 0: 

49 categories = data["receive_category"] + ["general"] 

50 

51 serializer.save(user=self.request.user, receive_category=categories) 

52 

53 

54class DeviceDetailView(RetrieveAPIView, UpdateAPIView): 

55 """Returns details of a device.""" 

56 

57 permission_classes = [ 

58 IsAuthenticatedOrTokenHasScope, 

59 IsAuthenticatedOwnerOrReadOnly, 

60 ] 

61 serializer_class = DeviceSerializer 

62 required_scopes = ["pushnotifications:read", "pushnotifications:write"] 

63 queryset = Device.objects.all() 

64 

65 def perform_update(self, serializer): 

66 serializer.save(user=self.request.user) 

67 

68 

69class CategoryListView(ListAPIView): 

70 """Returns an overview of all available categories for push notifications.""" 

71 

72 serializer_class = CategorySerializer 

73 queryset = Category.objects.all() 

74 required_scopes = ["pushnotifications:read"] 

75 

76 

77class MessageListView(ListAPIView): 

78 """Returns a list of message sent to the user.""" 

79 

80 serializer_class = MessageSerializer 

81 required_scopes = ["pushnotifications:read"] 

82 permission_classes = [ 

83 IsAuthenticatedOrTokenHasScope, 

84 ] 

85 filter_backends = (OrderingFilter, CategoryFilter) 

86 ordering_fields = ("sent",) 

87 

88 def get_queryset(self): 

89 if self.request.user: 

90 return Message.all_objects.filter(users=self.request.user) 

91 return Message.all_objects.all() 

92 

93 

94class MessageDetailView(RetrieveAPIView): 

95 """Returns a message.""" 

96 

97 serializer_class = MessageSerializer 

98 required_scopes = ["pushnotifications:read"] 

99 permission_classes = [ 

100 IsAuthenticatedOrTokenHasScope, 

101 ] 

102 

103 def get_queryset(self): 

104 if self.request.user: 

105 return Message.all_objects.filter(users=self.request.user) 

106 return Message.all_objects.all()