ZL-Q
adb2b3bcc3
- 整合 18 个分散的 migration 文件为 5 个模块化文件 - settings.py 支持 .env.local 覆盖 .env - 移除 user schema 中未使用的 country 字段正则 - 更新 profile protocol 文档移除 country 字段 - pyproject.toml 添加 ruff 到 dev 依赖 - 简化 integration test conftest 邮箱 fixture
94 lines
2.6 KiB
Python
94 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
|
|
import httpx
|
|
import pytest
|
|
from sqlalchemy import text
|
|
|
|
from core.config.settings import config
|
|
from core.db.session import AsyncSessionLocal
|
|
|
|
|
|
def pytest_configure(config): # noqa: ARG001
|
|
config.addinivalue_line(
|
|
"markers", "integration: integration test requiring live backend"
|
|
)
|
|
|
|
|
|
def pytest_collection_modifyitems(items):
|
|
for item in items:
|
|
if "integration" in item.nodeid:
|
|
item.add_marker(pytest.mark.integration)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def api_base_url() -> str:
|
|
return os.environ.get("ERYAO_TEST_BASE_URL", "http://localhost:5775")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def test_verify_code() -> str:
|
|
return os.environ.get("ERYAO_TEST__CODE", "123456")
|
|
|
|
|
|
@pytest.fixture
|
|
def test_email() -> str:
|
|
return os.environ.get("ERYAO_TEST__EMAIL", "test@example.com").strip().lower()
|
|
|
|
|
|
@pytest.fixture
|
|
def test_identity(test_email: str, test_verify_code: str) -> dict[str, str]:
|
|
return {"email": test_email, "code": test_verify_code}
|
|
|
|
|
|
@pytest.fixture
|
|
async def api_client(api_base_url: str) -> AsyncIterator[httpx.AsyncClient]:
|
|
async with httpx.AsyncClient(base_url=api_base_url, timeout=30.0) as client:
|
|
try:
|
|
health = await client.get("/health")
|
|
if health.status_code != 200:
|
|
pytest.skip(f"API not ready: /health={health.status_code}")
|
|
except Exception as exc:
|
|
pytest.skip(f"API unavailable: {exc}")
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_cleanup() -> AsyncIterator[list[str]]:
|
|
emails: list[str] = []
|
|
yield emails
|
|
|
|
if not emails:
|
|
return
|
|
|
|
hmac_key = config.points_policy.register_bonus_hmac_key.get_secret_value().strip()
|
|
email_hashes = [
|
|
hmac.new(
|
|
hmac_key.encode("utf-8"), email.encode("utf-8"), hashlib.sha256
|
|
).hexdigest()
|
|
for email in emails
|
|
]
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
await session.execute(
|
|
text(
|
|
"DELETE FROM points_audit_ledger WHERE lower(coalesce(user_email_snapshot, '')) = ANY(:emails)"
|
|
),
|
|
{"emails": emails},
|
|
)
|
|
await session.execute(
|
|
text(
|
|
"DELETE FROM register_bonus_claims WHERE email_hash = ANY(:email_hashes)"
|
|
),
|
|
{"email_hashes": email_hashes},
|
|
)
|
|
await session.execute(
|
|
text("DELETE FROM auth.users WHERE lower(email) = ANY(:emails)"),
|
|
{"emails": emails},
|
|
)
|
|
await session.commit()
|