danyang-match/tests/e2e/auth_refresh_test.py

"""
认证刷新链路 E2E。

背景：2026-09-22 用户在微信开发者工具里看到首页全站 401，追下去发现「刷新」
这条路整条是断的，而且断在三个互不相干的地方：

  1. 前端漏传 userId → 后端 `/auth/refresh` 直接 400；
  2. 后端把 refresh token 交给 `validateToken` 校验，而那是用 **access 密钥**
     验签的，refresh token 由 **refresh 密钥**签发 —— 必然失败；
  3. 失败被抛成裸 `RuntimeException` → HTTP 500，客户端当成服务端故障重试。

第 1 层修好后第 2 层才显形，第 2 层修好后第 3 层才显形 —— 单层修复都会「看起来好了」。
所以这里断言的是**整条链路的可用性**，而不是某一步返回 200：

  * 换发出来的 accessToken 必须真的能通过鉴权（访问受保护端点）；
  * 换发后旧 refreshToken 必须失效（可撤销性）；
  * 刷新失败必须回 401 而不是 500（客户端据此决定「重新登录」而非「重试」）。

此前全仓没有任何 /auth/refresh 用例，这正是该缺陷能长期存活的原因。
"""
import uuid

import pytest

pytestmark = pytest.mark.e2e

MISSING_UUID = "00000000-0000-0000-0000-000000000000"


def _refresh(api, token: str, refresh_token: str, **extra):
    return api.call(
        "POST",
        "/auth/refresh",
        json={"refreshToken": refresh_token, **extra},
    )


def test_refresh_returns_access_token_that_really_works(api, new_user):
    """刷新换发的 accessToken 必须能通过鉴权 —— 只看 200 会被「返回了空壳令牌」骗过。"""
    user = new_user()
    resp = _refresh(api, user["token"], user["refresh_token"], userId=user["user_id"])
    assert resp.status_code == 200, (
        f"/auth/refresh 必须成功（refreshToken 由 refresh 密钥签发，"
        f"用 access 密钥验签会必然失败）。实际 {resp.status_code}: {resp.text[:300]}"
    )
    body = resp.json().get("data") or {}
    new_access = body.get("accessToken")
    new_refresh = body.get("refreshToken")
    assert new_access, f"刷新响应必须包含 accessToken：{body}"
    assert new_refresh, f"刷新响应必须包含新的 refreshToken（令牌轮换）：{body}"
    # 令牌必须互异：JWT 的 iat 是秒级，若载荷只有 subject/iat/type，
    # 同一秒内签发的令牌会逐字节相同 —— 「刷新即轮换」随之失效（旧令牌仍然可用）。
    # 这个断言在 2026-09-22 咬出过该缺陷，故保留。
    assert new_access != user["token"], "刷新应换发新令牌，而不是回显旧令牌"
    assert new_refresh != user["refresh_token"], (
        "刷新必须换发不同的 refreshToken，否则无法作废旧令牌"
    )

    # 关键：新令牌不是装饰品，必须真的能过鉴权
    probe = api.call("GET", "/me/status", token=new_access)
    assert probe.status_code == 200, (
        f"刷新换发的 accessToken 必须能访问受保护端点，实际 {probe.status_code}: {probe.text[:300]}"
    )


def test_refresh_does_not_require_user_id(api, new_user):
    """
    userId 不是必填项：refresh token 的 subject 就是 userId，令牌自证身份。

    强制要求客户端重复传一个它已经持有的凭据里已包含的信息，只会制造
    「漏传 → 400」这类无意义失败（正是这次事故放大的起点）。
    """
    user = new_user()
    resp = _refresh(api, user["token"], user["refresh_token"])
    assert resp.status_code == 200, (
        f"只带 refreshToken 也必须能刷新（身份来自令牌 subject），实际 {resp.status_code}: {resp.text[:300]}"
    )
    assert (resp.json().get("data") or {}).get("accessToken"), "应换发 accessToken"


def test_refresh_rejects_mismatched_user_id(api, new_user):
    """显式传入的 userId 与令牌主体不一致时必须拒绝，防止拿 A 的令牌刷 B 的会话。"""
    user = new_user()
    other = new_user()
    resp = _refresh(api, user["token"], user["refresh_token"], userId=other["user_id"])
    assert resp.status_code in (401, 403), (
        f"userId 与令牌主体不一致必须被拒绝（401/403），实际 {resp.status_code}: {resp.text[:300]}"
    )


def test_rotating_refresh_token_invalidates_the_previous_one(api, new_user):
    """刷新后旧 refreshToken 必须失效（Redis 中登记值已被换发覆盖）。"""
    user = new_user()
    first = _refresh(api, user["token"], user["refresh_token"])
    assert first.status_code == 200, f"前置条件失败：首次刷新应成功: {first.text[:300]}"

    reused = _refresh(api, user["token"], user["refresh_token"])
    assert reused.status_code == 401, (
        f"已轮换掉的 refreshToken 必须失效并回 401，实际 {reused.status_code}: {reused.text[:300]}"
    )


def test_invalid_refresh_token_is_401_not_500(api, new_user):
    """
    刷新凭据无效 = 401，不是 500。

    500 会让客户端判定为服务端故障并重试，而重试必然再失败；
    401 才能让客户端停止重试、走「重新登录」。这也是小程序 api.js
    区分 401 / 其他失败分支的前提。
    """
    user = new_user()
    for bad_token in (str(uuid.uuid4()), "not-a-jwt"):
        resp = _refresh(api, user["token"], bad_token)
        assert resp.status_code == 401, (
            f"无效 refreshToken({bad_token[:12]}…) 应回 401，实际 {resp.status_code}: {resp.text[:300]}"
        )


def test_refresh_without_refresh_token_is_400(api):
    """完全没带凭据属于请求问题 → 400（与「凭据无效」的 401 区分开）。"""
    resp = api.call("POST", "/auth/refresh", json={})
    assert resp.status_code == 400, (
        f"缺少 refreshToken 应回 400，实际 {resp.status_code}: {resp.text[:300]}"
    )