qzl
55bac03eb0
后端 Pydantic schema 添加 Hant 繁体字段支持（guaNameHant, targetGuaNameHant, spiritNameHant, relationNameHant），解决 DerivedDivinationData extra=forbid 拒绝 AI 输出的繁体字段导致 agent_output 解析失败、历史记录为空的问题。 六爻解卦提示词修复：增加静卦五行生克链分析、假破假空降权、 六冲中性判断、用神核对防捏造、空亡数据对照等硬约束。 前端 Dart model 同步添加 Hant 字段（反向兼容，缺省为空字符串）。 其他：硬币翻转动画修复、弹窗单按钮居中、起卦按钮布局调整、 繁体 l10n 清理、pre-commit 排除集成测试。
100 lines
2.8 KiB
Python
100 lines
2.8 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
import hashlib
|
|
import hmac
|
|
import os
|
|
import time
|
|
|
|
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 unique_test_email() -> str:
|
|
base_email = os.environ.get("ERYAO_TEST__EMAIL", "test@example.com").strip().lower()
|
|
if "@" in base_email:
|
|
name, domain = base_email.split("@", 1)
|
|
else:
|
|
name, domain = base_email, "example.com"
|
|
return f"{name}+it{int(time.time() * 1000)}@{domain}"
|
|
|
|
|
|
@pytest.fixture
|
|
def test_identity(unique_test_email: str, test_verify_code: str) -> dict[str, str]:
|
|
return {"email": unique_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()
|