Coverage for users/api/views.py: 90%

1import logging 

2 

3from django.db.models import Count 

4from drf_spectacular.types import OpenApiTypes 

5from drf_spectacular.utils import extend_schema 

6from rest_framework.permissions import IsAuthenticated 

7from rest_framework.response import Response 

8from rest_framework.routers import APIRootView 

9from rest_framework.status import HTTP_201_CREATED 

10from rest_framework.views import APIView 

11from rest_framework.viewsets import ViewSet 

12 

13from netbox.api.viewsets import NetBoxModelViewSet 

14from users import filtersets 

15from users.models import Group, ObjectPermission, Owner, OwnerGroup, Token, User, UserConfig 

16from utilities.data import deepmerge 

17from utilities.querysets import RestrictedQuerySet 

18 

19from . import serializers 

20 

21 

22class UsersRootView(APIRootView): 

23 """ 

24 Users API root view 

25 """ 

26 def get_view_name(self): 

27 return 'Users' 

28 

29 

30# 

31# Users and groups 

32# 

33 

34class UserViewSet(NetBoxModelViewSet): 

35 queryset = RestrictedQuerySet(model=User).order_by('username') 

36 serializer_class = serializers.UserSerializer 

37 filterset_class = filtersets.UserFilterSet 

38 

39 

40class GroupViewSet(NetBoxModelViewSet): 

41 queryset = Group.objects.annotate(user_count=Count('user')) 

42 serializer_class = serializers.GroupSerializer 

43 filterset_class = filtersets.GroupFilterSet 

44 

45 

46# 

47# REST API tokens 

48# 

49 

50class TokenViewSet(NetBoxModelViewSet): 

51 queryset = Token.objects.all() 

52 serializer_class = serializers.TokenSerializer 

53 filterset_class = filtersets.TokenFilterSet 

54 # A created v2 Token's plaintext must not be retained in a job record. 

55 background_enabled = False 

56 

57 

58class TokenProvisionView(APIView): 

59 """ 

60 Non-authenticated REST API endpoint via which a user may create a Token. 

61 """ 

62 permission_classes = [] 

63 

64 @extend_schema( 

65 request=serializers.TokenProvisionSerializer, 

66 responses={ 

67 201: serializers.TokenProvisionSerializer, 

68 401: OpenApiTypes.OBJECT, 

69 } 

70 ) 

71 def post(self, request): 

72 serializer = serializers.TokenProvisionSerializer(data=request.data, context={'request': request}) 

73 serializer.is_valid(raise_exception=True) 

74 self.perform_create(serializer) 

75 return Response(serializer.data, status=HTTP_201_CREATED) 

76 

77 def perform_create(self, serializer): 

78 model = serializer.Meta.model 

79 logger = logging.getLogger('netbox.api.views.TokenProvisionView') 

80 logger.info(f"Creating new {model._meta.verbose_name}") 

81 serializer.save() 

82 

83 

84# 

85# ObjectPermissions 

86# 

87 

88class ObjectPermissionViewSet(NetBoxModelViewSet): 

89 queryset = ObjectPermission.objects.all() 

90 serializer_class = serializers.ObjectPermissionSerializer 

91 filterset_class = filtersets.ObjectPermissionFilterSet 

92 

93 

94# 

95# Owners 

96# 

97 

98class OwnerGroupViewSet(NetBoxModelViewSet): 

99 queryset = OwnerGroup.objects.all() 

100 serializer_class = serializers.OwnerGroupSerializer 

101 filterset_class = filtersets.OwnerGroupFilterSet 

102 

103 

104class OwnerViewSet(NetBoxModelViewSet): 

105 queryset = Owner.objects.all() 

106 serializer_class = serializers.OwnerSerializer 

107 filterset_class = filtersets.OwnerFilterSet 

108 

109 

110# 

111# User preferences 

112# 

113 

114class UserConfigViewSet(ViewSet): 

115 """ 

116 An API endpoint via which a user can update his or her own UserConfig data (but no one else's). 

117 """ 

118 permission_classes = [IsAuthenticated] 

119 

120 def get_queryset(self): 

121 return UserConfig.objects.filter(user=self.request.user) 

122 

123 @extend_schema(responses={200: OpenApiTypes.OBJECT}) 

124 def list(self, request): 

125 """ 

126 Return the UserConfig for the currently authenticated User. 

127 """ 

128 userconfig = self.get_queryset().first() 

129 

130 return Response(userconfig.data) 

131 

132 @extend_schema(methods=["patch"], responses={201: OpenApiTypes.OBJECT}) 

133 def patch(self, request): 

134 """ 

135 Update the UserConfig for the currently authenticated User. 

136 """ 

137 # TODO: How can we validate this data? 

138 userconfig = self.get_queryset().first() 

139 userconfig.data = deepmerge(userconfig.data, request.data) 

140 userconfig.save() 

141 

142 return Response(userconfig.data)