Merge pull request #607 from christianlouis/copilot/add-user-settings-management
fix(tests): update whoami_handler call sites to pass required db argument
This commit is contained in:
Christian Krakau-Louis
@@ -116,9 +116,13 @@ class TestProtectedAPIEndpoints:
|
||||
}
|
||||
}
|
||||
|
||||
# Mock DB: no UserProfile found (no custom avatar)
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
import asyncio
|
||||
|
||||
result = asyncio.run(whoami_handler(mock_request))
|
||||
result = asyncio.run(whoami_handler(mock_request, mock_db))
|
||||
|
||||
assert result["id"] == "test123"
|
||||
assert result["email"] == "test@example.com"
|
||||
@@ -135,9 +139,10 @@ class TestProtectedAPIEndpoints:
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
asyncio.run(whoami_handler(mock_request))
|
||||
asyncio.run(whoami_handler(mock_request, mock_db))
|
||||
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "Not logged in" in exc_info.value.detail
|
||||
|
||||
@@ -0,0 +1,509 @@
|
||||
"""Tests for app/api/profile.py — user self-service profile API.
|
||||
|
||||
Unit tests call handler functions directly with mock request objects.
|
||||
Integration tests use a dedicated TestClient with DB override.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.database import Base, get_db
|
||||
from app.models import LocalUser, UserProfile
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Test data constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Minimal valid 1×1 PNG image (base64-encoded) used across avatar upload tests
|
||||
_MINIMAL_VALID_PNG_BASE64 = (
|
||||
b"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mP8/Z+hHgAHggJ/PchI6QAAAABJRU5ErkJggg=="
|
||||
)
|
||||
_MINIMAL_VALID_PNG_BYTES = base64.b64decode(_MINIMAL_VALID_PNG_BASE64)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def prof_engine():
|
||||
"""In-memory SQLite engine scoped to one test."""
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield engine
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def prof_session(prof_engine):
|
||||
"""DB session for one profile test."""
|
||||
Session = sessionmaker(bind=prof_engine)
|
||||
session = Session()
|
||||
yield session
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def prof_client(prof_engine):
|
||||
"""TestClient with the in-memory DB injected."""
|
||||
from app.main import app
|
||||
|
||||
def override_db():
|
||||
Session = sessionmaker(bind=prof_engine)
|
||||
session = Session()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
app.dependency_overrides[get_db] = override_db
|
||||
with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as c:
|
||||
yield c
|
||||
app.dependency_overrides.pop(get_db, None)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — helper functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGravatarUrl:
|
||||
"""Tests for the _gravatar_url helper."""
|
||||
|
||||
def test_returns_gravatar_for_valid_email(self):
|
||||
from app.api.profile import _gravatar_url
|
||||
|
||||
url = _gravatar_url("Test@Example.COM")
|
||||
assert url.startswith("https://www.gravatar.com/avatar/")
|
||||
assert url.endswith("?d=identicon")
|
||||
|
||||
def test_fallback_for_none_email(self):
|
||||
from app.api.profile import _gravatar_url
|
||||
|
||||
url = _gravatar_url(None)
|
||||
assert url.startswith("https://www.gravatar.com/avatar/")
|
||||
assert url.endswith("?d=identicon")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetUserId:
|
||||
"""Tests for the _get_user_id helper."""
|
||||
|
||||
def test_extracts_sub(self):
|
||||
from app.api.profile import _get_user_id
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"sub": "sub-123", "email": "a@b.com"}}
|
||||
assert _get_user_id(req) == "sub-123"
|
||||
|
||||
def test_extracts_preferred_username_fallback(self):
|
||||
from app.api.profile import _get_user_id
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "alice", "email": "a@b.com"}}
|
||||
assert _get_user_id(req) == "alice"
|
||||
|
||||
def test_extracts_email_fallback(self):
|
||||
from app.api.profile import _get_user_id
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"email": "a@b.com"}}
|
||||
assert _get_user_id(req) == "a@b.com"
|
||||
|
||||
def test_raises_401_when_no_session_user(self):
|
||||
from app.api.profile import _get_user_id
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {}
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_get_user_id(req)
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
def test_raises_401_when_no_identifier(self):
|
||||
from app.api.profile import _get_user_id
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"name": "Someone"}}
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
_get_user_id(req)
|
||||
assert exc.value.status_code == 401
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — GET /api/profile handler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestGetProfileHandler:
|
||||
"""Unit tests for the get_profile endpoint handler."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_profile_from_db(self, prof_session):
|
||||
"""get_profile reads from DB and returns correct data."""
|
||||
from app.api.profile import get_profile
|
||||
|
||||
profile = UserProfile(
|
||||
user_id="alice",
|
||||
display_name="Alice",
|
||||
preferred_language="fr",
|
||||
preferred_theme="dark",
|
||||
)
|
||||
prof_session.add(profile)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "alice", "email": "alice@example.com"}}
|
||||
|
||||
result = await get_profile(req, prof_session)
|
||||
assert result.user_id == "alice"
|
||||
assert result.display_name == "Alice"
|
||||
assert result.preferred_language == "fr"
|
||||
assert result.preferred_theme == "dark"
|
||||
assert result.avatar_url.startswith("https://www.gravatar.com/avatar/")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_custom_avatar_when_stored(self, prof_session):
|
||||
"""get_profile returns the data: URI when avatar_data is set."""
|
||||
from app.api.profile import get_profile
|
||||
|
||||
profile = UserProfile(
|
||||
user_id="bob",
|
||||
avatar_data="data:image/png;base64,abc",
|
||||
)
|
||||
prof_session.add(profile)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "bob", "email": "bob@example.com"}}
|
||||
|
||||
result = await get_profile(req, prof_session)
|
||||
assert result.avatar_url == "data:image/png;base64,abc"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_creates_profile_if_missing(self, prof_session):
|
||||
"""get_profile creates a stub profile row when none exists."""
|
||||
from app.api.profile import get_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "newbie", "email": "newbie@example.com"}}
|
||||
|
||||
result = await get_profile(req, prof_session)
|
||||
assert result.user_id == "newbie"
|
||||
row = prof_session.query(UserProfile).filter_by(user_id="newbie").first()
|
||||
assert row is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — PATCH /api/profile handler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUpdateProfileHandler:
|
||||
"""Unit tests for the update_profile endpoint handler."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_updates_display_name(self, prof_session):
|
||||
"""update_profile updates display_name."""
|
||||
from app.api.profile import ProfileUpdateRequest, update_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "carol", "email": "carol@example.com"}}
|
||||
|
||||
body = ProfileUpdateRequest(display_name="Carol Smith")
|
||||
result = await update_profile(body, req, prof_session)
|
||||
assert result.display_name == "Carol Smith"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_updates_language(self, prof_session):
|
||||
"""update_profile updates preferred_language."""
|
||||
from app.api.profile import ProfileUpdateRequest, update_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "dave", "email": "dave@example.com"}}
|
||||
|
||||
body = ProfileUpdateRequest(preferred_language="de")
|
||||
result = await update_profile(body, req, prof_session)
|
||||
assert result.preferred_language == "de"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_updates_theme(self, prof_session):
|
||||
"""update_profile updates preferred_theme."""
|
||||
from app.api.profile import ProfileUpdateRequest, update_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "eve", "email": "eve@example.com"}}
|
||||
|
||||
body = ProfileUpdateRequest(preferred_theme="light")
|
||||
result = await update_profile(body, req, prof_session)
|
||||
assert result.preferred_theme == "light"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_invalid_language(self, prof_session):
|
||||
"""update_profile raises 422 for unsupported language code."""
|
||||
from app.api.profile import ProfileUpdateRequest, update_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "frank", "email": "frank@example.com"}}
|
||||
|
||||
body = ProfileUpdateRequest(preferred_language="xx")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await update_profile(body, req, prof_session)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_invalid_theme(self, prof_session):
|
||||
"""update_profile raises 422 for invalid theme value."""
|
||||
from app.api.profile import ProfileUpdateRequest, update_profile
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "grace", "email": "grace@example.com"}}
|
||||
|
||||
body = ProfileUpdateRequest(preferred_theme="rainbow")
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await update_profile(body, req, prof_session)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — POST /api/profile/avatar handler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestUploadAvatarHandler:
|
||||
"""Unit tests for the upload_avatar endpoint handler."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stores_base64_data_url(self, prof_session):
|
||||
"""upload_avatar stores the image as a data: URI."""
|
||||
from app.api.profile import upload_avatar
|
||||
|
||||
upload = MagicMock()
|
||||
upload.content_type = "image/png"
|
||||
upload.size = len(_MINIMAL_VALID_PNG_BYTES)
|
||||
upload.read = AsyncMock(return_value=_MINIMAL_VALID_PNG_BYTES)
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "avataruser", "email": "av@example.com"}}
|
||||
|
||||
result = await upload_avatar(req, prof_session, upload)
|
||||
assert result["avatar_url"].startswith("data:image/png;base64,")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_unsupported_mime(self, prof_session):
|
||||
"""upload_avatar raises 415 for non-image content types."""
|
||||
from app.api.profile import upload_avatar
|
||||
|
||||
upload = MagicMock()
|
||||
upload.content_type = "application/pdf"
|
||||
upload.size = 4
|
||||
upload.read = AsyncMock(return_value=b"%PDF")
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "pdfuser", "email": "pdf@example.com"}}
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await upload_avatar(req, prof_session, upload)
|
||||
assert exc.value.status_code == 415
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_oversized_image(self, prof_session):
|
||||
"""upload_avatar raises 413 when image exceeds 2 MB."""
|
||||
from app.api.profile import upload_avatar
|
||||
|
||||
big_size = 2 * 1024 * 1024 + 1
|
||||
|
||||
upload = MagicMock()
|
||||
upload.content_type = "image/png"
|
||||
upload.size = big_size # triggers early size check
|
||||
upload.read = AsyncMock(return_value=b"x" * big_size)
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "biguser", "email": "big@example.com"}}
|
||||
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await upload_avatar(req, prof_session, upload)
|
||||
assert exc.value.status_code == 413
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — DELETE /api/profile/avatar handler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestDeleteAvatarHandler:
|
||||
"""Unit tests for the delete_avatar endpoint handler."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clears_avatar_data(self, prof_session):
|
||||
"""delete_avatar removes avatar_data and returns a Gravatar URL."""
|
||||
from app.api.profile import delete_avatar
|
||||
|
||||
profile = UserProfile(
|
||||
user_id="delavatar",
|
||||
avatar_data="data:image/png;base64,abc",
|
||||
)
|
||||
prof_session.add(profile)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "delavatar", "email": "del@example.com"}}
|
||||
|
||||
result = await delete_avatar(req, prof_session)
|
||||
assert result["avatar_url"].startswith("https://www.gravatar.com/avatar/")
|
||||
|
||||
row = prof_session.query(UserProfile).filter_by(user_id="delavatar").first()
|
||||
assert row.avatar_data is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Unit tests — POST /api/profile/change-password handler
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestChangePasswordHandler:
|
||||
"""Unit tests for the change_password endpoint handler."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_non_local_user(self, prof_session):
|
||||
"""change_password raises 403 for OAuth-only accounts."""
|
||||
from app.api.profile import ChangePasswordRequest, change_password
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "oauthonly", "email": "oauth@example.com"}}
|
||||
|
||||
body = ChangePasswordRequest(
|
||||
current_password="old",
|
||||
new_password="newpassword1",
|
||||
new_password_confirm="newpassword1",
|
||||
)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await change_password(body, req, prof_session)
|
||||
assert exc.value.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_wrong_current_password(self, prof_session):
|
||||
"""change_password raises 403 when current password is wrong."""
|
||||
from app.api.profile import ChangePasswordRequest, change_password
|
||||
from app.utils.local_auth import hash_password
|
||||
|
||||
local_user = LocalUser(
|
||||
email="local@example.com",
|
||||
username="localwrong",
|
||||
hashed_password=hash_password("correctpassword"),
|
||||
is_active=True,
|
||||
)
|
||||
prof_session.add(local_user)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "localwrong", "email": "local@example.com"}}
|
||||
|
||||
body = ChangePasswordRequest(
|
||||
current_password="wrongpassword",
|
||||
new_password="newpassword1",
|
||||
new_password_confirm="newpassword1",
|
||||
)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await change_password(body, req, prof_session)
|
||||
assert exc.value.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rejects_password_mismatch(self, prof_session):
|
||||
"""change_password raises 422 when new passwords do not match."""
|
||||
from app.api.profile import ChangePasswordRequest, change_password
|
||||
from app.utils.local_auth import hash_password
|
||||
|
||||
local_user = LocalUser(
|
||||
email="mismatch@example.com",
|
||||
username="mismatchpw",
|
||||
hashed_password=hash_password("currentpw"),
|
||||
is_active=True,
|
||||
)
|
||||
prof_session.add(local_user)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "mismatchpw", "email": "mismatch@example.com"}}
|
||||
|
||||
body = ChangePasswordRequest(
|
||||
current_password="currentpw",
|
||||
new_password="newpassword1",
|
||||
new_password_confirm="differentpassword",
|
||||
)
|
||||
with pytest.raises(HTTPException) as exc:
|
||||
await change_password(body, req, prof_session)
|
||||
assert exc.value.status_code == 422
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_changes_password_successfully(self, prof_session):
|
||||
"""change_password updates hashed_password for correct input."""
|
||||
from app.api.profile import ChangePasswordRequest, change_password
|
||||
from app.utils.local_auth import hash_password, verify_password
|
||||
|
||||
local_user = LocalUser(
|
||||
email="success@example.com",
|
||||
username="successpw",
|
||||
hashed_password=hash_password("oldpassword"),
|
||||
is_active=True,
|
||||
)
|
||||
prof_session.add(local_user)
|
||||
prof_session.commit()
|
||||
|
||||
req = MagicMock()
|
||||
req.session = {"user": {"preferred_username": "successpw", "email": "success@example.com"}}
|
||||
|
||||
body = ChangePasswordRequest(
|
||||
current_password="oldpassword",
|
||||
new_password="newpassword1",
|
||||
new_password_confirm="newpassword1",
|
||||
)
|
||||
result = await change_password(body, req, prof_session)
|
||||
assert "successfully" in result["detail"].lower()
|
||||
|
||||
updated_user = prof_session.query(LocalUser).filter_by(username="successpw").first()
|
||||
assert verify_password("newpassword1", updated_user.hashed_password)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration tests — HTTP endpoint registration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestProfileEndpoints:
|
||||
"""Verify profile endpoints are registered and reachable."""
|
||||
|
||||
def test_get_profile_without_session_returns_401(self, prof_client):
|
||||
"""GET /api/profile returns 401 when no user in session."""
|
||||
response = prof_client.get("/api/profile")
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_patch_profile_without_session_returns_401(self, prof_client):
|
||||
"""PATCH /api/profile returns 401 when no user in session."""
|
||||
response = prof_client.patch("/api/profile", json={"display_name": "Test"})
|
||||
assert response.status_code == 401
|
||||
|
||||
def test_profile_page_accessible(self, prof_client):
|
||||
"""GET /profile page renders successfully (auth disabled in tests)."""
|
||||
response = prof_client.get("/profile", follow_redirects=False)
|
||||
# AUTH_ENABLED=False in tests so no redirect; page should render
|
||||
assert response.status_code in (200, 302)
|
||||
+41
-5
@@ -14,18 +14,38 @@ class TestWhoamiHandler:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_user_with_gravatar(self):
|
||||
"""Test that handler returns user data with gravatar URL."""
|
||||
"""Test that handler returns user data with gravatar URL when no custom avatar."""
|
||||
mock_request = MagicMock()
|
||||
email = "test@example.com"
|
||||
mock_request.session = {"user": {"id": "1", "email": email, "name": "Test"}}
|
||||
|
||||
result = await whoami_handler(mock_request)
|
||||
# Mock DB: no UserProfile found (no custom avatar)
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = None
|
||||
|
||||
result = await whoami_handler(mock_request, mock_db)
|
||||
assert result["id"] == "1"
|
||||
assert result["name"] == "Test"
|
||||
# Should have gravatar URL
|
||||
# Should have gravatar URL since no custom avatar
|
||||
expected_hash = md5(email.encode(), usedforsecurity=False).hexdigest()
|
||||
assert result["picture"] == f"https://www.gravatar.com/avatar/{expected_hash}?d=identicon"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_custom_avatar_when_set(self):
|
||||
"""Test that handler returns custom avatar URL when profile has avatar_data."""
|
||||
mock_request = MagicMock()
|
||||
email = "test@example.com"
|
||||
mock_request.session = {"user": {"id": "1", "email": email, "name": "Test"}}
|
||||
|
||||
# Mock DB: UserProfile with avatar_data
|
||||
mock_profile = MagicMock()
|
||||
mock_profile.avatar_data = "data:image/png;base64,abc123"
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.return_value.filter.return_value.first.return_value = mock_profile
|
||||
|
||||
result = await whoami_handler(mock_request, mock_db)
|
||||
assert result["picture"] == "data:image/png;base64,abc123"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_raises_401_when_no_user(self):
|
||||
"""Test that 401 is raised when no user in session."""
|
||||
@@ -33,9 +53,10 @@ class TestWhoamiHandler:
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {}
|
||||
mock_db = MagicMock()
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await whoami_handler(mock_request)
|
||||
await whoami_handler(mock_request, mock_db)
|
||||
assert exc_info.value.status_code == 401
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -45,11 +66,26 @@ class TestWhoamiHandler:
|
||||
|
||||
mock_request = MagicMock()
|
||||
mock_request.session = {"user": {"id": "1", "name": "Test"}}
|
||||
mock_db = MagicMock()
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await whoami_handler(mock_request)
|
||||
await whoami_handler(mock_request, mock_db)
|
||||
assert exc_info.value.status_code == 400
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_falls_back_to_gravatar_on_db_error(self):
|
||||
"""Test that gravatar is used when DB lookup raises an exception."""
|
||||
mock_request = MagicMock()
|
||||
email = "test@example.com"
|
||||
mock_request.session = {"user": {"id": "1", "email": email, "name": "Test"}}
|
||||
|
||||
mock_db = MagicMock()
|
||||
mock_db.query.side_effect = Exception("DB error")
|
||||
|
||||
result = await whoami_handler(mock_request, mock_db)
|
||||
expected_hash = md5(email.encode(), usedforsecurity=False).hexdigest()
|
||||
assert result["picture"] == f"https://www.gravatar.com/avatar/{expected_hash}?d=identicon"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestWhoamiEndpoints:
|
||||
|
||||
Reference in New Issue
Block a user