ai-kefu-admin/tests/test_auth.py

"""平台认证(Phase 2 Task 2):login/logout/change-password + 认证依赖。

经 app_client(jiancai Host)打真实路由——Task 3 前 /api/platform/* 仍过
租户 Host 解析,jiancai 是已知域名故直达(计划注:本任务即挂路由,TDD
循环才能闭环)。conftest 的 lifespan bootstrap 超管 admin/admin123
(must_change=True)每测试全新临时平台库(Task 1 就绪)。

改密后 token 语义(本实现选定,见 auth_routes.change_password docstring):
当前 token 不失效、must_change 就地清零——改完密无需重登。

跨测试隔离:conftest autouse _reset_platform_auth 复位内存 token 表与
登录限流计数(照抄 rate_limit 纪律)。
"""
import pytest
from fastapi import HTTPException

from app.platform.auth import (
    UserContext,
    login_limiter,
    require_super,
    require_tenant,
)

LOGIN = "/api/platform/auth/login"
LOGOUT = "/api/platform/auth/logout"
CHANGE_PW = "/api/platform/auth/change-password"


def _login(client, username="admin", password="admin123"):
    return client.post(LOGIN, json={"username": username, "password": password})


def _auth(token):
    return {"Authorization": f"Bearer {token}"}


# ---- 登录 ---------------------------------------------------------------------


def test_login_ok_returns_token(app_client):
    r = _login(app_client)
    assert r.status_code == 200
    data = r.json()
    assert set(data) == {"token", "role", "tenant_slug", "must_change_password"}
    assert data["role"] == "super"
    assert data["tenant_slug"] is None
    assert data["must_change_password"] is True    # 引导密码须首登改掉
    assert data["token"]


def test_login_bad_password_401(app_client):
    r = _login(app_client, password="wrong")
    assert r.status_code == 401
    # 防枚举: 用户不存在与密码错同一响应(状态+文案都不区分)
    r2 = _login(app_client, username="nosuchuser")
    assert r2.status_code == 401
    assert r2.json() == r.json()


def test_login_lockout_after_5_fails(app_client):
    for _ in range(5):
        assert _login(app_client, password="wrong").status_code == 401
    # 第 6 次 429——即使密码正确(锁定检查先于验证)
    assert _login(app_client).status_code == 429
    # 窗口恢复: 已记录失败拨到窗外(等效 15 分钟过后,免真实等待)→ 可再登录
    key = ("admin", "testclient")
    login_limiter._fails[key] = [
        t - login_limiter.window_seconds - 1
        for t in login_limiter._fails[key]]
    assert _login(app_client).status_code == 200


# ---- token 生命周期 ------------------------------------------------------------


def test_token_expiry(app_client, monkeypatch):
    """TTL 可注入: 拨成负值 → 签发即过期 → 受保护端点 401。"""
    from app.platform import auth
    monkeypatch.setattr(auth.token_store, "ttl_seconds", -1)
    r = _login(app_client)
    assert r.status_code == 200
    token = r.json()["token"]
    r2 = app_client.post(
        CHANGE_PW, headers=_auth(token),
        json={"old_password": "admin123", "new_password": "n3w-pass!"})
    assert r2.status_code == 401


def test_protected_requires_auth(app_client):
    """无 Authorization → 401(依赖链 get_current_user 先于 enforce)。"""
    assert app_client.post(LOGOUT).status_code == 401


def test_logout_invalidates_token(app_client):
    token = _login(app_client).json()["token"]
    # must_change=True → logout 409,先改密(enforce 语义连锁,见下)
    app_client.post(
        CHANGE_PW, headers=_auth(token),
        json={"old_password": "admin123", "new_password": "n3w-pass-2"})
    assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200
    # 同 token 再用 → 401(已撤销,立即失效)
    assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 401


# ---- 改密 + must_change 409 语义 -------------------------------------------------


def test_change_password_flow(app_client):
    token = _login(app_client).json()["token"]

    # 旧密改新密成功
    r = app_client.post(
        CHANGE_PW, headers=_auth(token),
        json={"old_password": "admin123", "new_password": "n3w-pass-1"})
    assert r.status_code == 200

    # 选定语义: 改密后当前 token 不失效、must_change 就地清零
    # (logout 受 enforce 保护,改密后可 200 即证 token 存活且标志已清)
    assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200

    # 旧密失效
    assert _login(app_client).status_code == 401
    # 新密可登录,引导标志已清
    data = _login(app_client, password="n3w-pass-1").json()
    assert data["must_change_password"] is False


def test_must_change_password_409_semantics(app_client):
    """引导密码强制语义: must_change=True 时受保护端点(logout)409;
    change-password 本身豁免(200);改完 409 消除。"""
    token = _login(app_client).json()["token"]
    assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 409
    r = app_client.post(
        CHANGE_PW, headers=_auth(token),
        json={"old_password": "admin123", "new_password": "n3w-pass-3"})
    assert r.status_code == 200
    assert app_client.post(LOGOUT, headers=_auth(token)).status_code == 200


# ---- 角色依赖(工厂直测;端点级覆盖在 Task 5)---------------------------------------


def test_tenant_admin_role_scoped():
    tenant_admin = UserContext(
        user_id=2, role="tenant", tenant_slug="jiancai",
        must_change_password=False)
    with pytest.raises(HTTPException) as ei:
        require_super(tenant_admin)
    assert ei.value.status_code == 403

    dep = require_tenant("jiancai")
    assert dep(tenant_admin) is tenant_admin       # 本租户放行
    super_user = UserContext(
        user_id=1, role="super", tenant_slug=None, must_change_password=False)
    assert dep(super_user) is super_user           # super 全通
    other = UserContext(
        user_id=3, role="tenant", tenant_slug="auto",
        must_change_password=False)
    with pytest.raises(HTTPException) as ei2:
        dep(other)
    assert ei2.value.status_code == 403            # 它租户 403