Merge pull request #37 from bescka/update-backup-setup

Update backup setup + tests

backend-app/app/api/auth.py

@@ -24,12 +24,15 @@
 
 
 def authenticate_user(email: str, password: str, db: Session = Depends(get_db)):
-    user = get_user_by_email(db, email)
-    if not user:
-        return False
-    if not verify_password(password, user.hashed_password):
+    """Authenticates a user using their email address and password."""
+    try:
+        user = get_user_by_email(db, email)
+        if not verify_password(password, user.hashed_password):
+            return False
+        return user
+    except Exception as e:
+        print(f"Error occurred while authenticating user: {e}")
         return False
-    return user
 
 
 def create_access_token(data: dict, expires_delta: timedelta | None = None):

backend-app/app/models/user.py

@@ -11,9 +11,9 @@ class UserCreate(UserBase):
     password: str
 
 
-# TODO: Needed?
-# class UserInDB(UserBase):
-#     hashed_password: str
+# only for testing
+class UserInDB(UserCreate):
+    hashed_password: str
 
 
 class User(UserBase):

backend-app/tests/unit/conftest.py

@@ -1,17 +1,19 @@
 from io import BytesIO
+from unittest.mock import MagicMock, patch
 
 import pandas as pd
 import pytest
 from fastapi.testclient import TestClient
+from jose import jwt
 from sqlalchemy import create_engine
-from sqlalchemy.orm import Session
+from sqlalchemy.orm import Session, sessionmaker
 
 from app.api.auth import get_current_active_admin, get_current_active_user, get_user_by_email
 from app.main import app
 from app.models import user as api_m
 from app.models.database import User as db_user
 from app.models.file_db import create_file_table_class, update_schema
-from app.models.user import UserCreate
+from app.models.user import UserCreate, UserInDB
 from app.sql_db.crud import create_user, get_db, update_is_active, update_is_admin
 from app.sql_db.database import Base
 from app.sql_db.file_crud import create_update_table, insert_data
@@ -179,3 +181,55 @@ def files_bad_column():
         )
     }
     return files
+
+
+@pytest.fixture
+def mock_db():
+    """Fixture for mocking the database session."""
+    db = MagicMock(spec=Session)
+    return db
+
+
+@pytest.fixture
+def mock_user():
+    """Fixture for mocking a user."""
+    user = UserInDB(email="[email protected]", password="test1", hashed_password="test1fake_hash")
+    return user
+
+
+@pytest.fixture
+def mock_get_user_by_email_success(mock_user):
+    """Fixture for mocking get_user_by_email to return the mock user."""
+    mock_function = MagicMock()
+    mock_function.return_value = mock_user
+    return mock_function
+
+
+@pytest.fixture
+def mock_get_user_by_email_none():
+    """Fixture for mocking get_user_by_email to retuern None (user not found)."""
+
+    mock_function = MagicMock()
+    mock_function.return_value = None
+
+    return mock_function
+
+
+@pytest.fixture
+def valid_token_payload():
+    return {"sub": "[email protected]"}
+
+
+@pytest.fixture
+def valid_token(valid_token_payload):
+    return jwt.encode(
+        valid_token_payload,
+        "SECRET",
+        algorithm="HS256",  # TODO: secret and "HS256" should be configurable
+    )  # TODO: secret and "HS256" should be configurable
+
+
+@pytest.fixture
+def mock_jwt_decode(valid_token_payload):
+    """Mock jwt.decode to return a valid payload."""
+    return MagicMock(return_value=valid_token_payload)

backend-app/tests/unit/test_auth.py

