Coverage for src/routes/auth.py: 46%

1from fastapi.responses import JSONResponse 

2from redis.asyncio.client import Redis 

3 

4from fastapi import APIRouter, Form, HTTPException, Depends, Security, status, BackgroundTasks, Request 

5 

6from fastapi.templating import Jinja2Templates 

7 

8from fastapi.security import HTTPAuthorizationCredentials, OAuth2PasswordRequestForm, HTTPBearer 

9from sqlalchemy.ext.asyncio import AsyncSession 

10 

11from src.services.email import send_email, send_password_reset_email 

12from src.entity.models import User 

13from src.database.db import get_db, get_redis_client 

14from src.repository import users as repositories_users 

15from src.schemas.user import RequestEmail, UserSchema, TokenSchema, UserResponseSchema 

16from src.services.auth import auth_serviсe 

17from src.conf.config import conf 

18 

19 

20router = APIRouter(prefix="/auth", tags=["auth"]) 

21 

22get_refresh_token = HTTPBearer() 

23 

24# Ініціалізація Jinja2Templates з вказівкою на папку з шаблонами 

25templates = Jinja2Templates(directory=conf.TEMPLATE_FOLDER) 

26 

27@router.post("/signup", response_model=UserResponseSchema, status_code=status.HTTP_201_CREATED) 

28async def signup(body: UserSchema, bt: BackgroundTasks, request: Request, db: AsyncSession = Depends(get_db)): 

29 """ 

30 Sign up a new user. 

31 

32 Parameters: 

33 - body (UserSchema): The user data to be created. 

34 - bt (BackgroundTasks): A task manager for asynchronous tasks. 

35 - request (Request): The HTTP request object. 

36 - db (AsyncSession): The database session. 

37 

38 Returns: 

39 - UserResponseSchema: The newly created user object. 

40 

41 Raises: 

42 - HTTPException: If the user already exists. 

43 """ 

44 exist_user = await repositories_users.get_user_by_email(body.email, db) 

45 if exist_user: 

46 raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="Account already exists") 

47 body.password = auth_serviсe.get_password_hash(body.password) 

48 new_user = await repositories_users.create_user(body, db) 

49 bt.add_task(send_email, new_user.email, new_user.username, str(request.base_url)) 

50 return new_user 

51 

52 

53@router.post("/login", response_model=TokenSchema) 

54async def login(body: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)): 

55 """ 

56 Logs in a user and returns a JWT token. 

57 

58 Parameters: 

59 - body (OAuth2PasswordRequestForm): The user credentials. 

60 - db (AsyncSession): The database session. 

61 

62 Returns: 

63 - TokenSchema: A dictionary containing the access token, refresh token, and token type. 

64 

65 Raises: 

66 - HTTPException: If the user does not exist, if the user is not confirmed, or if the password is incorrect. 

67 """ 

68 user = await repositories_users.get_user_by_email(body.username, db) 

69 if user is None: 

70 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") # Invalid email 

71 if not user.confirmed: 

72 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") # Not confirmed 

73 if not auth_serviсe.verify_password(body.password, user.password): 

74 raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid credentials") # Invalid password 

75 # Generate JWT 

76 access_token = await auth_serviсe.create_access_token(data={"sub": user.email}) 

77 refresh_token = await auth_serviсe.create_refresh_token(data={"sub": user.email}) 

78 await repositories_users.update_token(user, refresh_token, db) 

79 return {"access_token": access_token, "refresh_token": refresh_token, "token_type": "bearer"} 

80 

81 

82@router.get("/refresh_token", response_model=TokenSchema) 

83async def refresh_token(credentials: HTTPAuthorizationCredentials = Security(get_refresh_token), db: AsyncSession = Depends(get_db)): 

84 """ 

85 Refresh the user's access token. 

86 

87 Parameters: 

88 - credentials (HTTPAuthorizationCredentials): The HTTP Authorization header containing the refresh token. 

89 - db (AsyncSession): The database session. 

90 

91 Returns: 

92 - TokenSchema: A dictionary containing the new access token, refresh token, and token type. 

93 

94 Raises: 

95 - HTTPException: If the refresh token is invalid or expired. 

96 """ 

