104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from app.main import app
|
|
from app.api.deps import get_current_user
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_registered_user():
|
|
"""Return a mock user representing a successfully registered user."""
|
|
user = AsyncMock()
|
|
user.id = uuid.UUID("12345678-1234-1234-1234-123456789abc")
|
|
user.email = "new@example.com"
|
|
user.name = "New User"
|
|
user.plan = "free"
|
|
user.max_queries = 5
|
|
user.is_active = True
|
|
user.created_at = datetime.now()
|
|
return user
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_success(async_client, mock_registered_user):
|
|
with patch("app.api.auth.register_user", return_value=mock_registered_user):
|
|
response = await async_client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "new@example.com",
|
|
"password": "password123",
|
|
"name": "New User",
|
|
},
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["email"] == "new@example.com"
|
|
assert data["name"] == "New User"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_register_duplicate_email(async_client):
|
|
with patch(
|
|
"app.api.auth.register_user",
|
|
side_effect=ValueError("Email already registered"),
|
|
):
|
|
response = await async_client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "existing@example.com",
|
|
"password": "password123",
|
|
"name": "Existing User",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
data = response.json()
|
|
assert "注册失败" in data["detail"] or "已被使用" in data["detail"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_success(async_client, mock_registered_user):
|
|
with patch("app.api.auth.authenticate_user", return_value=mock_registered_user):
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": "test@example.com", "password": "password123"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["user"]["email"] == "new@example.com"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_wrong_password(async_client):
|
|
with patch("app.api.auth.authenticate_user", return_value=None):
|
|
response = await async_client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": "test@example.com", "password": "wrongpassword"},
|
|
)
|
|
assert response.status_code == 401
|
|
data = response.json()
|
|
assert "邮箱或密码错误" in data["detail"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_me_authenticated(
|
|
async_client, override_get_current_user, auth_headers
|
|
):
|
|
response = await async_client.get("/api/v1/auth/me", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == "test@example.com"
|
|
assert data["name"] == "Test User"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_me_unauthenticated(async_client):
|
|
# Ensure no auth override is active
|
|
app.dependency_overrides.pop(get_current_user, None)
|
|
response = await async_client.get("/api/v1/auth/me")
|
|
assert response.status_code == 401
|