273ebf4657
后端 FastAPI 六层结构（api/services/repositories/models/schemas/core）， 统一响应信封、全局异常、日志脱敏；6 张表经 Alembic 迁移建库，SQL 留存 db/migration；登录鉴权含图形验证码（Redis GETDEL 防重放）、JWT、登录限流、 登出黑名单（TTL=令牌剩余有效期）、登录日志。 前端 Vue3 + TS + Element Plus 骨架，含登录页、路由守卫、请求封装与 Vite 代理。 工程配置：uv 锁定 Python 3.12、ruff 检查与格式化、editorconfig、gitattributes。 测试 184 项（单元 155 / 集成 29）全绿，浏览器端到端 8 项通过。 Co-Authored-By: Claude Code <noreply@anthropic.com>
272 lines
11 KiB
Python
272 lines
11 KiB
Python
"""登录链路端到端测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是整条链路的**串联正确性**：验证码 → 登录 → 持令牌访问 → 登出失效。
|
||
各环节细节已由单元测试覆盖，这里验证它们接起来是否真的可用 ——
|
||
尤其是「判定顺序」「失败时日志是否真的落库」「禁用与提版能否立即生效」
|
||
这类只有串联运行才会暴露的问题。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
"""
|
||
|
||
import uuid
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import delete, select
|
||
|
||
from app.core.constants import REDIS_KEY_CAPTCHA, REDIS_KEY_LOGIN_FAIL
|
||
from app.core.db import get_session_factory
|
||
from app.core.redis_client import get_redis
|
||
from app.core.security import hash_password
|
||
from app.models import LoginLog, User
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
TEST_PASSWORD = "TestPwd@2026"
|
||
|
||
|
||
@pytest.fixture
|
||
def test_user():
|
||
"""创建独立测试账号，用后即删。
|
||
|
||
不复用 admin：测试会触发账号锁定、提升 token_version 等操作，
|
||
污染真实账号会让后续手工验证产生困惑。
|
||
"""
|
||
factory = get_session_factory()
|
||
username = f"t_{uuid.uuid4().hex[:8]}"
|
||
with factory() as session:
|
||
user = User(
|
||
username=username,
|
||
password=hash_password(TEST_PASSWORD),
|
||
real_name="测试账号",
|
||
status=1,
|
||
)
|
||
session.add(user)
|
||
session.commit()
|
||
user_id = user.id
|
||
|
||
yield {"id": user_id, "username": username}
|
||
|
||
with factory() as session:
|
||
session.execute(delete(LoginLog).where(LoginLog.username == username))
|
||
session.execute(delete(User).where(User.id == user_id)) # user_roles 级联删除
|
||
session.commit()
|
||
get_redis().delete(REDIS_KEY_LOGIN_FAIL.format(username=username))
|
||
|
||
|
||
def _fetch_captcha(client: TestClient) -> tuple[str, str]:
|
||
"""取验证码及其明文答案。
|
||
|
||
直接从 Redis 读取答案 —— **仅测试可用**，生产环境当然不能如此，
|
||
这里是为了让自动化用例不必做图像识别。
|
||
"""
|
||
data = client.get("/api/v1/auth/captcha").json()["data"]
|
||
captcha_id = data["captcha_id"]
|
||
answer = get_redis().get(REDIS_KEY_CAPTCHA.format(captcha_id=captcha_id))
|
||
return captcha_id, answer
|
||
|
||
|
||
def _login(client: TestClient, username: str, password: str, **overrides):
|
||
"""发起一次登录，默认使用新鲜验证码。"""
|
||
captcha_id, code = _fetch_captcha(client)
|
||
payload = {
|
||
"username": username,
|
||
"password": password,
|
||
"captcha_id": captcha_id,
|
||
"captcha_code": code,
|
||
}
|
||
payload.update(overrides)
|
||
return client.post("/api/v1/auth/login", json=payload)
|
||
|
||
|
||
def _bearer(token: str) -> dict[str, str]:
|
||
return {"Authorization": f"Bearer {token}"}
|
||
|
||
|
||
class TestHappyPath:
|
||
"""正常链路。"""
|
||
|
||
def test_验证码到登出的完整链路(self, client: TestClient, test_user: dict) -> None:
|
||
captcha_id, code = _fetch_captcha(client)
|
||
login = client.post(
|
||
"/api/v1/auth/login",
|
||
json={
|
||
"username": test_user["username"],
|
||
"password": TEST_PASSWORD,
|
||
"captcha_id": captcha_id,
|
||
"captcha_code": code,
|
||
},
|
||
)
|
||
assert login.status_code == 200
|
||
body = login.json()
|
||
assert body["code"] == 200
|
||
token = body["data"]["access_token"]
|
||
assert body["data"]["token_type"] == "Bearer"
|
||
assert body["data"]["user"]["username"] == test_user["username"]
|
||
|
||
me = client.get("/api/v1/auth/me", headers=_bearer(token))
|
||
assert me.json()["data"]["username"] == test_user["username"]
|
||
|
||
assert client.post("/api/v1/auth/logout", headers=_bearer(token)).json()["code"] == 200
|
||
after = client.get("/api/v1/auth/me", headers=_bearer(token))
|
||
assert after.status_code == 401
|
||
assert after.json()["code"] == 401
|
||
|
||
def test_登录成功后清除失败计数(self, client: TestClient, test_user: dict) -> None:
|
||
username = test_user["username"]
|
||
for _ in range(3):
|
||
_login(client, username, "WrongPwd@1")
|
||
assert get_redis().exists(REDIS_KEY_LOGIN_FAIL.format(username=username)) == 1
|
||
|
||
assert _login(client, username, TEST_PASSWORD).status_code == 200
|
||
assert get_redis().exists(REDIS_KEY_LOGIN_FAIL.format(username=username)) == 0
|
||
|
||
|
||
class TestCaptcha:
|
||
"""验证码环节。"""
|
||
|
||
def test_同一验证码不可重放(self, client: TestClient, test_user: dict) -> None:
|
||
captcha_id, code = _fetch_captcha(client)
|
||
payload = {
|
||
"username": test_user["username"],
|
||
"password": TEST_PASSWORD,
|
||
"captcha_id": captcha_id,
|
||
"captcha_code": code,
|
||
}
|
||
assert client.post("/api/v1/auth/login", json=payload).status_code == 200
|
||
replay = client.post("/api/v1/auth/login", json=payload)
|
||
assert replay.status_code == 400
|
||
assert "验证码" in replay.json()["message"]
|
||
|
||
def test_验证码错误时拒绝登录(self, client: TestClient, test_user: dict) -> None:
|
||
response = _login(client, test_user["username"], TEST_PASSWORD, captcha_code="ZZZZ")
|
||
assert response.status_code == 400
|
||
assert "验证码" in response.json()["message"]
|
||
|
||
def test_验证码错误不消耗失败次数(self, client: TestClient, test_user: dict) -> None:
|
||
"""验证码与密码是两类失败，混在一起计数会让用户被误锁。"""
|
||
username = test_user["username"]
|
||
for _ in range(5):
|
||
_login(client, username, TEST_PASSWORD, captcha_code="ZZZZ")
|
||
# 之后用正确验证码仍可正常登录
|
||
assert _login(client, username, TEST_PASSWORD).status_code == 200
|
||
|
||
|
||
class TestCredentials:
|
||
"""凭据校验。"""
|
||
|
||
def test_密码错误被拒(self, client: TestClient, test_user: dict) -> None:
|
||
response = _login(client, test_user["username"], "WrongPwd@1")
|
||
assert response.status_code == 400
|
||
|
||
def test_账号不存在与密码错误的提示完全相同(self, client: TestClient) -> None:
|
||
"""文案不一致会形成账号枚举信道，攻击者可据此筛出真实用户名。"""
|
||
missing = _login(client, f"nobody_{uuid.uuid4().hex[:8]}", "Whatever@1")
|
||
assert missing.status_code == 400
|
||
assert missing.json()["message"] == "用户名或密码错误"
|
||
|
||
def test_无令牌访问受保护接口返回401(self, client: TestClient) -> None:
|
||
assert client.get("/api/v1/auth/me").status_code == 401
|
||
|
||
def test_伪造令牌返回401(self, client: TestClient) -> None:
|
||
response = client.get("/api/v1/auth/me", headers=_bearer("not.a.real.token"))
|
||
assert response.status_code == 401
|
||
assert response.json()["code"] == 401
|
||
|
||
|
||
class TestAccountState:
|
||
"""账号状态与令牌失效。"""
|
||
|
||
def test_禁用账号无法登录(self, client: TestClient, test_user: dict) -> None:
|
||
with get_session_factory()() as session:
|
||
user = session.get(User, test_user["id"])
|
||
user.status = 0
|
||
session.commit()
|
||
|
||
response = _login(client, test_user["username"], TEST_PASSWORD)
|
||
assert response.status_code == 403
|
||
assert "禁用" in response.json()["message"]
|
||
|
||
def test_禁用后已签发的令牌立即失效(self, client: TestClient, test_user: dict) -> None:
|
||
token = _login(client, test_user["username"], TEST_PASSWORD).json()["data"]["access_token"]
|
||
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 200
|
||
|
||
with get_session_factory()() as session:
|
||
user = session.get(User, test_user["id"])
|
||
user.status = 0
|
||
session.commit()
|
||
|
||
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 401
|
||
|
||
def test_提升token_version使旧令牌立即失效(self, client: TestClient, test_user: dict) -> None:
|
||
"""这是「离职即失效」的实现手段，且**不依赖 Redis**。"""
|
||
token = _login(client, test_user["username"], TEST_PASSWORD).json()["data"]["access_token"]
|
||
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 200
|
||
|
||
with get_session_factory()() as session:
|
||
user = session.get(User, test_user["id"])
|
||
user.token_version += 1 # 相当于改密 / 强制下线
|
||
session.commit()
|
||
|
||
assert client.get("/api/v1/auth/me", headers=_bearer(token)).status_code == 401
|
||
|
||
|
||
class TestRateLimit:
|
||
"""登录限流。"""
|
||
|
||
def test_连续失败达阈值后正确密码也被拒(self, client: TestClient, test_user: dict) -> None:
|
||
username = test_user["username"]
|
||
for _ in range(5):
|
||
_login(client, username, "WrongPwd@1")
|
||
|
||
response = _login(client, username, TEST_PASSWORD)
|
||
assert response.status_code == 403
|
||
assert "锁定" in response.json()["message"]
|
||
|
||
def test_锁定期内失败计数未被继续放大(self, client: TestClient, test_user: dict) -> None:
|
||
"""锁定后仍不断尝试不应无限刷新 TTL——否则会被用来永久锁死他人账号。"""
|
||
username = test_user["username"]
|
||
for _ in range(5):
|
||
_login(client, username, "WrongPwd@1")
|
||
ttl_before = get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
|
||
|
||
for _ in range(3):
|
||
_login(client, username, TEST_PASSWORD)
|
||
|
||
ttl_after = get_redis().ttl(REDIS_KEY_LOGIN_FAIL.format(username=username))
|
||
assert ttl_after <= ttl_before
|
||
|
||
|
||
class TestLoginLogs:
|
||
"""登录日志落库。"""
|
||
|
||
def _logs(self, username: str) -> list[LoginLog]:
|
||
with get_session_factory()() as session:
|
||
return list(
|
||
session.execute(
|
||
select(LoginLog).where(LoginLog.username == username).order_by(LoginLog.id)
|
||
).scalars()
|
||
)
|
||
|
||
def test_成功与失败均落库且记录失败原因(self, client: TestClient, test_user: dict) -> None:
|
||
username = test_user["username"]
|
||
_login(client, username, TEST_PASSWORD)
|
||
_login(client, username, "WrongPwd@1")
|
||
_login(client, username, TEST_PASSWORD, captcha_code="ZZZZ")
|
||
|
||
reasons = [(log.success, log.fail_reason) for log in self._logs(username)]
|
||
assert (1, None) in reasons
|
||
assert (0, "用户名或密码错误") in reasons
|
||
assert (0, "验证码错误") in reasons
|
||
|
||
def test_日志不含任何密码字段(self, client: TestClient, test_user: dict) -> None:
|
||
"""日志是最容易泄漏凭据的地方，这条断言防止日后有人「顺手」加字段。"""
|
||
username = test_user["username"]
|
||
_login(client, username, TEST_PASSWORD)
|
||
_login(client, username, "WrongPwd@1")
|
||
|
||
columns = {column.name for column in LoginLog.__table__.columns}
|
||
assert not (columns & {"password", "passwd", "pwd", "credential"})
|
||
for log in self._logs(username):
|
||
assert TEST_PASSWORD not in str(log.__dict__)
|