yunke/yunke-server/tests/test_api_login.py

"""Tests for ``POST /api/auth/login`` (username/password → shared API token).

Behavior pinned here:

* Correct credentials -> ``200 {"token": <api_token>}`` (the SPA then uses it
  as the Bearer token for every subsequent call).
* Wrong username or password -> ``401`` with a generic message (no user
  enumeration — the same detail for both cases).
* The login endpoint itself is exempt from the shared-token middleware —
  obviously, the caller doesn't have a token yet.
* Brute-force guard: the 5th consecutive failure from one IP locks that IP
  out for 15 minutes (``429`` even with correct credentials); a successful
  login resets the failure counter.
* Works with auth disabled (empty ``API_TOKEN``) -> returns an empty token.
"""

from __future__ import annotations

import pytest
from httpx import ASGITransport, AsyncClient

from app.api import auth as auth_module
from app.api.main import app
from app.config import get_settings


@pytest.fixture
async def login_client(monkeypatch):
    """In-process client with a clean settings/login state per test."""
    monkeypatch.delenv("API_TOKEN", raising=False)
    monkeypatch.delenv("ADMIN_USER", raising=False)
    monkeypatch.delenv("ADMIN_PASSWORD", raising=False)
    get_settings.cache_clear()
    failures = getattr(auth_module, "_login_failures", None)
    if failures is not None:
        failures.clear()
    ac = AsyncClient(transport=ASGITransport(app=app), base_url="http://test")
    await ac.__aenter__()
    yield ac
    await ac.aclose()
    get_settings.cache_clear()


async def test_login_with_correct_credentials_returns_token(login_client, monkeypatch):
    monkeypatch.setenv("API_TOKEN", "test-secret")
    get_settings.cache_clear()
    r = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert r.status_code == 200
    assert r.json() == {"token": "test-secret"}


async def test_login_with_wrong_password_returns_401(login_client, monkeypatch):
    monkeypatch.setenv("API_TOKEN", "test-secret")
    get_settings.cache_clear()
    r = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "nope"}
    )
    assert r.status_code == 401
    assert r.json()["detail"] == "用户名或密码错误"


async def test_login_with_unknown_user_same_error_as_wrong_password(
    login_client, monkeypatch
):
    """No user enumeration: unknown user and wrong password read identically."""
    monkeypatch.setenv("API_TOKEN", "test-secret")
    get_settings.cache_clear()
    r = await login_client.post(
        "/api/auth/login", json={"username": "nobody", "password": "admin123"}
    )
    assert r.status_code == 401
    assert r.json()["detail"] == "用户名或密码错误"


async def test_login_credentials_come_from_env(login_client, monkeypatch):
    """ADMIN_USER / ADMIN_PASSWORD override the defaults."""
    monkeypatch.setenv("API_TOKEN", "s")
    monkeypatch.setenv("ADMIN_USER", "boss")
    monkeypatch.setenv("ADMIN_PASSWORD", "boss-pw")
    get_settings.cache_clear()
    ok = await login_client.post(
        "/api/auth/login", json={"username": "boss", "password": "boss-pw"}
    )
    assert ok.status_code == 200
    legacy = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert legacy.status_code == 401


async def test_login_with_auth_disabled_returns_empty_token(login_client):
    """Empty API_TOKEN (auth off) -> login still succeeds with an empty token."""
    r = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert r.status_code == 200
    assert r.json() == {"token": ""}


async def test_lockout_after_five_consecutive_failures(login_client, monkeypatch):
    """5 failures from one IP -> 6th attempt is 429 even with correct creds."""
    monkeypatch.setenv("API_TOKEN", "s")
    get_settings.cache_clear()
    for _ in range(5):
        r = await login_client.post(
            "/api/auth/login", json={"username": "admin", "password": "bad"}
        )
        assert r.status_code == 401
    locked = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert locked.status_code == 429


async def test_successful_login_resets_failure_counter(login_client, monkeypatch):
    """4 failures + a success + 4 more failures -> still not locked."""
    monkeypatch.setenv("API_TOKEN", "s")
    get_settings.cache_clear()
    for _ in range(4):
        await login_client.post(
            "/api/auth/login", json={"username": "admin", "password": "bad"}
        )
    ok = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert ok.status_code == 200
    for _ in range(4):
        await login_client.post(
            "/api/auth/login", json={"username": "admin", "password": "bad"}
        )
    still_ok = await login_client.post(
        "/api/auth/login", json={"username": "admin", "password": "admin123"}
    )
    assert still_ok.status_code == 200