@@ -0,0 +1,197 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from fastapi.exceptions import HTTPException
+from jose import JWTError, jwt
+
+from app.api.auth import authenticate_user, create_access_token, get_current_user
+
+
+def test_authenticate_user_success(mock_db, mock_get_user_by_email_success, mock_user, monkeypatch):
+    """Test successful user authentication."""
+
+    def mock_verify_password(password, hashed_password):
+        return True
+
+    # Monkeypatching external dependencies
+    monkeypatch.setattr("app.api.auth.get_user_by_email", mock_get_user_by_email_success)
+    monkeypatch.setattr("app.api.auth.verify_password", mock_verify_password)
+
+    # Call the function
+    result = authenticate_user("[email protected]", "test1", db=mock_db)
+    print(result)
+
+    # Assert that authentication was successful and returns the user
+    assert result == mock_user
+
+
+def test_authenticate_user_wrong_password(mock_db, mock_get_user_by_email_success, monkeypatch):
+    """Test user authentication with wrong password."""
+
+    def mock_verify_password(password, hashed_password):
+        return False
+
+    # Monkeypatching external dependencies
+    monkeypatch.setattr("app.api.auth.get_user_by_email", mock_get_user_by_email_success)
+    monkeypatch.setattr("app.api.auth.verify_password", mock_verify_password)
+
+    # Call the function
+    result = authenticate_user("[email protected]", "wrong_password", db=mock_db)
+
+    # Assert that authentication fails
+    assert result is False
+
+
+def test_authenticate_user_no_user_found(mock_db, mock_get_user_by_email_none, monkeypatch):
+    """Test user authentication when no user is found."""
+
+    # Monkeypatching external dependencies
+    monkeypatch.setattr("app.api.auth.get_user_by_email", mock_get_user_by_email_none)
+
+    # Call the function
+    result = authenticate_user("[email protected]", "password", db=mock_db)
+
+    # Assert that authentication fails
+    assert result is False
+
+
+def test_create_access_token(mock_user, monkeypatch):
+    TEST_SECRET_KEY = "SECRET"
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    expires_delta = timedelta(minutes=1)
+    access_token = create_access_token(data={"sub": mock_user.email}, expires_delta=expires_delta)
+    decoded_token = jwt.decode(access_token, TEST_SECRET_KEY, algorithms=["HS256"])
+
+    # Assert the sub claim is correct
+    assert decoded_token["sub"] == mock_user.email
+
+    # Assert the expiration time is correct and within a reasonable range
+    exp_time = datetime.fromtimestamp(decoded_token["exp"], tz=timezone.utc)
+    now_time = datetime.now(timezone.utc)
+
+    # Token should expire in approximately the given expiration delta
+    assert now_time < exp_time <= (now_time + expires_delta + timedelta(seconds=2))
+
+
+def test_create_access_token_default_expiration(mock_user, monkeypatch):
+    TEST_SECRET_KEY = "SECRET"
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Call create_access_token without specifying expires_delta
+    access_token = create_access_token(data={"sub": mock_user.email})
+
+    decoded_token = jwt.decode(access_token, TEST_SECRET_KEY, algorithms=["HS256"])
+
+    # Assert the sub claim is correct
+    assert decoded_token["sub"] == mock_user.email
+
+    # Assert the default expiration is 15 minutes
+    exp_time = datetime.fromtimestamp(decoded_token["exp"], tz=timezone.utc)
+    now_time = datetime.now(timezone.utc)
+    assert now_time < exp_time <= (now_time + timedelta(minutes=15) + timedelta(seconds=2))
+
+
+def test_create_access_token_expired_token(mock_user, monkeypatch):
+    TEST_SECRET_KEY = "SECRET"
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Create an expired token by setting expires_delta to -1 minute
+    access_token = create_access_token(
+        data={"sub": mock_user.email}, expires_delta=timedelta(minutes=-1)
+    )
+    with pytest.raises(jwt.ExpiredSignatureError):
+        jwt.decode(access_token, TEST_SECRET_KEY, algorithms=["HS256"])
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_valid_token(
+    mock_jwt_decode,
+    valid_token,
+    mock_user,
+    mock_db,
+    mock_get_user_by_email_success,
+    monkeypatch,
+):
+    TEST_SECRET_KEY = "SECRET"
+
+    # Mock jwt.decode and get_user_by_email using monkeypatch
+    monkeypatch.setattr("app.api.auth.jwt.decode", mock_jwt_decode)
+    monkeypatch.setattr("app.api.auth.get_user_by_email", mock_get_user_by_email_success)
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Call the function
+    user = await get_current_user(token=valid_token, db=mock_db)
+
+    # Assertions
+    assert user.email == mock_user.email
+    mock_jwt_decode.assert_called_once_with(valid_token, TEST_SECRET_KEY, algorithms=["HS256"])
+    mock_get_user_by_email_success.assert_called_once_with(
+        mock_db, email=mock_jwt_decode.return_value.get("sub")
+    )
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_missing_email(mock_jwt_decode, valid_token, mock_db, monkeypatch):
+    TEST_SECRET_KEY = "SECRET"
+
+    # Mock jwt.decode to return a payload without "sub" (missing email)
+    mock_jwt_decode.return_value = {}
+
+    monkeypatch.setattr("app.api.auth.jwt.decode", mock_jwt_decode)
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Expect the function to raise an HTTPException when email is missing
+    with pytest.raises(HTTPException) as exc_info:
+        await get_current_user(token=valid_token, db=mock_db)
+
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail == "Could not validate credentials"
+    mock_jwt_decode.assert_called_once_with(valid_token, TEST_SECRET_KEY, algorithms=["HS256"])
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_jwt_error(mock_jwt_decode, valid_token, mock_db, monkeypatch):
+    TEST_SECRET_KEY = "SECRET"
+
+    # Mock jwt.decode to raise JWTError
+    mock_jwt_decode.side_effect = JWTError
+
+    monkeypatch.setattr("app.api.auth.jwt.decode", mock_jwt_decode)
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Expect the function to raise an HTTPException when JWTError occurs
+    with pytest.raises(HTTPException) as exc_info:
+        await get_current_user(token=valid_token, db=mock_db)
+
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail == "Could not validate credentials"
+    mock_jwt_decode.assert_called_once_with(valid_token, TEST_SECRET_KEY, algorithms=["HS256"])
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_user_not_found(
+    mock_jwt_decode,
+    valid_token_payload,
+    valid_token,
+    mock_db,
+    mock_get_user_by_email_none,
+    monkeypatch,
+):
+    TEST_SECRET_KEY = "SECRET"
+
+    # Mock jwt.decode to return a valid payload with email
+    mock_jwt_decode.return_value = valid_token_payload
+
+    monkeypatch.setattr("app.api.auth.jwt.decode", mock_jwt_decode)
+    monkeypatch.setattr("app.api.auth.get_user_by_email", mock_get_user_by_email_none)
+    monkeypatch.setattr("app.api.auth.SECRET_KEY", TEST_SECRET_KEY)
+
+    # Expect the function to raise an HTTPException when user is not found
+    with pytest.raises(HTTPException) as exc_info:
+        await get_current_user(token=valid_token, db=mock_db)
+
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail == "Could not validate credentials"
+    mock_jwt_decode.assert_called_once_with(valid_token, TEST_SECRET_KEY, algorithms=["HS256"])
+    mock_get_user_by_email_none.assert_called_once_with(mock_db, email=valid_token_payload["sub"])

backend-app/tests/unit/test_crud.py

@@ -1,12 +1,12 @@
 import pytest
-from unittest.mock import MagicMock
 from sqlalchemy import create_engine
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import sessionmaker
-from app.sql_db import crud
+
 from app.models.database import User as db_user
+from app.models.user import UserCreate
+from app.sql_db import crud
 from app.sql_db.database import Base
-from app.models.user import UserCreate, User
 
 
 @pytest.fixture(scope="session")

backend-app/tests/unit/test_filedb_crud.py

@@ -1,7 +1,5 @@
 import logging
 
-import pytest
-
 from app.sql_db import file_crud
 
 

backend-app/tests/unit/test_users.py

@@ -1,5 +1,3 @@
-from fastapi.testclient import TestClient
-from app.api import users
 from app.sql_db import crud