Tests and custom exception types in authentication router

Author: chriscarrollsmith

routers/authentication.py

@@ -6,9 +6,8 @@
 from fastapi.responses import RedirectResponse
 from pydantic import BaseModel, EmailStr, ConfigDict
 from sqlmodel import Session, select
-from utils.models import User, UserPassword
+from utils.models import User, UserPassword, DataIntegrityError
 from utils.auth import (
-    AuthenticationError,
     get_session,
     get_user_from_reset_token,
     create_password_validator,
@@ -29,6 +28,40 @@
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
+# --- Custom Exceptions ---
+
+
+class EmailAlreadyRegisteredError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=409,
+            detail="This email is already registered"
+        )
+
+
+class InvalidCredentialsError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=401,
+            detail="Invalid credentials"
+        )
+
+
+class InvalidResetTokenError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=401,
+            detail="Invalid or expired password reset token; please request a new one"
+        )
+
+
+class InvalidEmailUpdateTokenError(HTTPException):
+    def __init__(self):
+        super().__init__(
+            status_code=401,
+            detail="Invalid or expired email update token; please request a new one"
+        )
+
 
 # --- Server Request and Response Models ---
 
@@ -145,7 +178,7 @@ async def register(
         User.email == user.email)).first()
 
     if db_user:
-        raise HTTPException(status_code=400, detail="Email already registered")
+        raise EmailAlreadyRegisteredError()
 
     # Hash the password
     hashed_password = get_password_hash(user.password)
@@ -179,7 +212,7 @@ async def login(
         User.email == user.email)).first()
 
     if not db_user or not db_user.password or not verify_password(user.password, db_user.password.hashed_password):
