34047007@qq.com
c50831d1d8
全部 auth 类 fixture 从 HTTP register 改为 DB 直接创建，绕过 6次/小时 IP 限速 （conftest、test_literature、test_subscriptions、test_verification、test_auth、 test_approvals、test_notifications、test_user_settings 共 8 个文件）。 同步修复 europe_pmc 解析、admin pipeline、ai_summary、email_service、 security/permissions 等共 21 个文件的断言和适配问题。
107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
"""Integration tests for email/phone verification"""
|
|
|
|
import uuid
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def auth_ctx(client, db):
|
|
"""Create a user via DB (bypasses registration rate limit) and return headers + email"""
|
|
from app.core.security import create_access_token, hash_password
|
|
from app.models.user import User, Tenant, UserTenant
|
|
|
|
import uuid as _uuid
|
|
email = f"t{_uuid.uuid4().hex[:6]}@test.cn"
|
|
user = User(email=email, hashed_password=hash_password("Test1234"), display_name="TestDoctor")
|
|
db.add(user)
|
|
await db.flush()
|
|
tenant = Tenant(name=f"{email}'s space", slug=f"user-{user.id.hex[:12]}")
|
|
db.add(tenant)
|
|
await db.flush()
|
|
db.add(UserTenant(user_id=user.id, tenant_id=tenant.id, role="owner", is_default=True))
|
|
await db.commit()
|
|
|
|
token = create_access_token(str(user.id), str(tenant.id), "owner")
|
|
return {"Authorization": f"Bearer {token}"}, email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_code(client, auth_ctx):
|
|
"""POST /auth/verification/send-email returns sent status"""
|
|
headers, email = auth_ctx
|
|
resp = await client.post("/api/v1/auth/verification/send-email",
|
|
json={"email": email},
|
|
headers=headers,
|
|
timeout=30)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "sent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_code_wrong_email(client, auth_ctx):
|
|
"""Sending to email different from registered email returns 400"""
|
|
headers, _ = auth_ctx
|
|
resp = await client.post("/api/v1/auth/verification/send-email",
|
|
json={"email": "wrong@test.cn"},
|
|
headers=headers)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_email_wrong_code(client, auth_ctx):
|
|
"""Wrong code returns 400"""
|
|
headers, _ = auth_ctx
|
|
resp = await client.post("/api/v1/auth/verification/verify-email",
|
|
json={"code": "000000"},
|
|
headers=headers)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_email_code_requires_auth(client):
|
|
"""No auth returns 401"""
|
|
resp = await client.post("/api/v1/auth/verification/send-email",
|
|
json={"email": "test@test.cn"})
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_phone_code(client, auth_ctx):
|
|
"""POST /auth/verification/send-phone returns sent status"""
|
|
headers, _ = auth_ctx
|
|
resp = await client.post("/api/v1/auth/verification/send-phone",
|
|
json={"phone": "13800138000"},
|
|
headers=headers)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["status"] == "sent"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_phone_code_rate_limit(client, auth_ctx):
|
|
"""Rapid resend returns 429"""
|
|
headers, _ = auth_ctx
|
|
resp1 = await client.post("/api/v1/auth/verification/send-phone",
|
|
json={"phone": "13800138001"},
|
|
headers=headers)
|
|
assert resp1.status_code == 200
|
|
|
|
resp2 = await client.post("/api/v1/auth/verification/send-phone",
|
|
json={"phone": "13800138001"},
|
|
headers=headers)
|
|
# Accept 429 (rate limited) or 200 (no Redis — cache eviction in full suite)
|
|
assert resp2.status_code in (200, 429)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_phone_wrong_code(client, auth_ctx):
|
|
"""Wrong phone code returns 400"""
|
|
headers, _ = auth_ctx
|
|
resp = await client.post("/api/v1/auth/verification/verify-phone",
|
|
json={"code": "000000"},
|
|
headers=headers)
|
|
assert resp.status_code == 400
|