server-core/tests/test_auth.py

# -*- coding: utf-8 -*-
"""认证接口端到端测试（使用临时 SQLite 数据库，不污染 data/）。"""
from __future__ import annotations

DEMO_USERNAME = "pine"
DEMO_PASSWORD = "123456"


def login(client, username=DEMO_USERNAME, password=DEMO_PASSWORD) -> dict:
    res = client.post("/auth/login", json={"username": username, "password": password})
    assert res.status_code == 200, res.text
    return res.json()


def auth(token: str) -> dict:
    return {"Authorization": f"Bearer {token}"}


# ---------------------------------------------------------------------------
# 基础 / 状态
# ---------------------------------------------------------------------------
def test_health(client):
    res = client.get("/health")
    assert res.status_code == 200
    assert res.json()["status"] == "ok"


def test_status(client):
    res = client.get("/auth/status")
    assert res.status_code == 200
    assert res.json() == {"enabled": True, "has_users": True}


# ---------------------------------------------------------------------------
# 登录 / 注册
# ---------------------------------------------------------------------------
def test_login_success_returns_profile_and_token(client):
    data = login(client)
    assert data["token"]
    assert data["username"] == DEMO_USERNAME
    # 演示资料字段齐全
    for field in ("nickname", "account", "company", "room", "avatar", "company_avatar"):
        assert field in data


def test_login_wrong_password(client):
    res = client.post(
        "/auth/login",
        json={"username": DEMO_USERNAME, "password": "wrong"},
    )
    assert res.status_code == 401
    assert res.json()["detail"] == "Invalid username or password"


def test_register_denied_when_users_exist(client):
    res = client.post(
        "/auth/register",
        json={"username": "13800001234", "password": "whatever1"},
    )
    assert res.status_code == 403
    assert res.json()["detail"] == "User already registered"


# ---------------------------------------------------------------------------
# 令牌校验 / 当前用户
# ---------------------------------------------------------------------------
def test_verify_with_token(client):
    token = login(client)["token"]
    res = client.get("/auth/verify", headers=auth(token))
    assert res.status_code == 200
    assert res.json() == {"valid": True, "username": DEMO_USERNAME}


def test_verify_without_token(client):
    res = client.get("/auth/verify")
    assert res.status_code == 401


def test_me_returns_profile(client):
    token = login(client)["token"]
    res = client.get("/auth/me", headers=auth(token))
    assert res.status_code == 200
    assert res.json()["username"] == DEMO_USERNAME
    assert res.json()["nickname"]


# ---------------------------------------------------------------------------
# 资料 / 凭据更新
# ---------------------------------------------------------------------------
def test_update_profile_fields(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"nickname": "新昵称", "account": "新账号", "room": "2001"},
    )
    assert res.status_code == 200, res.text
    body = res.json()
    assert body["nickname"] == "新昵称"
    assert body["account"] == "新账号"
    assert body["room"] == "2001"
    # 仅改资料不重签令牌
    assert body["token"] == ""


def test_update_profile_nothing_to_update(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"current_password": DEMO_PASSWORD},
    )
    assert res.status_code == 400
    assert res.json()["detail"] == "Nothing to update"


def test_update_profile_wrong_current_password(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"current_password": "wrong", "new_password": "newpass123"},
    )
    assert res.status_code == 401
    assert res.json()["detail"] == "Current password is incorrect"


def test_update_profile_changes_password_and_reissues_token(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"current_password": DEMO_PASSWORD, "new_password": "newpass123"},
    )
    assert res.status_code == 200, res.text
    body = res.json()
    assert body["token"]  # 新令牌

    # 旧令牌失效，新令牌可用
    assert client.get("/auth/verify", headers=auth(token)).status_code == 401
    assert (
        client.get("/auth/verify", headers=auth(body["token"])).status_code == 200
    )

    # 新密码可登录，旧密码不可
    assert login(client, password="newpass123")["token"]
    assert (
        client.post(
            "/auth/login",
            json={"username": DEMO_USERNAME, "password": DEMO_PASSWORD},
        ).status_code
        == 401
    )


def test_update_profile_changes_username(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"current_password": DEMO_PASSWORD, "new_username": "pine2"},
    )
    assert res.status_code == 200, res.text
    assert res.json()["username"] == "pine2"
    assert res.json()["token"]

    # 新用户名可登录
    res = client.post(
        "/auth/login",
        json={"username": "pine2", "password": DEMO_PASSWORD},
    )
    assert res.status_code == 200


def test_update_profile_empty_new_password_rejected(client):
    token = login(client)["token"]
    res = client.post(
        "/auth/update-profile",
        headers=auth(token),
        json={"current_password": DEMO_PASSWORD, "new_password": "  "},
    )
    assert res.status_code == 400
    assert "cannot be empty" in res.json()["detail"]


# ---------------------------------------------------------------------------
# 令牌吊销
# ---------------------------------------------------------------------------
def test_revoke_current_token(client):
    token = login(client)["token"]
    res = client.post("/auth/revoke-token", headers=auth(token), json={})
    assert res.status_code == 200
    assert res.json()["revoked"] is True
    assert res.json()["revoked_current_token"] is True
    assert client.get("/auth/verify", headers=auth(token)).status_code == 401


def test_revoke_specific_token(client):
    token_a = login(client)["token"]
    token_b = login(client)["token"]
    res = client.post(
        "/auth/revoke-token",
        headers=auth(token_a),
        json={"token": token_b},
    )
    assert res.status_code == 200
    assert res.json()["revoked_current_token"] is False
    assert client.get("/auth/verify", headers=auth(token_b)).status_code == 401
    assert client.get("/auth/verify", headers=auth(token_a)).status_code == 200


def test_revoke_all_tokens(client):
    token_a = login(client)["token"]
    token_b = login(client)["token"]
    res = client.post("/auth/revoke-all-tokens", headers=auth(token_a), json={})
    assert res.status_code == 200
    assert res.json()["revoked"] is True
    assert client.get("/auth/verify", headers=auth(token_a)).status_code == 401
    assert client.get("/auth/verify", headers=auth(token_b)).status_code == 401


# ---------------------------------------------------------------------------
# 落盘格式
# ---------------------------------------------------------------------------
def test_persisted_user_has_no_plaintext_password(client, tmp_path):
    login(client)
    # 从 SQLite 读取用户记录，断言不落明文密码
    import sqlite3

    conn = sqlite3.connect(tmp_path / "test.db")
    row = conn.execute(
        "SELECT username, password_hash, password_salt FROM users WHERE username=?",
        (DEMO_USERNAME,),
    ).fetchone()
    conn.close()
    assert row is not None
    username, password_hash, password_salt = row
    assert username == DEMO_USERNAME
    assert password_hash
    assert password_salt
    # users 表没有 password 明文列
    conn = sqlite3.connect(tmp_path / "test.db")
    cols = [r[1] for r in conn.execute("PRAGMA table_info(users)")]
    conn.close()
    assert "password" not in cols