-        raise HTTPException(status_code=400, detail="Invalid credentials")
+        raise InvalidCredentialsError()
 
     # Create access token
     access_token = create_access_token(
@@ -277,7 +310,7 @@ async def reset_password(
         user.email, user.token, session)
 
     if not authorized_user or not reset_token:
-        raise HTTPException(status_code=400, detail="Invalid or expired token")
+        raise InvalidResetTokenError()
 
     # Update password and mark token as used
     if authorized_user.password:
@@ -318,16 +351,10 @@ async def request_email_update(
     ).first()
 
     if existing_user:
-        raise HTTPException(
-            status_code=400,
-            detail="This email is already registered"
-        )
+        raise EmailAlreadyRegisteredError()
 
-    if not user or not user.id:
-        raise HTTPException(
-            status_code=400,
-            detail="User not found"
-        )
+    if not user.id:
+        raise DataIntegrityError(resource="User id")
 
     # Send confirmation email
     send_email_update_confirmation(
@@ -355,7 +382,7 @@ async def confirm_email_update(
     )
 
     if not user or not update_token:
-        raise AuthenticationError()
+        raise InvalidResetTokenError()
 
     # Update email and mark token as used
     user.email = new_email

routers/organization.py

@@ -10,6 +10,8 @@
 
 logger = getLogger("uvicorn.error")
 
+router = APIRouter(prefix="/organizations", tags=["organizations"])
+
 # --- Custom Exceptions ---
 
 
@@ -37,9 +39,6 @@ def __init__(self):
         )
 
 
-router = APIRouter(prefix="/organizations", tags=["organizations"])
-
-
 # --- Server Request and Response Models ---
 
 

tests/test_auth.py

@@ -215,9 +215,9 @@ def test_send_email_update_confirmation(mock_send: MagicMock) -> None:
     # Test existing token case
     session.reset_mock()
     session.exec.return_value.first.return_value = EmailUpdateToken()  # Existing token
-    
+
     send_email_update_confirmation(current_email, new_email, user_id, session)
-    
+
     # Verify no new token was created or email sent
     assert not session.add.called
     assert not session.commit.called
@@ -228,7 +228,7 @@ def test_get_user_from_email_update_token() -> None:
     Tests retrieving a user using an email update token.
     """
     session = MagicMock()
-    
+
     # Test valid token
     mock_user = User(id=1, email="[email protected]")
     mock_token = EmailUpdateToken(

tests/test_authentication.py

@@ -9,11 +9,12 @@
 from html import unescape
 
 from main import app
-from utils.models import User, PasswordResetToken
+from utils.models import User, PasswordResetToken, UserPassword, EmailUpdateToken
 from utils.auth import (
     create_access_token,
     verify_password,
     validate_token,
+    get_password_hash
 )
 from .conftest import SetupError
 
@@ -199,7 +200,7 @@ def test_register_with_existing_email(unauth_client: TestClient, test_user: User
             "confirm_password": "Test123!@#"
         }
     )
-    assert response.status_code == 400
+    assert response.status_code == 409
 
 
 def test_login_with_invalid_credentials(unauth_client: TestClient, test_user: User):
@@ -210,7 +211,7 @@ def test_login_with_invalid_credentials(unauth_client: TestClient, test_user: Us
             "password": "WrongPass123!@#"
         }
     )
-    assert response.status_code == 400
+    assert response.status_code == 401
 
 
 def test_password_reset_with_invalid_token(unauth_client: TestClient, test_user: User):
@@ -223,7 +224,7 @@ def test_password_reset_with_invalid_token(unauth_client: TestClient, test_user:
             "confirm_new_password": "NewPass123!@#"
         }
     )
-    assert response.status_code == 400
+    assert response.status_code == 401
 
 
 def test_password_reset_email_url(unauth_client: TestClient, session: Session, test_user: User, mock_resend_send):
@@ -264,3 +265,144 @@ def test_password_reset_email_url(unauth_client: TestClient, session: Session, t
     assert parsed.path == str(reset_password_path)
     assert query_params["email"][0] == test_user.email
     assert query_params["token"][0] == reset_token.token
+
+
+def test_request_email_update_success(auth_client: TestClient, test_user: User, mock_resend_send):
+    """Test successful email update request"""
+    new_email = "[email protected]"
+    
+    response = auth_client.post(
+        app.url_path_for("request_email_update"),
+        data={"new_email": new_email},
+        follow_redirects=False
+    )
+    
+    assert response.status_code == 303
+    assert "profile?email_update_requested=true" in response.headers["location"]
+    
+    # Verify email was "sent"
+    mock_resend_send.assert_called_once()
+    call_args = mock_resend_send.call_args[0][0]
+    
+    # Verify email content
+    assert call_args["to"] == [test_user.email]
+    assert call_args["from"] == "[email protected]"
+    assert "Confirm Email Update" in call_args["subject"]
+    assert "confirm_email_update" in call_args["html"]
+    assert new_email in call_args["html"]
+
+
+def test_request_email_update_already_registered(auth_client: TestClient, session: Session, test_user: User):
+    """Test email update request with already registered email"""
+    # Create another user with the target email
+    existing_email = "[email protected]"
+    existing_user = User(
+        name="Existing User",
+        email=existing_email,
+        password=UserPassword(hashed_password=get_password_hash("Test123!@#"))
+    )
+    session.add(existing_user)
+    session.commit()
+    
+    response = auth_client.post(
+        app.url_path_for("request_email_update"),
+        data={"new_email": existing_email}
+    )
+    
+    assert response.status_code == 409
+    assert "already registered" in response.text
+
+
+def test_request_email_update_unauthenticated(unauth_client: TestClient):
+    """Test email update request without authentication"""
+    response = unauth_client.post(
+        app.url_path_for("request_email_update"),
+        data={"new_email": "[email protected]"},
+        follow_redirects=False
+    )
+    
+    assert response.status_code == 303  # Redirect to login
+
+
+def test_confirm_email_update_success(unauth_client: TestClient, session: Session, test_user: User):
+    """Test successful email update confirmation"""
+    new_email = "[email protected]"
+    
+    # Create an email update token
+    update_token = EmailUpdateToken(user_id=test_user.id)
+    session.add(update_token)
+    session.commit()
+    
+    response = unauth_client.get(
+        app.url_path_for("confirm_email_update"),
+        params={
+            "user_id": test_user.id,
+            "token": update_token.token,
+            "new_email": new_email
+        },
+        follow_redirects=False
+    )
+    
+    assert response.status_code == 303
+    assert "profile?email_updated=true" in response.headers["location"]
+    
+    # Verify email was updated
+    session.refresh(test_user)
+    assert test_user.email == new_email
+    
+    # Verify token was marked as used
+    session.refresh(update_token)
+    assert update_token.used
+    
+    # Verify new auth cookies were set
+    cookies = response.cookies
+    assert "access_token" in cookies
+    assert "refresh_token" in cookies
+
+
+def test_confirm_email_update_invalid_token(unauth_client: TestClient, session: Session, test_user: User):
+    """Test email update confirmation with invalid token"""
+    response = unauth_client.get(
+        app.url_path_for("confirm_email_update"),
+        params={
+            "user_id": test_user.id,
+            "token": "invalid_token",
+            "new_email": "[email protected]"
+        }
+    )
+    
+    assert response.status_code == 401
+    assert "Invalid or expired" in response.text
+    
+    # Verify email was not updated
+    session.refresh(test_user)
+    assert test_user.email == "[email protected]"
+
+
+def test_confirm_email_update_used_token(unauth_client: TestClient, session: Session, test_user: User):
+    """Test email update confirmation with already used token"""
+    # Create an already used token
+    used_token = PasswordResetToken(
+        user_id=test_user.id,
+        token="test_used_token",
+        used=True,
+        token_type="email_update"
+    )
+    session.add(used_token)
+    session.commit()
+    
+    response = unauth_client.get(
+        app.url_path_for("confirm_email_update"),
+        params={
+            "user_id": test_user.id,
+            "token": used_token.token,
+            "new_email": "[email protected]"
+        }
+    )
+    
+    assert response.status_code == 401
+    assert "Invalid or expired" in response.text
+    
+    # Verify email was not updated
+    session.refresh(test_user)
+    assert test_user.email == "[email protected]"

utils/auth.py

@@ -87,6 +87,13 @@ def replacer(match: re.Match) -> str:
 # --- Custom Exceptions ---
 
 
+class NeedsNewTokens(Exception):
+    def __init__(self, user: User, access_token: str, refresh_token: str):
+        self.user = user
+        self.access_token = access_token
+        self.refresh_token = refresh_token
+
+
 class AuthenticationError(HTTPException):
     def __init__(self):
         super().__init__(
@@ -324,13 +331,6 @@ def get_optional_user(
     return None
 
 
-class NeedsNewTokens(Exception):
-    def __init__(self, user: User, access_token: str, refresh_token: str):
-        self.user = user
-        self.access_token = access_token
-        self.refresh_token = refresh_token
-
-
 def generate_password_reset_url(email: str, token: str) -> str:
     """
     Generates the password reset URL with proper query parameters.