88 lines
2.5 KiB
Python
88 lines
2.5 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
|
||
|
|
|
||
|
|
|
||
|
|
@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()
|