refactor: use shared ProfileUpdate validation in admin routes

This commit is contained in:
Johan Lundberg 2026-03-13 20:40:05 +01:00
parent 5fd63d61ff
commit 7bfea306ab
No known key found for this signature in database
GPG key ID: A6C152738D03C7D1
2 changed files with 56 additions and 16 deletions

View file

@ -1,11 +1,12 @@
from base64 import urlsafe_b64decode from base64 import urlsafe_b64decode
from urllib.parse import urlparse
from fastapi import APIRouter, Form, Request, Response from fastapi import APIRouter, Form, Request, Response
from fastapi.responses import HTMLResponse, RedirectResponse from fastapi.responses import HTMLResponse, RedirectResponse
from pydantic import ValidationError
from porchlight.dependencies import get_session_user from porchlight.dependencies import get_session_user
from porchlight.models import User from porchlight.models import User
from porchlight.validation import ProfileUpdate
router = APIRouter(prefix="/admin", tags=["admin"]) router = APIRouter(prefix="/admin", tags=["admin"])
@ -143,12 +144,35 @@ async def update_user_profile(
return HTMLResponse("Forbidden", status_code=403) return HTMLResponse("Forbidden", status_code=403)
# Validate # Validate
if email and "@" not in email: try:
return HTMLResponse('<div role="alert">Invalid email address</div>') profile = ProfileUpdate(
if picture: given_name=given_name,
parsed = urlparse(picture) family_name=family_name,
if parsed.scheme not in ("http", "https") or not parsed.netloc: preferred_username=preferred_username,
return HTMLResponse('<div role="alert">Picture URL must be a valid HTTP or HTTPS URL</div>') email=email,
phone_number=phone_number,
picture=picture,
locale=locale,
)
except ValidationError as exc:
error = exc.errors()[0]
field = error["loc"][-1] if error["loc"] else "input"
msg = error["msg"]
labels = {
"given_name": "Given name",
"family_name": "Family name",
"preferred_username": "Display name",
"email": "Email",
"phone_number": "Phone number",
"picture": "Picture URL",
"locale": "Locale",
}
label = labels.get(str(field), str(field))
if error["type"] == "value_error":
display_msg = msg.removeprefix("Value error, ")
else:
display_msg = f"{label}: {msg}"
return HTMLResponse(f'<div role="alert">{display_msg}</div>')
user_repo = request.app.state.user_repo user_repo = request.app.state.user_repo
user = await user_repo.get_by_userid(userid) user = await user_repo.get_by_userid(userid)
@ -157,13 +181,13 @@ async def update_user_profile(
updated = user.model_copy( updated = user.model_copy(
update={ update={
"given_name": given_name or None, "given_name": profile.given_name or None,
"family_name": family_name or None, "family_name": profile.family_name or None,
"preferred_username": preferred_username or None, "preferred_username": profile.preferred_username or None,
"email": email or None, "email": profile.email,
"phone_number": phone_number or None, "phone_number": profile.phone_number,
"picture": picture or None, "picture": profile.picture,
"locale": locale or None, "locale": profile.locale or None,
} }
) )
await user_repo.update(updated) await user_repo.update(updated)

View file

@ -153,7 +153,7 @@ async def test_update_profile(client: AsyncClient) -> None:
"family_name": "Smith", "family_name": "Smith",
"preferred_username": "bobby", "preferred_username": "bobby",
"email": "bob@example.com", "email": "bob@example.com",
"phone_number": "+1234567890", "phone_number": "+46701234567",
"picture": "https://example.com/bob.jpg", "picture": "https://example.com/bob.jpg",
"locale": "en-US", "locale": "en-US",
}, },
@ -183,7 +183,23 @@ async def test_update_profile_invalid_email(client: AsyncClient) -> None:
headers={"X-CSRF-Token": token}, headers={"X-CSRF-Token": token},
) )
assert response.status_code == 200 assert response.status_code == 200
assert "Invalid email" in response.text assert "alert" in response.text
assert "email" in response.text.lower()
@pytest.mark.asyncio
async def test_update_profile_invalid_phone(client: AsyncClient) -> None:
await _login(client)
target = await _create_target_user(client, username="bob")
token = await get_csrf_token(client)
response = await client.post(
f"/admin/users/{target.userid}/profile",
data={"phone_number": "not-a-phone"},
headers={"X-CSRF-Token": token},
)
assert response.status_code == 200
assert "alert" in response.text
assert "phone" in response.text.lower()
@pytest.mark.asyncio @pytest.mark.asyncio