7108137235

Committed 15 Jun 2024 07:46PM UTC coverage: 93.523%. Remained the same

Build Type

push

coveralls-python

Committed by memphis-tools

Commit Message

Merge branch 'development' into 'master'

Run Details

691 of 756 branches covered (91.4%)

Branch coverage included in aggregate %.

6 of 7 new or added lines in 1 file covered. (85.71%)

3 existing lines in 1 file now uncovered.

1807 of 1915 relevant lines covered (94.36%)

0.94 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.13

/app/packages/fastapi/routes/routes_and_authentication.py

"""The FastAPI routes"""

from datetime import datetime, timedelta, timezone
from typing import Annotated
from fastapi import Depends, FastAPI, HTTPException, status, Request
from fastapi.responses import HTMLResponse
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.templating import Jinja2Templates
from werkzeug.security import generate_password_hash
import jwt

from app.packages import handle_passwords, log_events
from app.packages.database.models import models
from app.packages.fastapi.models.fastapi_models import (
    NewUserInDBModel,
    Token,
)
from .dependencies import (
    get_current_active_user,
    session,
    authenticate_user,
    SECRET_KEY,
    ALGORITHM,
)


# Because of the current architecture, in order to run tests_users and tests_authentication
# We set "noqa: F401" for the unused get_user dependencie.
from .dependencies import get_user  # noqa: F401
from .routers import books_categories, books, comments, quotes, users


ACCESS_TOKEN_EXPIRE_MINUTES = 30
app: FastAPI = FastAPI(
    title="DUMMY-OPS API",
    docs_url="/api/v1/docs/",
    openapi_url="/api/v1/openapi.json",
    swagger_ui_parameters={"defaultModelsExpandDepth": -1},
)
templates = Jinja2Templates(directory="templates")

protected_routes = [
    books_categories.router,
    books.router,
    comments.router,
    quotes.router,
    users.router,
]

for router in protected_routes:
    app.include_router(router, dependencies=[Depends(get_current_active_user)])


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    """create an access token for authenticated user"""
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.now(timezone.utc) + expires_delta
    else:
        expire = datetime.now(timezone.utc) + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(payload=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


def get_password_hash(password):
    """get hash from password"""
    return generate_password_hash(password, "pbkdf2:sha256", salt_length=8)


@app.get("/docs", response_class=HTMLResponse)
async def custom_swagger_ui(request: Request):
   return templates.TemplateResponse("custom_swagger_ui.html", {"request": request})


@app.post("/api/v1/token/", tags=["DEFAULT"])
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
) -> Token:
    """return a jwt access token to authenticated user"""
    user = authenticate_user(str(form_data.username).lower(), form_data.password)
    if not user:
        logs_context = {"username": f"{str(form_data.username).lower()}"}
        log_events.log_event(
            "[+] FastAPI - Utilisateur inconnu cherche à obtenir un token.",
            logs_context,
        )
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nom utilisateur ou mot de passe invalide",
            headers={"WWW-Authenticate": "Bearer"},
        )
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": form_data.username}, expires_delta=access_token_expires
    )
    return Token(access_token=access_token, token_type="bearer")


@app.post("/api/v1/register/", tags=["DEFAULT"])
async def register(user: NewUserInDBModel):
    """
    register new user.
    """
    if any(
        [
            str(user.username) == "string",
            str(user.email) == "string",
            str(user.password) == "string",
            str(user.password_check) == "string",
        ]
    ):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Saisie invalide, mot clef string non utilisable.",
        )
    valid_password = handle_passwords.check_password(str(user.password))
    if not valid_password:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Mot de passe trop simple, essayez de nouveau.",
        )
    hashed_password = generate_password_hash(
        user.password, "pbkdf2:sha256", salt_length=8
    )
    user_in_db = (
        session.query(models.User)
        .filter_by(username=str(user.username).lower())
        .first()
    )
    if user_in_db:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Nom utilisateur existe deja, veuillez le modifier",
        )
    user_email = (
        session.query(models.User).filter_by(email=str(user.email).lower()).first()
    )
    if user_email:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED, detail="Email existe deja en base"
        )
    if user.password != user.password_check:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Mots de passe ne correspondent pas",
        )

    new_user = models.User(
        username=str(user.username).lower(),
        email=str(user.email).lower(),
        hashed_password=hashed_password,
    )
    logs_context = {
        "username": f"{str(user.username).lower()}",
        "email": f"{str(user.email).lower()}",
    }
    log_events.log_event("[+] FastAPI - Création compte utilisateur.", logs_context)
    session.add(new_user)
    session.commit()
    return new_user

