Coverage for app/api/routes/login.py: 84%
57 statements
« prev ^ index » next coverage.py v7.6.1, created at 2026-03-24 11:40 +0000
« prev ^ index » next coverage.py v7.6.1, created at 2026-03-24 11:40 +0000
1from datetime import timedelta
2from typing import Annotated, Any
4from fastapi import APIRouter, Depends, HTTPException
5from fastapi.responses import HTMLResponse
6from fastapi.security import OAuth2PasswordRequestForm
8from app import crud
9from app.api.deps import CurrentUser, SessionDep, get_current_active_superuser
10from app.core import security
11from app.core.config import settings
12from app.core.security import get_password_hash
13from app.models import Message, NewPassword, Token, UserPublic
14from app.utils import (
15 generate_password_reset_token,
16 generate_reset_password_email,
17 send_email,
18 verify_password_reset_token,
19)
21router = APIRouter()
24@router.post("/login/access-token")
25def login_access_token(
26 session: SessionDep, form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
27) -> Token:
28 """
29 OAuth2 compatible token login, get an access token for future requests
30 """
31 user = crud.authenticate(
32 session=session, email=form_data.username, password=form_data.password
33 )
34 if not user:
35 raise HTTPException(status_code=400, detail="Incorrect email or password")
36 elif not user.is_active:
37 raise HTTPException(status_code=400, detail="Inactive user")
38 access_token_expires = timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES)
39 return Token(
40 access_token=security.create_access_token(
41 user.id, expires_delta=access_token_expires
42 )
43 )
46@router.post("/login/test-token", response_model=UserPublic)
47def test_token(current_user: CurrentUser) -> Any:
48 """
49 Test access token
50 """
51 return current_user
54@router.post("/password-recovery/{email}")
55def recover_password(email: str, session: SessionDep) -> Message:
56 """
57 Password Recovery
58 """
59 user = crud.get_user_by_email(session=session, email=email)
61 if not user:
62 raise HTTPException(
63 status_code=404,
64 detail="The user with this email does not exist in the system.",
65 )
66 password_reset_token = generate_password_reset_token(email=email)
67 email_data = generate_reset_password_email(
68 email_to=user.email, email=email, token=password_reset_token
69 )
70 send_email(
71 email_to=user.email,
72 subject=email_data.subject,
73 html_content=email_data.html_content,
74 )
75 return Message(message="Password recovery email sent")
78@router.post("/reset-password/")
79def reset_password(session: SessionDep, body: NewPassword) -> Message:
80 """
81 Reset password
82 """
83 email = verify_password_reset_token(token=body.token)
84 if not email:
85 raise HTTPException(status_code=400, detail="Invalid token")
86 user = crud.get_user_by_email(session=session, email=email)
87 if not user:
88 raise HTTPException(
89 status_code=404,
90 detail="The user with this email does not exist in the system.",
91 )
92 elif not user.is_active:
93 raise HTTPException(status_code=400, detail="Inactive user")
94 hashed_password = get_password_hash(password=body.new_password)
95 user.hashed_password = hashed_password
96 session.add(user)
97 session.commit()
98 return Message(message="Password updated successfully")
101@router.post(
102 "/password-recovery-html-content/{email}",
103 dependencies=[Depends(get_current_active_superuser)],
104 response_class=HTMLResponse,
105)
106def recover_password_html_content(email: str, session: SessionDep) -> Any:
107 """
108 HTML Content for Password Recovery
109 """
110 user = crud.get_user_by_email(session=session, email=email)
112 if not user:
113 raise HTTPException(
114 status_code=404,
115 detail="The user with this username does not exist in the system.",
116 )
117 password_reset_token = generate_password_reset_token(email=email)
118 email_data = generate_reset_password_email(
119 email_to=user.email, email=email, token=password_reset_token
120 )
122 return HTMLResponse(
123 content=email_data.html_content, headers={"subject:": email_data.subject}
124 )