"""登录链路端到端测试（需要 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__)