1	"""The FastAPI routes"""
2
3	from datetime import datetime, timedelta, timezone	1✔
4	from typing import Annotated	1✔
5	from fastapi import Depends, FastAPI, HTTPException, status, Request	1✔
6	from fastapi.responses import HTMLResponse	1✔
7	from fastapi.security import OAuth2PasswordRequestForm	1✔
8	from fastapi.templating import Jinja2Templates	1✔
9	from werkzeug.security import generate_password_hash	1✔
10	import jwt	1✔
11
12	from app.packages import handle_passwords, log_events	1✔
13	from app.packages.database.models import models	1✔
14	from app.packages.fastapi.models.fastapi_models import (	1✔
15	NewUserInDBModel,
16	Token,
17	)
18	from .dependencies import (	1✔
19	get_current_active_user,
20	session,
21	authenticate_user,
22	SECRET_KEY,
23	ALGORITHM,
24	)
25
26
27	# Because of the current architecture, in order to run tests_users and tests_authentication
28	# We set "noqa: F401" for the unused get_user dependencie.
29	from .dependencies import get_user # noqa: F401	1✔
30	from .routers import books_categories, books, comments, quotes, users	1✔
31
32
33	ACCESS_TOKEN_EXPIRE_MINUTES = 30	1✔
34	app: FastAPI = FastAPI(	1✔
35	title="DUMMY-OPS API",
36	docs_url="/api/v1/docs/",
37	openapi_url="/api/v1/openapi.json",
38	swagger_ui_parameters={"defaultModelsExpandDepth": -1},
39	)
40	templates = Jinja2Templates(directory="templates")	1✔
41
42	protected_routes = [	1✔
43	books_categories.router,
44	books.router,
45	comments.router,
46	quotes.router,
47	users.router,
48	]
49
50	for router in protected_routes:	1✔
51	app.include_router(router, dependencies=[Depends(get_current_active_user)])	1✔
52
53
54	def create_access_token(data: dict, expires_delta: timedelta \| None = None):	1✔
55	"""create an access token for authenticated user"""
56	to_encode = data.copy()	1✔
57	if expires_delta:	1!
58	expire = datetime.now(timezone.utc) + expires_delta	1✔
59	else:
UNCOV 60	expire = datetime.now(timezone.utc) + timedelta(minutes=15)	×
61	to_encode.update({"exp": expire})	1✔
62	encoded_jwt = jwt.encode(payload=to_encode, key=SECRET_KEY, algorithm=ALGORITHM)	1✔
63	return encoded_jwt	1✔
64
65
66	def get_password_hash(password):	1✔
67	"""get hash from password"""
68	return generate_password_hash(password, "pbkdf2:sha256", salt_length=8)	1✔
69
70
71	@app.get("/docs", response_class=HTMLResponse)	1✔
72	async def custom_swagger_ui(request: Request):	1✔
NEW 73	return templates.TemplateResponse("custom_swagger_ui.html", {"request": request})	×
74
75
76	@app.post("/api/v1/token/", tags=["DEFAULT"])	1✔
77	async def login_for_access_token(	1✔
78	form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
79	) -> Token:
80	"""return a jwt access token to authenticated user"""
81	user = authenticate_user(str(form_data.username).lower(), form_data.password)	1✔
82	if not user:	1✔
83	logs_context = {"username": f"{str(form_data.username).lower()}"}	1✔
84	log_events.log_event(	1✔
85	"[+] FastAPI - Utilisateur inconnu cherche à obtenir un token.",
86	logs_context,
87	)
88	raise HTTPException(	1✔
89	status_code=status.HTTP_401_UNAUTHORIZED,
90	detail="Nom utilisateur ou mot de passe invalide",
91	headers={"WWW-Authenticate": "Bearer"},
92	)
93	access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)	1✔
94	access_token = create_access_token(	1✔
95	data={"sub": form_data.username}, expires_delta=access_token_expires
96	)
97	return Token(access_token=access_token, token_type="bearer")	1✔
98
99
100	@app.post("/api/v1/register/", tags=["DEFAULT"])	1✔
101	async def register(user: NewUserInDBModel):	1✔
102	"""
103	register new user.
104	"""
105	if any(	1✔
106	[
107	str(user.username) == "string",
108	str(user.email) == "string",
109	str(user.password) == "string",
110	str(user.password_check) == "string",
111	]
112	):
113	raise HTTPException(	1✔
114	status_code=status.HTTP_401_UNAUTHORIZED,
115	detail="Saisie invalide, mot clef string non utilisable.",
116	)
117	valid_password = handle_passwords.check_password(str(user.password))	1✔
118	if not valid_password:	1!
UNCOV 119	raise HTTPException(	×
120	status_code=status.HTTP_401_UNAUTHORIZED,
121	detail="Mot de passe trop simple, essayez de nouveau.",
122	)
123	hashed_password = generate_password_hash(	1✔
124	user.password, "pbkdf2:sha256", salt_length=8
125	)
126	user_in_db = (	1✔
127	session.query(models.User)
128	.filter_by(username=str(user.username).lower())
129	.first()
130	)
131	if user_in_db:	1✔
132	raise HTTPException(	1✔
133	status_code=status.HTTP_401_UNAUTHORIZED,
134	detail="Nom utilisateur existe deja, veuillez le modifier",
135	)
136	user_email = (	1✔
137	session.query(models.User).filter_by(email=str(user.email).lower()).first()
138	)
139	if user_email:	1✔
140	raise HTTPException(	1✔
141	status_code=status.HTTP_401_UNAUTHORIZED, detail="Email existe deja en base"
142	)
143	if user.password != user.password_check:	1!
UNCOV 144	raise HTTPException(	×
145	status_code=status.HTTP_401_UNAUTHORIZED,
146	detail="Mots de passe ne correspondent pas",
147	)
148
149	new_user = models.User(	1✔
150	username=str(user.username).lower(),
151	email=str(user.email).lower(),
152	hashed_password=hashed_password,
153	)
154	logs_context = {	1✔
155	"username": f"{str(user.username).lower()}",
156	"email": f"{str(user.email).lower()}",
157	}
158	log_events.log_event("[+] FastAPI - Création compte utilisateur.", logs_context)	1✔
159	session.add(new_user)	1✔
160	session.commit()	1✔
161	return new_user	1✔

memphis-tools / dummy_fastapi_flask_blog_app / 7108137235

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous