feat(api): Add the admin only endpoint to update the user role

src/app/api/api_v1/endpoints/users.py

@@ -12,7 +12,7 @@
 from app.crud import UserCRUD
 from app.models import User, UserRole
 from app.schemas.login import TokenPayload
-from app.schemas.users import Cred, CredHash, UserCreate
+from app.schemas.users import Cred, CredHash, RoleUpdate, UserCreate
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
@@ -85,6 +85,33 @@ async def update_user_password(
     return await users.update(user_id, CredHash(hashed_password=pwd))
 
 
+@router.patch("/{user_id}/role", status_code=status.HTTP_200_OK, summary="Updates a user's role")
+async def update_user_role(
+    payload: RoleUpdate,
+    user_id: int = Path(..., gt=0),
+    users: UserCRUD = Depends(get_user_crud),
+    token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN]),
+) -> User:
+    """Promote or demote a user between the `agent` role and the `user` role.
+    Admins are out of scope: neither the requester nor the target user can have their admin role changed here.
+
+    Beware that the role is baked into the access tokens that were already issued, and those are long-lived
+    (see `JWT_EXPIRE_MINUTES`). The new role only applies to tokens minted afterwards, so the user has to log in
+    again for the change to take effect.
+    """
+    telemetry_client.capture(
+        token_payload.sub, event="user-role", properties={"user_id": user_id, "role": payload.role}
+    )
+    if user_id == token_payload.sub:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "Admins cannot change their own role : it can lead to deadlock")
+
+    user = cast(User, await users.get(user_id, strict=True))
+    if user.role == UserRole.ADMIN:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "Cannot change an admin's role")
+
+    return await users.update(user_id, payload)
+
+
 @router.delete("/{user_id}", status_code=status.HTTP_200_OK, summary="Delete a user")
 async def delete_user(
     user_id: int = Path(..., gt=0),

src/app/crud/crud_user.py

@@ -9,12 +9,12 @@
 
 from app.crud.base import BaseCRUD
 from app.models import User
-from app.schemas.users import CredHash
+from app.schemas.users import CredHash, RoleUpdate
 
 __all__ = ["UserCRUD"]
 
 
-class UserCRUD(BaseCRUD[User, User, CredHash]):
+class UserCRUD(BaseCRUD[User, User, Union[CredHash, RoleUpdate]]):
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, User)
 

src/app/schemas/users.py

@@ -3,11 +3,13 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
+from typing import Literal
+
 from pydantic import BaseModel, Field
 
 from app.models import UserRole
 
-__all__ = ["Cred", "CredHash", "UserCreate", "UserCreation"]
+__all__ = ["Cred", "CredHash", "RoleUpdate", "UserCreate", "UserCreation"]
 
 
 # Accesses
@@ -27,6 +29,12 @@ class Role(BaseModel):
     role: UserRole = Field(UserRole.USER)
 
 
+class RoleUpdate(BaseModel):
+    """Admin is excluded : if an admin retrograde himself that can lead to deadlock"""
+
+    role: Literal[UserRole.AGENT, UserRole.USER] = Field(..., examples=["agent"])
+
+
 class UserCreate(Role):
     login: str = Field(..., min_length=3, max_length=50, examples=["JohnDoe"])
     password: str = Field(..., min_length=3, examples=["PickARobustOne"])

src/tests/endpoints/test_users.py

@@ -174,6 +174,49 @@ async def test_delete_user(
         assert response.json() is None
 
 
+@pytest.mark.parametrize(
+    ("user_idx", "user_id", "payload", "status_code", "status_detail"),
+    [
+        (None, 2, {"role": "user"}, 401, "Not authenticated"),
+        (1, 3, {"role": "agent"}, 403, "Incompatible token scope."),
+        (2, 2, {"role": "user"}, 403, "Incompatible token scope."),
+        (0, 0, {"role": "user"}, 422, None),
+        (0, 2, {"role": "admin"}, 422, None),
+        (0, 2, {}, 422, None),
+        (0, 400, {"role": "user"}, 404, "Table User has no corresponding entry."),
+        (0, 1, {"role": "user"}, 403, "Admins cannot change their own role : it can lead to deadlock"),
+        (0, 2, {"role": "user"}, 200, None),
+        (0, 3, {"role": "agent"}, 200, None),
+        (0, 3, {"role": "user"}, 200, None),  # setting role to the actual role
+    ],
+)
+@pytest.mark.asyncio
+async def test_update_user_role(
+    async_client: AsyncClient,
+    user_session: AsyncSession,
+    user_idx: Union[int, None],
+    user_id: int,
+    payload: Dict[str, Any],
+    status_code: int,
+    status_detail: Union[str, None],
+):
+    auth = None
+    if isinstance(user_idx, int):
+        auth = pytest.get_token(
+            pytest.user_table[user_idx]["id"],
+            pytest.user_table[user_idx]["role"].split(),
+            pytest.user_table[user_idx]["organization_id"],
+        )
+
+    response = await async_client.patch(f"/users/{user_id}/role", json=payload, headers=auth)
+    assert response.status_code == status_code, print(response.__dict__)
+    if isinstance(status_detail, str):
+        assert response.json()["detail"] == status_detail
+    if response.status_code // 100 == 2:
+        expected = next(entry for entry in pytest.user_table if entry["id"] == user_id)
+        assert response.json() == {**expected, "role": payload["role"]}
+
+
 @pytest.mark.parametrize(
     ("user_idx", "user_id", "payload", "status_code", "status_detail", "expected_idx"),
     [