293 lines
8.7 KiB
Python
293 lines
8.7 KiB
Python
"""
|
|
认证集成测试
|
|
|
|
测试完整的认证流程，包括注册、登录、令牌刷新、登出等。
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from apps.api.main import app
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
class TestUserRegistration:
|
|
"""用户注册集成测试"""
|
|
|
|
def test_register_with_valid_data(self):
|
|
"""测试使用有效数据进行注册"""
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "newuser@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "newuser",
|
|
"display_name": "New User",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["email"] == "newuser@example.com"
|
|
assert data["username"] == "newuser"
|
|
assert data["display_name"] == "New User"
|
|
assert "user_id" in data
|
|
|
|
def test_register_with_invalid_email(self):
|
|
"""测试使用无效邮箱进行注册"""
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "invalid-email",
|
|
"password": "SecurePass123",
|
|
"username": "testuser",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
def test_register_with_weak_password(self):
|
|
"""测试使用弱密码进行注册"""
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "weak@example.com",
|
|
"password": "123", # Too short and simple
|
|
"username": "weakuser",
|
|
},
|
|
)
|
|
|
|
# Should fail validation or business logic
|
|
assert response.status_code in [400, 422]
|
|
|
|
def test_register_duplicate_email(self):
|
|
"""测试重复邮箱注册"""
|
|
# First registration
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "duplicate@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "user1",
|
|
"display_name": "User 1",
|
|
},
|
|
)
|
|
|
|
# Second registration with same email
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "duplicate@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "user2",
|
|
"display_name": "User 2",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert "already" in response.json()["detail"].lower() or "exists" in response.json()["detail"].lower()
|
|
|
|
|
|
class TestUserLogin:
|
|
"""用户登录集成测试"""
|
|
|
|
def setup_method(self):
|
|
"""每个测试前的准备：注册用户"""
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "loginuser@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "loginuser",
|
|
"display_name": "Login User",
|
|
},
|
|
)
|
|
|
|
def test_login_with_correct_credentials(self):
|
|
"""测试使用正确凭据登录"""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "loginuser@example.com",
|
|
"password": "SecurePass123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["email"] == "loginuser@example.com"
|
|
|
|
def test_login_with_wrong_password(self):
|
|
"""测试使用错误密码登录"""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "loginuser@example.com",
|
|
"password": "WrongPassword123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "error" in response.json() or "detail" in response.json()
|
|
|
|
def test_login_with_nonexistent_email(self):
|
|
"""测试使用不存在的邮箱登录"""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "nonexistent@example.com",
|
|
"password": "AnyPassword123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
def test_login_case_insensitive_email(self):
|
|
"""测试邮箱大小写不敏感登录"""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "LOGINUSER@EXAMPLE.COM", # Uppercase email
|
|
"password": "SecurePass123",
|
|
},
|
|
)
|
|
|
|
# Should still work because email is normalized
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestTokenRefresh:
|
|
"""令牌刷新集成测试"""
|
|
|
|
def setup_method(self):
|
|
"""每个测试前的准备：注册并登录获取令牌"""
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "refresh@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "refreshuser",
|
|
"display_name": "Refresh User",
|
|
},
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "refresh@example.com",
|
|
"password": "SecurePass123",
|
|
},
|
|
)
|
|
self.refresh_token = response.json().get("refresh_token")
|
|
self.access_token = response.json().get("access_token")
|
|
|
|
def test_refresh_token_success(self):
|
|
"""测试成功刷新令牌"""
|
|
if not self.refresh_token:
|
|
pytest.skip("Refresh token not available")
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/refresh",
|
|
json={"refresh_token": self.refresh_token},
|
|
)
|
|
|
|
# If refresh endpoint exists
|
|
if response.status_code != 404:
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
|
|
|
|
class TestCurrentUser:
|
|
"""当前用户信息集成测试"""
|
|
|
|
def setup_method(self):
|
|
"""每个测试前的准备：注册并登录获取令牌"""
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "me@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "meuser",
|
|
"display_name": "Me User",
|
|
},
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
json={
|
|
"email": "me@example.com",
|
|
"password": "SecurePass123",
|
|
},
|
|
)
|
|
self.token = response.json()["access_token"]
|
|
self.headers = {"Authorization": f"Bearer {self.token}"}
|
|
|
|
def test_get_current_user_success(self):
|
|
"""测试获取当前用户信息成功"""
|
|
response = client.get("/api/v1/auth/me", headers=self.headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["email"] == "me@example.com"
|
|
assert data["username"] == "meuser"
|
|
assert "user_id" in data
|
|
|
|
def test_get_current_user_without_token(self):
|
|
"""测试无令牌获取当前用户信息"""
|
|
response = client.get("/api/v1/auth/me")
|
|
|
|
assert response.status_code == 403
|
|
|
|
def test_get_current_user_with_invalid_token(self):
|
|
"""测试使用无效令牌获取当前用户信息"""
|
|
response = client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": "Bearer invalid-token"},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
class TestPasswordReset:
|
|
"""密码重置集成测试"""
|
|
|
|
def test_request_password_reset_success(self):
|
|
"""测试请求密码重置成功"""
|
|
# Register user first
|
|
client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"email": "reset@example.com",
|
|
"password": "SecurePass123",
|
|
"username": "resetuser",
|
|
},
|
|
)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/password/forgot",
|
|
json={"email": "reset@example.com"},
|
|
)
|
|
|
|
# Should return 202 Accepted (even if email not sent)
|
|
assert response.status_code == 202
|
|
|
|
def test_request_password_reset_nonexistent_user(self):
|
|
"""测试请求不存在的用户密码重置"""
|
|
response = client.post(
|
|
"/api/v1/auth/password/forgot",
|
|
json={"email": "nonexistent@example.com"},
|
|
)
|
|
|
|
# Should still return 202 for security (don't reveal if email exists)
|
|
assert response.status_code == 202
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|