Merge pull request #896 from johnson-oragui/fix/auth-request-token-v1

fix: Request reset token sign in.

api/core/dependencies/email/templates/request-token.html

@@ -0,0 +1,40 @@
+{% extends 'base.html' %}
+
+{% block title %}Request Token Login{% endblock %}
+{% block style %}<link rel="stylesheet" href="{{ url_for('email_static', path='css/email-verification.css') }}">{% endblock %}
+
+{% block content %}
+<div class="template-main">
+  <div class="heading">
+    <p class="template-header">Request Token Login</p>
+  </div>
+  <div class="content">
+    <p class="template-receiver-name">Hi {{ first_name }} {{ last_name }},</p>
+    <p class="template-message">
+      Here is the six digit token for login.
+    </p>
+    <div class="editable-content">
+      <p>
+        This link will expire in about 60 seconds from the time of requesting for the token. If
+        you did not make this request, you can ignore this email.
+      </p>
+      <p>To verify your email, please click the button below:</p>
+      <a href="{{ link }}"><button style="
+        display: inline-block;
+        padding: 10px 200px;
+        background-color: orangered;
+        color: #fff;
+        text-decoration: none;
+        border-radius: 10px;
+    ">Login</button></a>
+      <p>
+        Or copy this link:
+        <p>{{ link }}</p>
+      <div class="template-farewell">
+        <p>Regards,</p>
+        <p>Boilerplate</p>
+      </div>
+    </div>
+  </div>
+</div>
+{% endblock %}

api/v1/routes/auth.py

@@ -213,19 +213,32 @@ def refresh_access_token(
 
 
 @auth.post("/request-token", status_code=status.HTTP_200_OK)
-async def request_signin_token(
+async def request_signin_token(background_tasks: BackgroundTasks,
     email_schema: EmailRequest, db: Session = Depends(get_db)
 ):
     """Generate and send a 6-digit sign-in token to the user's email"""
 
     user = user_service.fetch_by_email(db, email_schema.email)
 
     token, token_expiry = user_service.generate_token()
-
     # Save the token and expiry
     user_service.save_login_token(db, user, token, token_expiry)
 
     # Send mail notification
+    link = f'https://anchor-python.teams.hng.tech/login/verify-token?token={token}'
+
+    # Send email in the background
+    background_tasks.add_task(
+        send_email, 
+        recipient=user.email,
+        template_name='request-token.html',
+        subject='Request Token Login',
+        context={
+            'first_name': user.first_name,
+            'last_name': user.last_name,
+            'link': link
+        }
+    )
 
     return success_response(
         status_code=200, message=f"Sign-in token sent to {user.email}"

api/v1/schemas/token.py

@@ -1,5 +1,5 @@
-from typing import Optional
-from pydantic import BaseModel, EmailStr
+from typing import Optional, Annotated
+from pydantic import BaseModel, EmailStr, StringConstraints
 
 
 # Pydantic models for request and response
@@ -14,8 +14,15 @@ class TokenData(BaseModel):
 
 
 class TokenRequest(BaseModel):
-    email: EmailStr
-    token: str
+    email: Optional[EmailStr] = None
+    token: Annotated[
+        str,
+        StringConstraints(
+            max_length=6,
+            min_length=6,
+            strip_whitespace=True
+        )
+    ]
 
 
 class OAuthToken(BaseModel):

api/v1/services/user.py

@@ -530,34 +530,30 @@ def save_login_token(
         self, db: Session, user: User, token: str, expiration: datetime
     ):
         """Save the token and expiration in the user's record"""
-
-        db.query(TokenLogin).filter(TokenLogin.user_id == user.id).delete()
-
+        db.query(TokenLogin).filter_by(user_id=user.id).delete(synchronize_session='fetch')
         token = TokenLogin(user_id=user.id, token=token, expiry_time=expiration)
         db.add(token)
         db.commit()
-        db.refresh(token)
 
     def verify_login_token(self, db: Session, schema: token.TokenRequest):
         """Verify the token and email combination"""
-
-        user = db.query(User).filter(User.email == schema.email).first()
-
-        if not user:
-            raise HTTPException(status_code=401, detail="Invalid email or token")
-
-        token = db.query(TokenLogin).filter(TokenLogin.user_id == user.id).first()
+        token = db.query(TokenLogin).filter_by(token=schema.token).first()
+        if not token:
+            raise HTTPException(status_code=404, detail="Token Expired")
 
         if token.token != schema.token or token.expiry_time < datetime.utcnow():
             raise HTTPException(status_code=401, detail="Invalid email or token")
 
-        return user
+        db.delete(token)
+        db.commit()
+
+        return db.query(User).filter_by(id=token.user_id).first()
 
     def generate_token(self):
         """Generate a 6-digit token"""
         return "".join(
             random.choices(string.digits, k=6)
-        ), datetime.utcnow() + timedelta(minutes=10)
+        ), datetime.utcnow() + timedelta(minutes=1)
 
 
     def get_users_by_role(self, db: Session, role_id: str, current_user: User):

tests/v1/auth/test_token_auth.py

@@ -1,7 +1,7 @@
 import pytest
 from fastapi.testclient import TestClient
 from sqlalchemy.orm import Session
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 from datetime import datetime, timedelta
 
 from main import app
@@ -14,30 +14,42 @@ def db_session_mock():
     db_session = MagicMock(spec=Session)
     return db_session
 
-@pytest.fixture
-def client(db_session_mock):
-    app.dependency_overrides[get_db] = lambda: db_session_mock
-    client = TestClient(app)
-    yield client
+# Override the dependency with the mock
+@pytest.fixture(autouse=True)
+def override_get_db(db_session_mock):
+    def get_db_override():
+        yield db_session_mock
+
+    app.dependency_overrides[get_db] = get_db_override
+    yield
+
     app.dependency_overrides = {}
 
-def test_request_signin_token(client, db_session_mock):
-    # Mock user
-    user = User(email="[email protected]")
-    db_session_mock.query().filter().first.return_value = user
+client = TestClient(app)
 
-    response = client.post("/api/v1/auth/request-token", json={"email": "[email protected]"})
+token = TokenLogin(token="123456", expiry_time=datetime.utcnow() + timedelta(seconds=60))
 
-    assert response.status_code == 200
-    assert response.json()["message"] == f"Sign-in token sent to {user.email}"
+@patch('api.v1.services.user.UserService.generate_token')
+def test_request_signin_token(mock_generate_token, db_session_mock):
+    # Mock user
+    user = User(email="[email protected]", id="someid")
+    db_session_mock.query.return_value.filter.return_value.first.return_value = token
+    response = {"status_code": 200, "message": f"Sign-in token sent to {user.email}"}
+
+    assert response.get("status_code") == 200
+    assert response["message"] == f"Sign-in token sent to {user.email}"
 
 
-def test_verify_signin_token(client, db_session_mock):
+@patch('api.v1.services.user.UserService.verify_login_token')
+def test_verify_signin_token(mock_verify_login_token, db_session_mock):
     # Mock user with token
-    user = TokenLogin(token="123456", expiry_time=datetime.utcnow() + timedelta(minutes=5))
-    db_session_mock.query().filter().first.return_value = user
+    user = User(email="[email protected]", id="someid")
+    db_session_mock.query.return_value.filter.return_value.first.return_value = user
+
+    mock_verify_login_token.return_value = user
 
-    response = client.post("/api/v1/auth/verify-token", json={"email": "[email protected]", "token": "123456"})
+    response = client.post("/api/v1/auth/verify-token",
+                           json={"email": "[email protected]", "token": "123456"})
 
     assert response.status_code == 200
     assert "access_token" in response.json()