PayrollManagementSystem/backend/tests/integration/test_auth_flow.py

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