97 token = credentials.credentials 

98 email = await auth_serviсe.decode_refresh_token(token) 

99 user = await repositories_users.get_user_by_email(email, db) 

100 if user.refresh_token != token: 

101 await repositories_users.update_token(user=user, refresh_token=None, db=db) 

102 raise HTTPException( 

103 status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid refresh token" 

104 ) 

105 

106 access_token = await auth_serviсe.create_access_token(data={"sub": email}) 

107 refresh_token = await auth_serviсe.create_refresh_token(data={"sub": email}) 

108 await repositories_users.update_token(user, refresh_token, db) 

109 return { 

110 "access_token": access_token, 

111 "refresh_token": refresh_token, 

112 "token_type": "bearer", 

113 } 

114 

115 

116@router.get("/confirmed_email/{token}") 

117async def confirmed_email(token: str, db: AsyncSession = Depends(get_db)): 

118 """ 

119 Confirm the user's email address. 

120 

121 Parameters: 

122 - token (str): The token sent to the user's email address for confirmation. 

123 - db (AsyncSession): The database session. 

124 

125 Returns: 

126 - dict: A dictionary containing a message indicating whether the email address has been confirmed. 

127 

128 Raises: 

129 - HTTPException: If the token is invalid or expired. 

130 """ 

131 email = await auth_serviсe.get_email_from_token(token) 

132 user = await repositories_users.get_user_by_email(email, db) 

133 if user is None: 

134 raise HTTPException( 

135 status_code=status.HTTP_400_BAD_REQUEST, detail="Verification error" 

136 ) 

137 if user.confirmed: 

138 return {"message": "Your email is already confirmed"} 

139 await repositories_users.confirmed_email(email, db) 

140 return {"message": "Email confirmed"} 

141 

142 

143@router.post("/request_email") 

144async def request_email( 

145 body: RequestEmail, 

146 background_tasks: BackgroundTasks, 

147 request: Request, 

148 db: AsyncSession = Depends(get_db), 

149): 

150 """ 

151 Request a confirmation email for the user's email address. 

152 

153 Parameters: 

154 - body (RequestEmail): A dictionary containing the user's email address. 

155 - background_tasks (BackgroundTasks): A task manager for asynchronous tasks. 

156 - request (Request): The HTTP request object. 

157 - db (AsyncSession): The database session. 

158 

159 Returns: 

160 - dict: A dictionary containing a message indicating whether the email address has been confirmed. 

161 

162 Raises: 

163 - HTTPException: If the user already has a confirmed email address. 

164 """ 

165 user = await repositories_users.get_user_by_email(body.email, db) 

166 if user.confirmed: 

167 return {"message": "Your email is already confirmed"} 

168 if user: 

169 background_tasks.add_task(send_email, user.email, user.username, str(request.base_url)) 

170 return {"message": "Check your email for confirmation."} 

171 

172 

173@router.post("/request_password_reset") 

174async def request_password_reset( 

175 body: RequestEmail, 

176 background_tasks: BackgroundTasks, 

177 request: Request, 

178 db: AsyncSession = Depends(get_db), 

179): 

180 """ 

181 Request a password reset email for the user's email address. 

182 

183 Parameters: 

184 - body (RequestEmail): A dictionary containing the user's email address. 

185 - background_tasks (BackgroundTasks): A task manager for asynchronous tasks. 

186 - request (Request): The HTTP request object. 

187 - db (AsyncSession): The database session. 

188 

189 Returns: 

190 - dict: A dictionary containing a message indicating whether the email address has been confirmed. 

191 

192 Raises: 

193 - HTTPException: If the user already has a confirmed email address. 

194 """ 

195 user = await repositories_users.get_user_by_email(body.email, db) 

196 if user: 

