TradeMate Dev
c6206787da
项目结构: - backend/ Python FastAPI 后端 - uni-app/ uni-app跨端前端 - docs/ 设计文档 - docker-compose.yml Docker编排 - nginx/scripts/systemd 运维配置 已完成功能: - 用户认证 (JWT) - 智能翻译 + 回复建议 - 营销素材生成 - 客户管理 + 沉默检测 - 报价单管理 - 产品库管理 - 汇率换算 - 推送通知 (uni-push) - WhatsApp Webhook框架 - Celery定时任务
94 lines
3.2 KiB
Python
94 lines
3.2 KiB
Python
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
|
|
class TestAuthAPI:
|
|
async def test_health_endpoint(self, client: AsyncClient):
|
|
response = await client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert data["app"] == "TradeMate"
|
|
|
|
async def test_register_new_user(self, client: AsyncClient):
|
|
response = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"phone": "13900139001",
|
|
"password": "test123456",
|
|
"username": "newuser",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["phone"] == "13900139001"
|
|
assert data["username"] == "newuser"
|
|
assert data["tier"] == "free"
|
|
|
|
async def test_register_duplicate_phone(self, client: AsyncClient, test_user):
|
|
response = await client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"phone": "13800138000",
|
|
"password": "test123456",
|
|
"username": "duplicate",
|
|
},
|
|
)
|
|
assert response.status_code == 400
|
|
assert "already registered" in response.json()["detail"]
|
|
|
|
async def test_login_success(self, client: AsyncClient, test_user):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
data={
|
|
"username": "13800138000",
|
|
"password": "test123456",
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
async def test_login_wrong_password(self, client: AsyncClient, test_user):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
data={
|
|
"username": "13800138000",
|
|
"password": "wrongpassword",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
async def test_login_nonexistent_user(self, client: AsyncClient):
|
|
response = await client.post(
|
|
"/api/v1/auth/login",
|
|
data={
|
|
"username": "13999999999",
|
|
"password": "test123456",
|
|
},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
async def test_get_current_user(self, client: AsyncClient, auth_headers):
|
|
response = await client.get("/api/v1/auth/me", headers=auth_headers)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["phone"] == "13800138000"
|
|
assert data["username"] == "test_user"
|
|
|
|
async def test_get_user_unauthorized(self, client: AsyncClient):
|
|
response = await client.get("/api/v1/auth/me")
|
|
assert response.status_code == 401
|
|
|
|
async def test_refresh_token(self, client: AsyncClient, test_user):
|
|
from app.core.security import create_refresh_token
|
|
refresh = create_refresh_token({"sub": str(test_user.id)})
|
|
|
|
response = await client.post(
|
|
"/api/v1/auth/refresh",
|
|
json={"refresh_token": refresh},
|
|
)
|
|
assert response.status_code == 200
|
|
assert "access_token" in response.json() |