Remove DELETE method from /user endpoint

src/api.py

@@ -17,7 +17,6 @@
 import crud
 from db_config import DbContextManager, create_db_and_tables
 import crud_auth
-from create_admin_user import init_user_admin
 import email_config
 
 ACCESS_TOKEN_EXPIRE_MINUTES = int(os.environ.get("ACCESS_TOKEN_EXPIRE_MINUTES"))
@@ -43,7 +42,7 @@
 @app.on_event("startup")
 async def on_startup():
     await create_db_and_tables()
-    await init_user_admin()
+    await crud_auth.init_user_admin()
 
 
 @app.get("/", include_in_schema=False)
@@ -188,36 +187,6 @@ async def get_user(
         )
 
 
-@app.delete(
-    "/user/{email}",
-    summary="Deleta usuário na api-pgd",
-    tags=["Auth"],
-)
-async def delete_user(
-    user_logged: Annotated[
-        schemas.UsersInputSchema, Depends(crud_auth.get_current_admin_user)
-    ],
-    email: str,
-    db: DbContextManager = Depends(DbContextManager),
-):
-    # Validações
-    if user_logged.email == email:
-        raise HTTPException(
-            status.HTTP_401_UNAUTHORIZED,
-            detail="Usuário não pode se auto deletar",
-        )
-
-    # Call
-    try:
-        return await crud_auth.delete_user(db, email)
-
-    except IntegrityError as exception:
-        raise HTTPException(
-            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
-            detail=f"IntegrityError: {str(exception)}",
-        ) from exception
-
-
 @app.post(
     "/user/forgot_password/{email}",
     summary="Recuperação de Acesso",

src/crud.py

@@ -440,3 +440,11 @@ def truncate_status_participante():
     with SyncSession.begin() as session:
         result = session.execute(text("TRUNCATE status_participante CASCADE;"))
     return result
+
+def truncate_user():
+    """Apaga a tabela users.
+    Usado no ambiente de testes de integração contínua.
+    """
+    with SyncSession.begin() as session:
+        result = session.execute(text("TRUNCATE users CASCADE;"))
+    return result

src/crud_auth.py

@@ -9,11 +9,13 @@
 from passlib.context import CryptContext
 
 import models, schemas
-from db_config import DbContextManager
+from db_config import DbContextManager, async_session_maker
 
 
 SECRET_KEY = str(os.environ.get("SECRET_KEY"))
 ALGORITHM = str(os.environ.get("ALGORITHM"))
+API_PGD_ADMIN_USER = os.environ.get("API_PGD_ADMIN_USER")
+API_PGD_ADMIN_PASSWORD = os.environ.get("API_PGD_ADMIN_PASSWORD")
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@@ -147,6 +149,26 @@ async def get_current_active_user(
     return current_user
 
 
+async def init_user_admin():
+    db_session = async_session_maker()
+
+    if not await get_user(db_session=db_session, email=API_PGD_ADMIN_USER):
+        new_user = models.Users(
+            email=API_PGD_ADMIN_USER,
+            # b-crypt
+            password=get_password_hash(API_PGD_ADMIN_PASSWORD),
+            is_admin=True,
+            cod_SIAPE_instituidora=1,
+        )
+
+        async with db_session as session:
+            session.add(new_user)
+            await session.commit()
+        print(f"API_PGD_ADMIN:  Usuário administrador `{API_PGD_ADMIN_USER}` criado")
+    else:
+        print(f"API_PGD_ADMIN:  Usuário administrador `{API_PGD_ADMIN_USER}` já existe")
+
+
 async def get_current_admin_user(
     current_user: Annotated[schemas.UsersSchema, Depends(get_current_user)]
 ):
@@ -217,36 +239,6 @@ async def update_user(
     return schemas.UsersSchema.model_validate(user)
 
 
-async def delete_user(
-    db_session: DbContextManager,
-    email: str,
-) -> str:
-    """Delete user on api database.
-
-    Args:
-        db_session (DbContextManager): Session with api database
-        email (str): email of the user to be deleted
-
-    Raises:
-        HTTPException: User does not exist
-
-    Returns:
-        str: Message about deletion true or false
-    """
-
-    async with db_session as session:
-        result = await session.execute(select(models.Users).filter_by(email=email))
-        user_to_del = result.unique().scalar_one_or_none()
-
-        if not user_to_del:
-            raise HTTPException(status_code=404, detail=f"Usuário `{email}` não existe")
-
-        await session.delete(user_to_del)
-        await session.commit()
-
-    return f"Usuário `{email}` deletado"
-
-
 async def user_reset_password(
     db_session: DbContextManager, token: str, new_password: str
 ) -> str:

tests/conftest.py

@@ -6,6 +6,7 @@
 import sys
 import json
 from typing import Generator, Optional
+import asyncio
 
 import httpx
 from fastapi.testclient import TestClient
@@ -17,7 +18,9 @@
     truncate_plano_entregas,
     truncate_plano_trabalho,
     truncate_status_participante,
+    truncate_user,
 )
+from crud_auth import init_user_admin
 from api import app
 
 USERS_CREDENTIALS = [
@@ -267,16 +270,8 @@ def truncate_participantes():
 
 @pytest.fixture(scope="module", name="truncate_users")
 def fixture_truncate_users(admin_credentials: dict):
-    for del_user_email in get_all_users(
-        admin_credentials["username"], admin_credentials["password"]
-    ):
-        if del_user_email != admin_credentials["username"]:
-            response = delete_user(
-                admin_credentials["username"],
-                admin_credentials["password"],
-                del_user_email,
-            )
-            response.raise_for_status()
+    truncate_user()
+    asyncio.get_event_loop().run_until_complete(init_user_admin())
 
 
 @pytest.fixture(scope="module", name="register_user_1")

tests/user_test.py

@@ -185,42 +185,6 @@ def test_update_user(client: Client, header_usr_1: dict):  # user is_admin=True
     )
 
 
-# delete /user
-def test_delete_user_not_logged_in(client: Client, header_not_logged_in: dict):
-    response = client.delete(
-        f"/user/{USERS_TEST[0]['email']}", headers=header_not_logged_in
-    )
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_delete_user_logged_in_not_admin(
-    client: Client, header_usr_2: dict  # user is_admin=False
-):
-    response = client.delete(f"/user/{USERS_TEST[0]['email']}", headers=header_usr_2)
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
-def test_delete_user_logged_in_admin(
-    client: Client, header_usr_1: dict  # user is_admin=True
-):
-    response = client.delete(f"/user/{USERS_TEST[0]['email']}", headers=header_usr_1)
-    assert response.status_code == status.HTTP_200_OK
-
-
-def test_delete_user_not_exists_logged_in_admin(
-    client: Client, header_usr_1: dict  # user is_admin=True
-):
-    response = client.delete(f"/user/{USERS_TEST[1]['email']}", headers=header_usr_1)
-    assert response.status_code == status.HTTP_404_NOT_FOUND
-
-
-def test_delete_yourself(client: Client, user1_credentials: dict, header_usr_1: dict):
-    response = client.delete(
-        f"/user/{user1_credentials['email']}", headers=header_usr_1
-    )
-    assert response.status_code == status.HTTP_401_UNAUTHORIZED
-
-
 # forgot/reset password