Merge pull request #528 from akvo/develop

[#527] Make email case insensitive

backend/db/crud_user.py

@@ -1,6 +1,6 @@
 from uuid import uuid4
 from fastapi import HTTPException, status
-from sqlalchemy import or_
+from sqlalchemy import or_, func
 from sqlalchemy.orm import Session
 from models.user import User, UserBase, UserDict
 from models.user import UserInvitation, UserUpdateByAdmin
@@ -10,34 +10,42 @@
 
 
 def get_user_by_email(session: Session, email: str) -> User:
-    user = session.query(User).filter(User.email == email).first()
+    user = (
+        session.query(User)
+        .filter(func.lower(User.email) == email.lower().strip())
+        .first()
+    )
     return user
 
 
 def get_user_by_id(session: Session, id: int) -> UserDict:
     user = session.query(User).filter(User.id == id).first()
     if user is None:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND,
-                            detail=f"user {id} not found")
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"user {id} not found",
+        )
     return user
 
 
-def add_user(session: Session,
-             payload: UserBase,
-             invitation: Optional[bool] = False) -> UserDict:
+def add_user(
+    session: Session, payload: UserBase, invitation: Optional[bool] = False
+) -> UserDict:
     try:
         password = payload.password.get_secret_value()
     except AttributeError:
         password = payload.password
-    user = User(name=payload.name,
-                email=payload.email,
-                phone_number=payload.phone_number,
-                password=password,
-                role=payload.role,
-                organisation=payload.organisation,
-                invitation=str(uuid4()) if invitation else None,
-                questionnaires=payload.questionnaires,
-                approved=True if invitation else False)
+    user = User(
+        name=payload.name,
+        email=payload.email,
+        phone_number=payload.phone_number,
+        password=password,
+        role=payload.role,
+        organisation=payload.organisation,
+        invitation=str(uuid4()) if invitation else None,
+        questionnaires=payload.questionnaires,
+        approved=True if invitation else False,
+    )
     session.add(user)
     session.commit()
     session.flush()
@@ -54,12 +62,13 @@ def verify_user_email(session: Session, email: str) -> UserDict:
     return user
 
 
-def update_user_by_admin(session: Session, id: int, approved: bool,
-                         payload: UserUpdateByAdmin) -> UserDict:
+def update_user_by_admin(
+    session: Session, id: int, approved: bool, payload: UserUpdateByAdmin
+) -> UserDict:
     user = get_user_by_id(session=session, id=id)
-    user.organisation = payload['organisation']
-    user.role = payload['role']
-    user.questionnaires = payload['questionnaires']
+    user.organisation = payload["organisation"]
+    user.role = payload["role"]
+    user.questionnaires = payload["questionnaires"]
     if approved:
         user.approved = approved
     session.commit()
@@ -77,45 +86,56 @@ def update_password(session: Session, id: int, password: str) -> UserDict:
     return user
 
 
-def filter_user(session: Session,
-                approved: bool,
-                search: Optional[str] = None,
-                organisation: Optional[List[int]] = None):
+def filter_user(
+    session: Session,
+    approved: bool,
+    search: Optional[str] = None,
+    organisation: Optional[List[int]] = None,
+):
     user = session.query(User)
     user = user.filter(User.approved == approved)
     if search:
         user = user.filter(
             or_(
                 User.name.ilike("%{}%".format(search.lower().strip())),
                 User.email.ilike("%{}%".format(search.lower().strip())),
-            ))
+            )
+        )
     if organisation:
         user = user.filter(User.organisation.in_(organisation))
     return user
 
 
-def count(session: Session,
-          approved: bool,
-          search: Optional[str] = None,
-          organisation: Optional[List[int]] = None) -> int:
-    user = filter_user(session=session,
-                       approved=approved,
-                       search=search,
-                       organisation=organisation)
+def count(
+    session: Session,
+    approved: bool,
+    search: Optional[str] = None,
+    organisation: Optional[List[int]] = None,
+) -> int:
+    user = filter_user(
+        session=session,
+        approved=approved,
+        search=search,
+        organisation=organisation,
+    )
     user = user.count()
     return user
 
 
-def get_all_user(session: Session,
-                 approved: bool,
-                 search: Optional[str] = None,
-                 organisation: Optional[List[int]] = None,
-                 skip: int = 0,
-                 limit: int = 10) -> List[UserDict]:
-    user = filter_user(session=session,
-                       approved=approved,
-                       search=search,
-                       organisation=organisation)
+def get_all_user(
+    session: Session,
+    approved: bool,
+    search: Optional[str] = None,
+    organisation: Optional[List[int]] = None,
+    skip: int = 0,
+    limit: int = 10,
+) -> List[UserDict]:
+    user = filter_user(
+        session=session,
+        approved=approved,
+        search=search,
+        organisation=organisation,
+    )
     user = user.order_by(User.id.desc()).offset(skip).limit(limit).all()
     return user
 
@@ -127,9 +147,9 @@ def get_invitation(session: Session, invitation: str) -> UserInvitation:
     return user.serialize
 
 
-def accept_invitation(session: Session,
-                      invitation: str,
-                      password=str) -> UserInvitation:
+def accept_invitation(
+    session: Session, invitation: str, password=str
+) -> UserInvitation:
     user = session.query(User).filter(User.invitation == invitation).first()
     if not user:
         return None
@@ -147,8 +167,11 @@ def new_reset_password(session: Session, email: str) -> ResetPasswordBase:
     user = session.query(User).filter(User.email == email).first()
     if not user:
         return None
-    reset_password = session.query(ResetPassword).filter(
-        ResetPassword.user == user.id).first()
+    reset_password = (
+        session.query(ResetPassword)
+        .filter(ResetPassword.user == user.id)
+        .first()
+    )
     if not reset_password:
         reset_password = ResetPassword(user=user.id)
         session.add(reset_password)
@@ -162,8 +185,9 @@ def new_reset_password(session: Session, email: str) -> ResetPasswordBase:
 
 
 def get_reset_password(session: Session, url: str) -> ResetPasswordBase:
-    reset_password = session.query(ResetPassword).filter(
-        ResetPassword.url == url).first()
+    reset_password = (
+        session.query(ResetPassword).filter(ResetPassword.url == url).first()
+    )
     return reset_password
 
 

backend/tests/test_001_auth.py

@@ -214,6 +214,25 @@ async def test_user_register_with_same_email(
         )
         assert res.status_code == 409
 
+    @pytest.mark.asyncio
+    async def test_user_register_with_same_email_case_insensitive(
+        self, app: FastAPI, session: Session, client: AsyncClient
+    ) -> None:
+        # create organisation
+        user_payload = {
+            "name": "Lorem Ipsum",
+            "email": "Support@akvo.org",
+            "phone_number": None,
+            "password": "pass",
+            "role": UserRole.member_user.value,
+            "organisation": 1,
+            "questionnaires": [1],
+        }
+        res = await client.post(
+            app.url_path_for("user:register"), data=user_payload
+        )
+        assert res.status_code == 409
+
     @pytest.mark.asyncio
     async def test_verify_user_email(
         self, app: FastAPI, session: Session, client: AsyncClient