Coverage for src/accounts/views/user_credential.py: 59%

1""" 

2This module defines API views for user registration, login, logout, password change, and account deletion. 

3 

4Key features: 

5- User registration with email and password 

6- User login with email or username 

7- JWT token issuance and blacklisting 

8- Password change for authenticated users 

9- Account deletion for authenticated users 

10""" 

11 

12from django.contrib.auth import get_user_model, authenticate 

13 

14from rest_framework.generics import CreateAPIView, DestroyAPIView 

15from rest_framework.permissions import AllowAny 

16from rest_framework.views import APIView 

17from rest_framework.response import Response 

18from rest_framework import status 

19from rest_framework_simplejwt.tokens import RefreshToken 

20from rest_framework_simplejwt.exceptions import TokenError 

21 

22from src.accounts.serializers.user_credential import ( 

23 PasswordResetConfirmSerializer, 

24 PasswordResetRequestSerializer, 

25 UserRegisterSerializer, 

26 UserLoginRequestSerializer, 

27 PasswordChangeSerializer, 

28) 

29from src.accounts.constants import UserErrorMessages, UserSuccessMessages 

30 

31UserModel = get_user_model() 

32 

33 

34class UserRegisterView(CreateAPIView): 

35 """ 

36 Uses a signal to create related UserProfile and UserPhoto models. 

37 """ 

38 

39 queryset = UserModel.objects.all() 

40 serializer_class = UserRegisterSerializer 

41 permission_classes = [AllowAny] 

42 

43 def create(self, request, *args, **kwargs): 

44 # Validate incoming registration data 

45 serializer = self.get_serializer(data=request.data) 

46 serializer.is_valid(raise_exception=True) 

47 

48 # Save the new user instance 

49 user = serializer.save() 

50 

51 # Issue JWT refresh and access tokens for the new user 

52 refresh = RefreshToken.for_user(user) 

53 

54 # Return tokens and user info to the frontend 

55 

56 return Response( 

57 { 

58 'refresh': str(refresh), 

59 'access': str(refresh.access_token), 

60 'email': user.email, 

61 'username': user.username, 

62 'id': user.pk, 

63 }, 

64 status=status.HTTP_201_CREATED, 

65 ) 

66 

67 

68class UserLoginView(APIView): 

69 """ 

70 - Accepts email or username and password. 

71 - Authenticates the user and issues JWT tokens. 

72 - Allows any user to attempt login. 

73 """ 

74 

75 permission_classes = [AllowAny] 

76 

77 def post(self, request, *args, **kwargs): 

78 # Validate login credentials 

79 serializer = UserLoginRequestSerializer(data=request.data) 

80 if not serializer.is_valid(): 

81 return Response( 

82 serializer.errors, 

83 status=status.HTTP_400_BAD_REQUEST, 

84 ) 

85 

86 # Extract validated credentials 

87 email_or_username = serializer.validated_data['email_or_username'] 

88 password = serializer.validated_data['password'] 

89 

90 # Authenticate user (returns user instance or None) 

91 user = authenticate( 

92 username=email_or_username, 

93 password=password, 

94 ) 

95 

96 if user is None: 

97 # Invalid credentials 

98 return Response( 

99 { 

100 'error': UserErrorMessages.INCORRECT_CREDENTIALS, 

101 }, 

102 status=status.HTTP_401_UNAUTHORIZED, 

103 ) 

104 

105 # Issue JWT tokens 

106 refresh = RefreshToken.for_user(user) 

107 

108 # Collect user permissions for frontend role-based functionality 

109 permissions = [] 

110 if user.has_perm('products.approve_review'): 

111 permissions.append('products.approve_review') 

112 

113 # Return tokens, user info, and permissions 

114 

115 return Response( 

116 { 

117 'refresh': str(refresh), 

118 'access': str(refresh.access_token), 

119 'message': 'Login successful', 

120 'username': user.username, 

121 'email': user.email, 

122 'id': user.pk, 

123 'permissions': permissions, 

124 }, 

125 status=status.HTTP_200_OK, 

126 ) 

127 

128 

129class UserLogoutView(APIView): 

130 """ 

131 - Accepts a refresh token and blacklists it (invalidates the session). 

132 - Returns a success or error message. 

133 """ 

134 

135 permission_classes = [AllowAny] 

136 

137 def post(self, request, *args, **kwargs): 

138 try: 

139 # Get refresh token from request data 

140 refresh_token = request.data.get('refresh') 

141 token = RefreshToken(refresh_token) 

142 

143 # Blacklist the token (requires blacklist app enabled) 

144 token.blacklist() 

145 

146 return Response( 

147 { 

148 'message': UserSuccessMessages.LOGOUT_SUCCESS, 

149 }, 

150 status=status.HTTP_200_OK, 

151 ) 

152 

153 except TokenError: 

154 # Token is invalid or already expired/blacklisted 

155 return Response( 

156 { 

157 'error': UserErrorMessages.INVALID_TOKEN, 

158 }, 

159 status=status.HTTP_400_BAD_REQUEST, 

160 ) 

161 

162 

163class PasswordChangeView(APIView): 

164 """ 

165 - Only accessible to authenticated users. 

166 - Validates current and new passwords. 

167 - Saves the new password if validation passes. 

168 """ 

169 

170 def patch(self, request): 

171 # Validate current and new passwords 

172 serializer = PasswordChangeSerializer( 

173 data=request.data, 

174 context={ 

175 'request': request, 

176 }, 

177 ) 

178 

179 if serializer.is_valid(): 

180 serializer.save() 

181 

182 return Response( 

183 { 

184 'message': UserSuccessMessages.PASSWORD_CHANGED, 

185 }, 

186 status=status.HTTP_200_OK, 

187 ) 

188 

189 return Response( 

190 serializer.errors, 

191 status=status.HTTP_400_BAD_REQUEST, 

192 ) 

193 

194 

195class UserDeleteView(DestroyAPIView): 

196 """ 

197 - Only accessible to authenticated users. 

198 - Deletes the user instance associated with the current request. 

199 """ 

200 

201 def get_object(self): 

202 # Return the current user instance for deletion 

203 

204 return self.request.user 

205 

206 

207class PasswordResetRequestView(APIView): 

208 """Send password reset email""" 

209 

210 permission_classes = [AllowAny] 

211 

212 def post(self, request): 

213 serializer = PasswordResetRequestSerializer(data=request.data) 

214 

215 if serializer.is_valid(): 

216 serializer.save() 

217 

218 return Response( 

219 { 

220 'message': UserSuccessMessages.RESET_LINK_SENT, 

221 }, 

222 status=status.HTTP_200_OK, 

223 ) 

224 

225 return Response( 

226 serializer.errors, 

227 status=status.HTTP_400_BAD_REQUEST, 

228 ) 

229 

230 

231class PasswordResetConfirmView(APIView): 

232 """Reset password with token""" 

233 

234 permission_classes = [AllowAny] 

235 

236 def post(self, request): 

237 serializer = PasswordResetConfirmSerializer(data=request.data) 

238 if serializer.is_valid(): 

239 serializer.save() 

240 

241 return Response( 

242 { 

243 'message': UserSuccessMessages.PASSWORD_RESET, 

244 }, 

245 status=status.HTTP_200_OK, 

246 ) 

247 

248 return Response( 

249 serializer.errors, 

250 status=status.HTTP_400_BAD_REQUEST, 

251 )