127 lines
3.7 KiB
Python
127 lines
3.7 KiB
Python
"""认证端点 API 测试。"""
|
|
|
|
from collections.abc import AsyncGenerator
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from app.main import app
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _jwt_enabled(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""所有 API 测试需要 JWT_SECRET 配置。"""
|
|
monkeypatch.setattr("app.core.config.settings.jwt_secret", "test-jwt-secret-for-testing")
|
|
|
|
|
|
@pytest.fixture
|
|
async def client() -> AsyncGenerator[AsyncClient, None]:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as http_client:
|
|
yield http_client
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_and_login(client: AsyncClient) -> None:
|
|
# Register
|
|
resp = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "testuser", "email": "test@example.com", "password": "testpass123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["username"] == "testuser"
|
|
assert data["email"] == "test@example.com"
|
|
user_id = data["id"]
|
|
assert user_id.startswith("user_")
|
|
|
|
# Login
|
|
resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "testuser", "password": "testpass123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["user"]["id"] == user_id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_duplicate_username(client: AsyncClient) -> None:
|
|
# Register first user
|
|
resp = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "dupe", "email": "dupe1@example.com", "password": "pass123"},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Try duplicate username
|
|
resp = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "dupe", "email": "dupe2@example.com", "password": "pass456"},
|
|
)
|
|
assert resp.status_code == 409
|
|
assert resp.json()["error"]["code"] == "conflict"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_wrong_password_login(client: AsyncClient) -> None:
|
|
await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "wrongpass", "email": "wrong@example.com", "password": "correct"},
|
|
)
|
|
resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "wrongpass", "password": "wrong"},
|
|
)
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_refresh_token(client: AsyncClient) -> None:
|
|
# Register & login
|
|
await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "refreshtest", "email": "refresh@example.com", "password": "pass"},
|
|
)
|
|
login_resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "refreshtest", "password": "pass"},
|
|
)
|
|
login_data = login_resp.json()
|
|
refresh_token = login_data["refresh_token"]
|
|
|
|
# Refresh
|
|
resp = await client.post(
|
|
"/api/v1/auth/refresh",
|
|
json={"refresh_token": refresh_token},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "access_token" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_me(client: AsyncClient) -> None:
|
|
await client.post(
|
|
"/api/v1/auth/register",
|
|
json={"username": "metest", "email": "me@example.com", "password": "pass"},
|
|
)
|
|
login_resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"username": "metest", "password": "pass"},
|
|
)
|
|
access_token = login_resp.json()["access_token"]
|
|
|
|
# GET /auth/me
|
|
resp = await client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {access_token}"},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["username"] == "metest"
|
|
assert data["email"] == "me@example.com"
|