feat: implement user profile management and oauth/traefik integration
User Profile (US-004): - Add authenticated profile endpoints (GET/PUT /users/me) - Add avatar upload with file validation (PNG/JPEG, max 2MB) - Create frontend profile page with edit form and avatar upload - Update app shell to link to profile page OAuth/Traefik Integration: - Externalize all Authentik URLs to environment variables - Add domain configuration (API_DOMAIN, WEB_DOMAIN, AUTHENTIK_DOMAIN) - Create docker-compose.traefik.yml for reverse proxy deployment - Update OAuth redirect/callback URLs to use configured domains - Add VITE_APP_URL for frontend public URL configuration Quality gates: pytest (50 passed), ruff, mypy, npm test (12 passed), typecheck, lint, build
This commit is contained in:
Fusion
@@ -0,0 +1,225 @@
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import asyncio
|
||||
import io
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
from src.auth.jwt_service import mint_access_token
|
||||
from src.config import Settings, build_database_url
|
||||
from src.models import Base
|
||||
from src.models.user import User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def configure_local_database(monkeypatch) -> None:
|
||||
local_url = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
monkeypatch.setenv("DATABASE_URL", local_url)
|
||||
|
||||
|
||||
def _prepare_users_test_db() -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
await connection.execute(text("TRUNCATE TABLE refresh_tokens, users RESTART IDENTITY CASCADE"))
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _load_app():
|
||||
import importlib
|
||||
import src.database as database_module
|
||||
import src.api.users as users_module
|
||||
import src.main as main_module
|
||||
|
||||
importlib.reload(database_module)
|
||||
importlib.reload(users_module)
|
||||
importlib.reload(main_module)
|
||||
return main_module.app
|
||||
|
||||
|
||||
def _insert_test_user(user_id: str) -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
user = User(
|
||||
id=uuid.UUID(user_id),
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
authentik_id="test-user",
|
||||
avatar_url=None,
|
||||
)
|
||||
await session.merge(user)
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _create_auth_cookie(user_id: str) -> str:
|
||||
settings = Settings()
|
||||
return mint_access_token(
|
||||
settings=settings,
|
||||
subject=user_id,
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
expires_at=datetime.now(UTC) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
|
||||
def test_get_profile_returns_401_without_cookie() -> None:
|
||||
_prepare_users_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get("/users/me")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
def test_get_profile_returns_user_data() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.get("/users/me")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["email"] == "test@headquarter.local"
|
||||
assert data["name"] == "Test User"
|
||||
assert data["avatar_url"] is None
|
||||
|
||||
|
||||
def test_update_profile_changes_name_and_email() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"name": "Updated Name", "email": "updated@headquarter.local"})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Updated Name"
|
||||
assert data["email"] == "updated@headquarter.local"
|
||||
|
||||
|
||||
def test_update_profile_rejects_empty_name() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"name": " "})
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_update_profile_rejects_invalid_email() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"email": "not-an-email"})
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_avatar_updates_avatar_url() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
# Create a simple 1x1 PNG
|
||||
png_data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("test.png", io.BytesIO(png_data), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["avatar_url"] is not None
|
||||
assert data["avatar_url"].startswith("/uploads/avatars/")
|
||||
|
||||
|
||||
def test_upload_avatar_rejects_invalid_file_type() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("test.txt", io.BytesIO(b"not an image"), "text/plain")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
def test_upload_avatar_rejects_oversized_file() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
large_png = b"\x89PNG\r\n\x1a\n" + b"\x00" * (3 * 1024 * 1024) # 3MB
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("large.png", io.BytesIO(large_png), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
Reference in New Issue
Block a user