feat: complete auth/profile username migration and runtime safeguards

This commit is contained in:
qzl
2026-02-25 10:20:43 +08:00
parent 8bdcb674bb
commit 7d6dda57c1
24 changed files with 720 additions and 166 deletions
+145 -1
View File
@@ -1,14 +1,18 @@
from __future__ import annotations
from typing import Callable
from uuid import UUID
from fastapi import HTTPException
from fastapi.testclient import TestClient
from app import app
from core.auth.models import CurrentUser
from v1.auth.dependencies import get_auth_service
from v1.profile.dependencies import get_current_user
from v1.auth.schemas import (
AuthTokenResponse,
AuthUserByEmailResponse,
AuthUser,
LoginRequest,
RefreshRequest,
@@ -33,6 +37,16 @@ class FakeAuthService(AuthService):
async def logout(self, refresh_token: str | None) -> None:
return None
async def get_user_by_email(self, email: str) -> AuthUserByEmailResponse:
if email == "missing@example.com":
raise HTTPException(status_code=404, detail="User not found")
return AuthUserByEmailResponse(
id="user-1",
email=email,
created_at="2026-02-24T00:00:00Z",
email_confirmed_at=None,
)
def _override_auth_service(service: AuthService) -> Callable[[], AuthService]:
def _get_service() -> AuthService:
@@ -58,7 +72,11 @@ def test_signup_returns_token_response() -> None:
try:
response = client.post(
"/api/v1/auth/signup",
json={"email": "user@example.com", "password": "secret123"},
json={
"username": "demo",
"email": "user@example.com",
"password": "secret123",
},
)
assert response.status_code == 200
body = response.json()
@@ -176,3 +194,129 @@ def test_signup_validation_error_returns_problem_details() -> None:
assert body["detail"] == "Invalid request"
finally:
app.dependency_overrides = {}
def test_signup_missing_username_returns_problem_details() -> None:
user = AuthUser(id="user-1", email="user@example.com")
token_response = AuthTokenResponse(
access_token="access",
refresh_token="refresh",
expires_in=3600,
token_type="bearer",
user=user,
)
app.dependency_overrides[get_auth_service] = _override_auth_service(
FakeAuthService(token_response)
)
client = TestClient(app)
try:
response = client.post(
"/api/v1/auth/signup",
json={"email": "user@example.com", "password": "secret123"},
)
assert response.status_code == 422
assert response.headers["content-type"].startswith("application/problem+json")
body = response.json()
assert body["title"] == "Unprocessable Content"
assert body["status"] == 422
assert body["detail"] == "Invalid request"
finally:
app.dependency_overrides = {}
def test_get_user_by_email_returns_user() -> None:
user = AuthUser(id="user-1", email="user@example.com")
token_response = AuthTokenResponse(
access_token="access",
refresh_token="refresh",
expires_in=3600,
token_type="bearer",
user=user,
)
app.dependency_overrides[get_auth_service] = _override_auth_service(
FakeAuthService(token_response)
)
app.dependency_overrides[get_current_user] = lambda: CurrentUser(
id=UUID("00000000-0000-0000-0000-000000000001"),
email="user@example.com",
)
client = TestClient(app)
try:
response = client.get(
"/api/v1/auth/users/by-email",
params={"email": "user@example.com"},
)
assert response.status_code == 200
body = response.json()
assert body["email"] == "user@example.com"
assert body["id"] == "user-1"
finally:
app.dependency_overrides = {}
def test_get_user_by_email_not_found_returns_problem_details() -> None:
user = AuthUser(id="user-1", email="user@example.com")
token_response = AuthTokenResponse(
access_token="access",
refresh_token="refresh",
expires_in=3600,
token_type="bearer",
user=user,
)
app.dependency_overrides[get_auth_service] = _override_auth_service(
FakeAuthService(token_response)
)
app.dependency_overrides[get_current_user] = lambda: CurrentUser(
id=UUID("00000000-0000-0000-0000-000000000001"),
email="missing@example.com",
)
client = TestClient(app)
try:
response = client.get(
"/api/v1/auth/users/by-email",
params={"email": "missing@example.com"},
)
assert response.status_code == 404
assert response.headers["content-type"].startswith("application/problem+json")
body = response.json()
assert body["title"] == "Not Found"
assert body["status"] == 404
assert body["detail"] == "User not found"
finally:
app.dependency_overrides = {}
def test_get_user_by_email_forbidden_when_querying_other_user() -> None:
user = AuthUser(id="user-1", email="user@example.com")
token_response = AuthTokenResponse(
access_token="access",
refresh_token="refresh",
expires_in=3600,
token_type="bearer",
user=user,
)
app.dependency_overrides[get_auth_service] = _override_auth_service(
FakeAuthService(token_response)
)
app.dependency_overrides[get_current_user] = lambda: CurrentUser(
id=UUID("00000000-0000-0000-0000-000000000001"),
email="self@example.com",
)
client = TestClient(app)
try:
response = client.get(
"/api/v1/auth/users/by-email",
params={"email": "target@example.com"},
)
assert response.status_code == 403
assert response.headers["content-type"].startswith("application/problem+json")
body = response.json()
assert body["title"] == "Forbidden"
assert body["status"] == 403
assert body["detail"] == "Forbidden"
finally:
app.dependency_overrides = {}
@@ -29,11 +29,10 @@ class FakeProfileService:
raise HTTPException(status_code=404, detail="Profile not found")
return ProfileResponse(
id=self._profile.id,
username=self._profile.username,
display_name=(
update.display_name
if update.display_name is not None
else self._profile.display_name
username=(
update.username
if update.username is not None
else self._profile.username
),
avatar_url=(
update.avatar_url
@@ -70,7 +69,6 @@ def test_get_me_returns_profile() -> None:
profile = ProfileResponse(
id=str(user_id),
username="demo",
display_name="Demo User",
avatar_url=None,
bio=None,
)
@@ -94,7 +92,6 @@ def test_patch_me_updates_profile() -> None:
profile = ProfileResponse(
id=str(user_id),
username="demo",
display_name="Demo User",
avatar_url=None,
bio=None,
)
@@ -107,11 +104,11 @@ def test_patch_me_updates_profile() -> None:
try:
response = client.patch(
"/api/v1/profile/me",
json={"display_name": "Updated"},
json={"username": "updated"},
)
assert response.status_code == 200
body = response.json()
assert body["display_name"] == "Updated"
assert body["username"] == "updated"
finally:
app.dependency_overrides = {}
@@ -120,7 +117,6 @@ def test_get_profile_by_username() -> None:
profile = ProfileResponse(
id="00000000-0000-0000-0000-000000000001",
username="demo",
display_name="Demo User",
avatar_url=None,
bio=None,
)
@@ -142,7 +138,6 @@ def test_profile_not_found_returns_problem_details() -> None:
profile = ProfileResponse(
id="00000000-0000-0000-0000-000000000001",
username="demo",
display_name="Demo User",
avatar_url=None,
bio=None,
)
@@ -167,7 +162,6 @@ def test_patch_me_validation_error_returns_problem_details() -> None:
profile = ProfileResponse(
id=str(user_id),
username="demo",
display_name="Demo User",
avatar_url=None,
bio=None,
)
@@ -186,3 +180,25 @@ def test_patch_me_validation_error_returns_problem_details() -> None:
assert body["status"] == 422
finally:
app.dependency_overrides = {}
def test_patch_me_rejects_display_name_field() -> None:
user_id = UUID("00000000-0000-0000-0000-000000000001")
profile = ProfileResponse(
id=str(user_id),
username="demo",
avatar_url=None,
bio=None,
)
app.dependency_overrides[get_profile_service] = _override_profile_service(
FakeProfileService(profile)
)
app.dependency_overrides[get_current_user] = _override_current_user(user_id)
client = TestClient(app)
try:
response = client.patch("/api/v1/profile/me", json={"display_name": "x"})
assert response.status_code == 422
assert response.headers["content-type"].startswith("application/problem+json")
finally:
app.dependency_overrides = {}