197 background_tasks.add_task( 

198 send_password_reset_email, user.email, user.username, str(request.base_url) 

199 ) 

200 return {"message": "Check your email to reset password."} 

201 else: 

202 return {"message": "User not found."} 

203 

204 

205@router.get("/reset_password/{token}") 

206async def confirmed_reset_password_email( 

207 token: str, request: Request, db: AsyncSession = Depends(get_db) 

208): 

209 """ 

210 Retrieves the reset password form for the user with the given token. 

211 

212 Parameters: 

213 - token (str): The token sent to the user's email address for resetting the password. 

214 - request (Request): The HTTP request object. 

215 - db (AsyncSession): The database session. 

216 

217 Returns: 

218 - TemplateResponse: A Flask template response containing the reset password form. 

219 

220 Raises: 

221 - HTTPException: If the token is invalid or expired. 

222 """ 

223 email = await auth_serviсe.get_email_from_token(token) 

224 user = await repositories_users.get_user_by_email(email, db) 

225 if user is None: 

226 raise HTTPException( 

227 status_code=status.HTTP_400_BAD_REQUEST, detail="Verification error" 

228 ) 

229 return templates.TemplateResponse( 

230 "reset_password_form.html", 

231 {"request": request, "token": token}, 

232 ) 

233 

234 

235@router.post("/reset-password/") 

236async def reset_password( 

237 token: str = Form(...), 

238 new_password: str = Form(...), 

239 db: AsyncSession = Depends(get_db), 

240): 

241 """ 

242 Reset the user's password using a token sent to their email address. 

243 

244 Parameters: 

245 - token (str): The token sent to the user's email address for resetting the password. 

246 - new_password (str): The new password to be set for the user. 

247 - db (AsyncSession): The database session. 

248 

249 Returns: 

250 - JSONResponse: A JSON response containing a message indicating the success of the password reset. 

251 

252 Raises: 

253 - HTTPException: If the token is invalid or expired, or if the user is not found. 

254 """ 

255 email = await auth_serviсe.get_email_from_token(token) 

256 user = await repositories_users.get_user_by_email(email, db) 

257 if not user: 

258 raise HTTPException(status_code=404, detail="User not found") 

259 

260 new_hashed_password = auth_serviсe.get_password_hash(new_password) 

261 await repositories_users.update_password(user, new_hashed_password, db) 

262 return JSONResponse( 

263 content={"message": "Password reset successful"}, status_code=200 

264 ) 

265 

266 

267@router.post("/cash/set") 

268async def set_cash( 

269 key: str, 

270 value: str, 

271 redis_client: Redis = Depends(get_redis_client), 

272 user: User = Depends(auth_serviсe.get_current_user), 

273): 

274 """ 

275 Sets a key-value pair in the Redis cache. 

276 

277 Parameters: 

278 - key (str): The key to be set in the cache. 

279 - value (str): The value to be associated with the key in the cache. 

280 - redis_client (Redis): The Redis client used to interact with the cache. 

281 - user (User): The authenticated user making the request. 

282 

283 Returns: 

284 - None: This method does not return a value. 

285 

286 Raises: 

287 - HTTPException: If the user is not authenticated. 

288 """ 

289 await redis_client.set(key, value) 

290 

291 

292@router.get("/cash/get/{key}") 

293async def get_cash( 

294 key: str, 

295 redis_client: Redis = Depends(get_redis_client), 

296 user: User = Depends(auth_serviсe.get_current_user), 

297): 

298 """ 

299 Retrieves the value associated with the given key from the Redis cache. 

300 

301 Parameters: 

302 - key (str): The key to be retrieved from the cache. 

303 - redis_client (Redis): The Redis client used to interact with the cache. 

304 - user (User): The authenticated user making the request. 

305 

306 Returns: 

307 - dict: A dictionary containing the key and its associated value in the cache. 

308 

309 Raises: 

310 - HTTPException: If the user is not authenticated. 

311 """ 

312 value = await redis_client.get(key) 

313 return {key, value}