Merge pull request #197 from amosproj/feature/11-login/out-test-user

added frontend mask for login and handling for failed login

Backend/.env_example

@@ -1,3 +1,4 @@
 TEST_STAGE=local
 PASSWORD=SomePassword
+SECRET_KEY=Some_random_32byte_key
 MONGODB_URL=mongodb://localhost:27017/

Backend/app/api/v1/auth_api.py

@@ -0,0 +1,70 @@
+import os
+from dotenv import load_dotenv
+from fastapi import APIRouter, Depends, HTTPException, Response
+from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
+from app.repository.user_repository import UserRepository
+from app.dependency.repository import get_user_repository
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+
+router = APIRouter()
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+load_dotenv()
+SECRET_KEY = os.getenv("SECRET_KEY")
+ALGORITHM = "HS256"
+
+
+def create_access_token(data: dict, expires_delta: timedelta = None):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+@router.post("/token")
+async def login_for_access_token(
+    response: Response,
+    form_data: OAuth2PasswordRequestForm = Depends(),
+    user_repo: UserRepository = Depends(get_user_repository),
+):
+    # check userdata
+    is_authenticated = user_repo.authenticate_user(
+        email=form_data.username, password=form_data.password
+    )
+    if not is_authenticated:
+        raise HTTPException(
+            status_code=402,
+            detail="Incorrect email or password",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    # read user data
+    user_data = user_repo.read_users_by_email(email=form_data.username)
+    if not user_data:
+        raise HTTPException(status_code=404, detail="User not found")
+
+    # take first element, because email_adresse should be unique
+    user = user_data[0]
+
+    access_token_expires = timedelta(minutes=60)
+    access_token = create_access_token(
+        data={"sub": user["email_address"]}, expires_delta=access_token_expires
+    )
+    response.set_cookie(key="access_token", value=access_token, httponly=True)
+    return {"access_token": access_token, "token_type": "bearer", "success": True}
+
+
+@router.get("/verify-token")
+async def verify_token(token: str = Depends(oauth2_scheme)):
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        email = payload.get("sub")
+        if email is None:
+            raise HTTPException(status_code=401, detail="Invalid token")
+        return {"email": email}
+    except JWTError:
+        raise HTTPException(status_code=401, detail="Invalid token")

Backend/app/main.py

@@ -2,6 +2,8 @@
 from fastapi.middleware.cors import CORSMiddleware
 
 from app.api.v1 import ticket_api
+from app.api.v1 import auth_api
+
 from app.dependency.collection import (
     get_user_collection,
     get_service_collection,
@@ -16,6 +18,7 @@
     get_category_repository,
     get_location_repository,
 )
+
 from app.repository.user_repository import UserRepository
 from app.repository.service_repository import ServiceRepository
 from app.repository.location_repository import LocationRepository
@@ -47,6 +50,9 @@
 # Include the router from the text_endpoint module
 app.include_router(ticket_api.router, prefix="/api/v1")
 
+# Include the router from the auth_endpoint module
+app.include_router(auth_api.router, prefix="/api/v1")
+
 
 @app.on_event("startup")
 async def startup_event():

Backend/app/repository/user_repository.py

@@ -32,3 +32,8 @@ def read_users_by_email(self, email: str) -> list[UserEntity]:
         logger.info(f"Reading user(s) with email {email} from the database...")
         users = list(self.collection.find({"email_address": email}))
         return users if users else []
+
+    def authenticate_user(self, email: str, password: str) -> bool:
+        logger.info(f"Authenticating user with email {email}...")
+        user = self.collection.find_one({"email_address": email, "password": password})
+        return user is not None

Backend/requirements.txt

@@ -13,4 +13,5 @@ fastapi-utils==0.2.1
 requests==2.31.0
 python-multipart==0.0.6
 beautifulsoup4==4.12.2
-openai==1.7.2
+openai==1.7.2
+python-jose==3.3.0

Backend/test/api/v1/auth_api_test.py

@@ -0,0 +1,73 @@
+import os
+from datetime import datetime, timedelta
+from dotenv import load_dotenv
+from jose import JWTError, jwt
+import pytest
+from fastapi.testclient import TestClient
+from unittest.mock import Mock
+from test.config.pytest import SKIP_TEST
+from app.main import app
+from app.dependency.repository import get_user_repository
+
+
+load_dotenv()
+SECRET_KEY = os.getenv("SECRET_KEY")
+
+
+# Mock dependencies
+@pytest.fixture
+def mock_user_repository():
+    user_repo = Mock()
+    user_repo.authenticate_user.return_value = True
+    user_repo.read_users_by_email.return_value = [{"email_address": "[email protected]"}]
+    return user_repo
+
+
+@pytest.fixture
+def client(mock_user_repository):
+    app.dependency_overrides[get_user_repository] = lambda: mock_user_repository
+    return TestClient(app)
+
+
+def generate_valid_token():
+    ALGORITHM = "HS256"
+    data = {"sub": "[email protected]"}
+    expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode = data.copy()
+    to_encode.update({"exp": expire})
+    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+
+
+@pytest.mark.skipif(condition=SKIP_TEST, reason=".env on git")
+class TestAPI:
+    def test_login_for_access_token_success(self, client):
+        response = client.post(
+            "/api/v1/token",
+            data={"username": "[email protected]", "password": "password"},
+        )
+        assert response.status_code == 200
+        assert "access_token" in response.json()
+
+    def test_login_for_access_token_failure(self, client, mock_user_repository):
+        # Configure the mock to return False for authentication
+        mock_user_repository.authenticate_user.return_value = False
+        response = client.post(
+            "/api/v1/token",
+            data={"username": "[email protected]", "password": "wrongpassword"},
+        )
+        assert response.status_code == 402
+
+    def test_verify_token_success(self, client):
+        valid_token = generate_valid_token()
+        response = client.get(
+            "/api/v1/verify-token", headers={"Authorization": f"Bearer {valid_token}"}
+        )
+        assert response.status_code == 200
+
+    def test_verify_token_failure(self, client):
+        # Use an altered valid token for this test
+        valid_token = generate_valid_token() + "invalid_part"
+        response = client.get(
+            "/api/v1/verify-token", headers={"Authorization": f"Bearer {valid_token}"}
+        )
+        assert response.status_code == 401

Backend/test/repository/ticket_repository_test.py

@@ -33,6 +33,7 @@ def setUp(self):
             description="",
             priority=Prio.low,
             attachments=[],
+            requestType="",
         )
 
     def test_create_ticket(self):
@@ -98,6 +99,7 @@ def setUp(self):
             "description": "",
             "priority": Prio.low,
             "attachments": [],
+            "requestType": "",
         }
 
     def test_crud(self):