fix: recover-password route, add tests and reformat all files

README.md

@@ -67,6 +67,8 @@ An example of .env file:
 RELOAD="True"
 PORT="8000"
 ENVIRONMENT="dev"
+EMAIL_ADDRESS= "YOUR-EMAIL-ADDRESS"
+EMAIL_APP_PASSWORD= "YOUR-EMAIL-PASSWORD-FOR-APPs"
 ```
 
 You can read more about BaseSettings class here: https://pydantic-docs.helpmanual.io/usage/settings/

docker-compose.yml

@@ -24,8 +24,6 @@ services:
       DB_PASS: gestao
       DB_BASE: gestao
       RELOAD: "True"
-      EMAIL_ADDRESS: "YOUR EMAIL ADDRESS"
-      EMAIL_PASSWORD: "YOUR-EMAIL-PASSWORD"
     networks:
     - sindpol_network
 

gestao/db/migrations/versions/2023-11-25-07-13_89f33fa986f8.py

@@ -5,9 +5,8 @@
 Create Date: 2023-11-25 07:13:16.617587
 
 """
-from alembic import op
 import sqlalchemy as sa
-
+from alembic import op
 
 # revision identifiers, used by Alembic.
 revision = "89f33fa986f8"

gestao/db/migrations/versions/2023-12-01-21-33_02379f0ee4ac.py

@@ -5,9 +5,8 @@
 Create Date: 2023-12-01 21:33:24.578159
 
 """
-from alembic import op
 import sqlalchemy as sa
-
+from alembic import op
 
 # revision identifiers, used by Alembic.
 revision = "02379f0ee4ac"

gestao/db/models/user.py

@@ -1,8 +1,9 @@
 from datetime import date, datetime
-from typing import Optional
-from gestao.db.base import BaseMeta
+
 import ormar
 
+from gestao.db.base import BaseMeta
+
 
 class User(ormar.Model):
     class Meta(BaseMeta):

gestao/tests/test_login.py

@@ -3,6 +3,7 @@
 from httpx import AsyncClient
 from starlette import status
 
+from gestao.db.models.user import User
 from gestao.tests.utils import generate_fake_user
 from gestao.web.api.login.utils import generate_password, send_email
 
@@ -17,13 +18,35 @@ def test_generate_password_correct() -> None:
     assert len(password) == 15
 
 
+def test_generate_password_incorrect() -> None:
+    try:
+        generate_password("5")
+    except Exception as e:
+        assert isinstance(e, TypeError)
+
+
+def test_send_email_incorrect() -> None:
+    send_email_value = None
+    try:
+        send_email_value = send_email(
+            user_name="joao",
+            user_email="[email protected]",
+            new_password="null",
+            logo_path="null",
+        )
+    except Exception as e:
+        assert send_email_value is None
+
+
 @pytest.mark.anyio
 async def test_login_user_correct(client: AsyncClient, fastapi_app: FastAPI) -> None:
     url = fastapi_app.url_path_for("create_user")
     user = generate_fake_user()
     response = await client.post(url, json=user)
     assert response.status_code == 200
     user_data = response.json()
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
     user_credentials = {
         "registration": user_data["registration"],
         "password": user_data["password"],
@@ -47,7 +70,7 @@ async def test_login_user_incorrect(client: AsyncClient, fastapi_app: FastAPI) -
 
 
 @pytest.mark.anyio
-async def test_recover_password_incorrect(
+async def test_recover_password_UserNotFound(
     client: AsyncClient, fastapi_app: FastAPI
 ) -> None:
     user_credentials = {
@@ -56,4 +79,25 @@ async def test_recover_password_incorrect(
 
     url = fastapi_app.url_path_for("recover_password")
     response = await client.post(url, json=user_credentials)
-    assert response.status_code == 404
+    assert response.status_code == 400
+
+
+@pytest.mark.anyio
+async def test_recover_password_AuthenticationError(
+    client: AsyncClient, fastapi_app: FastAPI
+) -> None:
+    url = fastapi_app.url_path_for("create_user")
+    user = generate_fake_user()
+    response = await client.post(url, json=user)
+    assert response.status_code == 200
+    user_data = response.json()
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
+
+    user_credentials = {
+        "email": user_data["email"],
+    }
+
+    url = fastapi_app.url_path_for("recover_password")
+    response = await client.post(url, json=user_credentials)
+    assert response.status_code == 400

gestao/tests/test_user.py

@@ -1,9 +1,10 @@
+from uuid import uuid4
+
 import pytest
 from fastapi import FastAPI
 from httpx import AsyncClient
-from starlette import status
-from uuid import uuid4
 
+from gestao.db.models.user import User
 from gestao.tests.utils import generate_fake_user
 
 
@@ -24,10 +25,15 @@ async def test_get_user_correct(client: AsyncClient, fastapi_app: FastAPI) -> No
     assert response.status_code == 200
     user_data = response.json()
     user_id = user_data["id"]
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
     url = fastapi_app.url_path_for("get_user", user_id=user_id)
     response = await client.get(url)
     assert response.status_code == 200
+    user_data = response.json()
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
 
 @pytest.mark.anyio
@@ -43,6 +49,9 @@ async def test_create_user_correct(client: AsyncClient, fastapi_app: FastAPI) ->
     user = generate_fake_user()
     response = await client.post(url, json=user)
     assert response.status_code == 200
+    user_data = response.json()
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
 
 @pytest.mark.anyio
@@ -54,6 +63,20 @@ async def test_create_user_incorrect(client: AsyncClient, fastapi_app: FastAPI)
     assert response.status_code == 422
 
 
+@pytest.mark.anyio
+async def test_create_user_UniqueViolation(
+    client: AsyncClient,
+    fastapi_app: FastAPI,
+) -> None:
+    url = fastapi_app.url_path_for("create_user")
+    user = generate_fake_user()
+    response = await client.post(url, json=user)
+    assert response.status_code == 200
+
+    response = await client.post(url, json=user)
+    assert response.status_code == 400
+
+
 @pytest.mark.anyio
 async def test_update_user_correct(client: AsyncClient, fastapi_app: FastAPI) -> None:
     url = fastapi_app.url_path_for("create_user")
@@ -62,6 +85,8 @@ async def test_update_user_correct(client: AsyncClient, fastapi_app: FastAPI) ->
     assert response.status_code == 200
     user_data = response.json()
     user_id = user_data["id"]
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
     url = fastapi_app.url_path_for("update_user", user_id=user_id)
     response = await client.put(
@@ -71,6 +96,9 @@ async def test_update_user_correct(client: AsyncClient, fastapi_app: FastAPI) ->
         },
     )
     assert response.status_code == 200
+    user_data = response.json()
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
 
 @pytest.mark.anyio
@@ -88,6 +116,8 @@ async def test_delete_user_correct(client: AsyncClient, fastapi_app: FastAPI) ->
     assert response.status_code == 200
     user_data = response.json()
     user_id = user_data["id"]
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
     url = fastapi_app.url_path_for("delete_user", user_id=user_id)
     response = await client.delete(url)
@@ -109,12 +139,24 @@ async def test_disable_user_correct(client: AsyncClient, fastapi_app: FastAPI) -
     assert response.status_code == 200
     user_data = response.json()
     var = user_data["id"]
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
     url = fastapi_app.url_path_for("disable_user", user_id=var)
     response = await client.patch(url)
     assert response.status_code == 200
 
 
+@pytest.mark.anyio
+async def test_disable_user_incorrect(
+    client: AsyncClient,
+    fastapi_app: FastAPI,
+) -> None:
+    url = fastapi_app.url_path_for("disable_user", user_id=str(uuid4()))
+    response = await client.patch(url)
+    assert response.status_code == 404
+
+
 @pytest.mark.anyio
 async def test_enable_user_correct(client: AsyncClient, fastapi_app: FastAPI) -> None:
     url = fastapi_app.url_path_for("create_user")
@@ -123,21 +165,14 @@ async def test_enable_user_correct(client: AsyncClient, fastapi_app: FastAPI) ->
     assert response.status_code == 200
     user_data = response.json()
     var = user_data["id"]
+    user_instance = User(**user_data)
+    assert isinstance(user_instance, User)
 
     url = fastapi_app.url_path_for("enable_user", user_id=var)
     response = await client.patch(url)
     assert response.status_code == 200
 
 
-@pytest.mark.anyio
-async def test_disable_user_incorrect(
-    client: AsyncClient, fastapi_app: FastAPI
-) -> None:
-    url = fastapi_app.url_path_for("disable_user", user_id=str(uuid4()))
-    response = await client.patch(url)
-    assert response.status_code == 404
-
-
 @pytest.mark.anyio
 async def test_enable_user_incorrect(client: AsyncClient, fastapi_app: FastAPI) -> None:
     url = fastapi_app.url_path_for("enable_user", user_id=str(uuid4()))

gestao/web/api/login/utils.py

@@ -1,7 +1,7 @@
-import os
-import smtplib
 import email.message
+import os
 import secrets
+import smtplib
 import string
 
 
@@ -11,16 +11,21 @@ def generate_password(length: int = 8) -> str:
     return password
 
 
-def send_email(user_name: str, user_email: str, user_message: str) -> None:
-
+def send_email(
+    user_name: str,
+    user_email: str,
+    new_password: str,
+    logo_path: str,
+) -> None:
     email_data = {
         "email_address": os.getenv("EMAIL_ADDRESS"),
-        "email_password": os.getenv("EMAIL_PASSWORD"),
-        "email_subject": "recover-password SINDPOL-DF",
-        "email_body": """
-                        <p>recover-password<p>
-                        <p>name: {}<p>
-                        <p>new-password: {}<p>
+        "email_password": os.getenv("EMAIL_APP_PASSWORD"),
+        "email_subject": "recover-password SG-SINDPOL",
+        "email_body": f"""
+                        <img src="{logo_path}" width=200 style="height: auto;">
+                        <p>recover-password system from SG-SINDPOL<p>
+                        <p>name: {user_name}<p>
+                        <p>new-password: {new_password}<p>
                     """,
     }
 
@@ -29,7 +34,7 @@ def send_email(user_name: str, user_email: str, user_message: str) -> None:
     msg["From"] = email_data["email_address"]
     msg["To"] = user_email
     msg.add_header("Content-Type", "text/html")
-    msg.set_payload(email_data["email_body"].format(user_name, user_message))
+    msg.set_payload(email_data["email_body"])
 
     s = smtplib.SMTP("smtp.gmail.com: 587")
     s.starttls()

gestao/web/api/login/views.py

@@ -1,6 +1,6 @@
 import logging
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Request
 
 from gestao.db.models.user import User
 from gestao.web.api.login.schemas import AuthUserDTO, RecoverPasswordDTO
@@ -23,13 +23,22 @@ async def login_user(login_data: AuthUserDTO) -> User:
         )
 
 
-@router.post("/recover_password")
-async def recover_password(recover_data: RecoverPasswordDTO) -> None:
+@router.post("/recover-password")
+async def recover_password(request: Request, recover_data: RecoverPasswordDTO) -> None:
     try:
+        url_logo = str(request.url_for("static", path="logo.png"))
         user = await User.objects.get(**recover_data.dict())
         new_password = generate_password()
-        send_email(user.name, user.email, new_password)
+        send_email(
+            user_name=user.fullName,
+            user_email=user.email,
+            new_password=new_password,
+            logo_path=url_logo,
+        )
         await user.update(password=new_password)
     except Exception:
-        logging.error("User not found", exc_info=True)
-        raise HTTPException(status_code=404, detail="User not found")
+        logging.error("Error occurred while sending email", exc_info=True)
+        raise HTTPException(
+            status_code=400,
+            detail="Error occurred while sending email",
+        )

gestao/web/api/router.py

@@ -1,6 +1,6 @@
 from fastapi.routing import APIRouter
 
-from gestao.web.api import echo, monitoring, user, login
+from gestao.web.api import echo, login, monitoring, user
 
 api_router = APIRouter()
 api_router.include_router(monitoring.router)

gestao/web/application.py

@@ -3,6 +3,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import UJSONResponse
+from fastapi.staticfiles import StaticFiles
 
 from gestao.web.api.router import api_router
 from gestao.web.lifetime import register_shutdown_event, register_startup_event
@@ -41,4 +42,6 @@ def get_app() -> FastAPI:
     # Main router for the API.
     app.include_router(router=api_router, prefix="/api")
 
+    app.mount("/api/static", StaticFiles(directory="files"), name="static")
+
     return app