102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
|
"""批次 3 回归：管理员重置密码走 JSON body。
|
||
|
|
|
||
|
|
此前后端把 new_password 声明为裸 str 参数（FastAPI 解析为 query param），
|
||
|
|
前端两个调用点均发送 JSON body，重置密码端到端断裂（必 422）。
|
||
|
|
现收敛为 Pydantic 请求模型 { new_password }，与 api-client.ts 结构一致。
|
||
|
|
"""
|
||
|
|
from types import SimpleNamespace
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from httpx import AsyncClient, ASGITransport
|
||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
|
||
|
|
from sqlalchemy import select
|
||
|
|
|
||
|
|
from shared.database.database import get_db_session
|
||
|
|
from shared.models.identity import User
|
||
|
|
from shared.services.auth_service import (
|
||
|
|
get_current_active_user,
|
||
|
|
get_password_hash,
|
||
|
|
verify_password,
|
||
|
|
)
|
||
|
|
from shared.services.auth_routes import router as auth_router
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def reset_env(async_engine, seeded_db):
|
||
|
|
"""播种被重置目标用户（id=2，已知旧密码）；返回 (client, app, session_factory)。"""
|
||
|
|
session_factory = async_sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)
|
||
|
|
|
||
|
|
async with session_factory() as session:
|
||
|
|
target = User(
|
||
|
|
id=2, username="resetme", email="resetme@example.com",
|
||
|
|
hashed_password=get_password_hash("oldpass123"), is_active=True,
|
||
|
|
)
|
||
|
|
session.add(target)
|
||
|
|
await session.commit()
|
||
|
|
|
||
|
|
test_app = FastAPI()
|
||
|
|
test_app.include_router(auth_router) # router 自带 /api/auth 前缀
|
||
|
|
|
||
|
|
async def override_get_db_session():
|
||
|
|
async with session_factory() as session:
|
||
|
|
yield session
|
||
|
|
|
||
|
|
test_app.dependency_overrides[get_db_session] = override_get_db_session
|
||
|
|
|
||
|
|
transport = ASGITransport(app=test_app)
|
||
|
|
async with AsyncClient(transport=transport, base_url="http://test") as ac:
|
||
|
|
yield ac, test_app, session_factory
|
||
|
|
|
||
|
|
test_app.dependency_overrides.clear()
|
||
|
|
|
||
|
|
|
||
|
|
def _login_as(app: FastAPI, *, superuser: bool):
|
||
|
|
# User.is_superuser 为只读 hybrid property，覆写用户用 SimpleNamespace 承载
|
||
|
|
app.dependency_overrides[get_current_active_user] = lambda: SimpleNamespace(
|
||
|
|
id=500, username="admin", is_superuser=superuser
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_reset_password_with_json_body(reset_env):
|
||
|
|
"""JSON body { new_password } 生效：密码真实更新且可用新口令验证。"""
|
||
|
|
ac, app, session_factory = reset_env
|
||
|
|
_login_as(app, superuser=True)
|
||
|
|
|
||
|
|
resp = await ac.put(
|
||
|
|
"/api/auth/users/2/reset-password",
|
||
|
|
json={"new_password": "brandnew456"},
|
||
|
|
)
|
||
|
|
assert resp.status_code == 200
|
||
|
|
|
||
|
|
async with session_factory() as session:
|
||
|
|
user = (await session.execute(select(User).where(User.id == 2))).scalar_one()
|
||
|
|
assert verify_password("brandnew456", user.hashed_password)
|
||
|
|
assert not verify_password("oldpass123", user.hashed_password)
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_reset_password_rejects_short_password(reset_env):
|
||
|
|
"""最短 6 位对齐前端校验，违约 422。"""
|
||
|
|
ac, app, _ = reset_env
|
||
|
|
_login_as(app, superuser=True)
|
||
|
|
|
||
|
|
resp = await ac.put(
|
||
|
|
"/api/auth/users/2/reset-password",
|
||
|
|
json={"new_password": "abc"},
|
||
|
|
)
|
||
|
|
assert resp.status_code == 422
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_reset_password_requires_admin(reset_env):
|
||
|
|
ac, app, _ = reset_env
|
||
|
|
_login_as(app, superuser=False)
|
||
|
|
|
||
|
|
resp = await ac.put(
|
||
|
|
"/api/auth/users/2/reset-password",
|
||
|
|
json={"new_password": "brandnew456"},
|
||
|
|
)
|
||
|
|
assert resp.status_code == 403
|