xsl
64175a1ea1
ob_event 落库通道：形状校验 + props 字节上限 + client_ts 合理性 + PII 键剔除，整批原子（FR-SYS-06 服务端部分）。 实现：packages/analytics/schemas.py（EventIn/EventBatchRequest/ EventBatchResponse + 事件名形状正则常量）；service.py 拆出纯函数 validate_batch（批量/props 字节/事件名形状/client_ts 窗口/PII 剔除， 任一不过整批 40020），AnalyticsService.record 组装 ObEvent 逐条 add + 一次 flush；routers/events.py 提供 POST /events/batch （/api/v1 与 /v1 别名，认证必需）。 测试：单测 tests/unit/test_events_validation.py 8 例（合法批、超批量、 props 超字节/不可序列化、事件名形状含尾随换行、client_ts 双边界、 naive 时区按 UTC、PII 键含大小写变体剔除）；集成 tests/integration/test_events.py 3 例（两前缀可用+落库断言、 非法批 40020 且库无半条、无 token 40100）。门禁全绿：ruff check/format、mypy（T2 列表 + domain strict）、lint-imports 4 契约、 单测 177 通过、集成 6 通过。 偏差说明：event_name 正则未写成 Field(pattern=...)，而是作为 EVENT_NAME_PATTERN 常量放 schemas.py、由服务层 validate_batch 强制—— 若挂在请求体校验上，非法事件名会先被 RequestValidationError 拦截、 返回通用 40001/422 信封，无法满足任务契约 §4「大写 event_name → 40020」的集成行为。正则本体与文档一致。
138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
"""埋点上报集成用例（真实 PostgreSQL，FR-SYS-06）。
|
||
|
||
覆盖：/v1 与 /api/v1 两个前缀、落库行内容（含 PII 剔除）、
|
||
非法批 40020 且整批原子（库里没有半条）、无 token 40100。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
|
||
def _infra_available() -> bool:
|
||
return bool(os.getenv("MEIDA_RUN_INTEGRATION"))
|
||
|
||
|
||
pytestmark = [
|
||
pytest.mark.integration,
|
||
pytest.mark.skipif(
|
||
not _infra_available(),
|
||
reason="set MEIDA_RUN_INTEGRATION=1 with PostgreSQL and Redis running",
|
||
),
|
||
]
|
||
|
||
|
||
@pytest.fixture
|
||
async def client() -> Any:
|
||
import httpx
|
||
|
||
from apps.api.main import create_app
|
||
from packages.shared.config import get_settings
|
||
|
||
# 埋点链路不碰 LLM，无需替换 provider；lifespan 照常起网关与连接池
|
||
app = create_app(get_settings())
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with (
|
||
httpx.AsyncClient(transport=transport, base_url="http://test") as http,
|
||
app.router.lifespan_context(app),
|
||
):
|
||
yield http
|
||
|
||
|
||
@pytest.fixture
|
||
async def auth(client: Any) -> dict[str, Any]:
|
||
"""dev-token 换一个专属测试用户，避免与其它用例的 ob_event 行串号。"""
|
||
resp = await client.post(
|
||
"/api/v1/auth/dev-token", json={"device_key": f"t3-{uuid.uuid4().hex[:8]}"}
|
||
)
|
||
assert resp.json()["code"] == 0
|
||
data = resp.json()["data"]
|
||
headers = {"Authorization": f"Bearer {data['access_token']}"}
|
||
return {"headers": headers, "user_id": int(data["user_id"])}
|
||
|
||
|
||
async def _count_events(user_id: int) -> int:
|
||
from sqlalchemy import func, select
|
||
|
||
from packages.shared.db import get_session_factory
|
||
from packages.shared.models import ObEvent
|
||
|
||
async with get_session_factory()() as session:
|
||
stmt = select(func.count()).select_from(ObEvent).where(ObEvent.user_id == user_id)
|
||
return int(await session.scalar(stmt))
|
||
|
||
|
||
async def _fetch_events(user_id: int) -> list[Any]:
|
||
from sqlalchemy import select
|
||
|
||
from packages.shared.db import get_session_factory
|
||
from packages.shared.models import ObEvent
|
||
|
||
async with get_session_factory()() as session:
|
||
result = await session.execute(
|
||
select(ObEvent).where(ObEvent.user_id == user_id).order_by(ObEvent.event_id)
|
||
)
|
||
return list(result.scalars().all())
|
||
|
||
|
||
async def test_batch_lands_two_rows_with_pii_stripped(client: Any, auth: dict[str, Any]) -> None:
|
||
resp = await client.post(
|
||
"/v1/events/batch",
|
||
json={
|
||
"events": [
|
||
{
|
||
"event_name": "page_view",
|
||
"props": {"page": "home", "phone": "13800001234"},
|
||
"client_ts": datetime.now(UTC).isoformat(),
|
||
},
|
||
{"event_name": "outfit_card.tap", "props": {"card_id": 42}},
|
||
]
|
||
},
|
||
headers=auth["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["code"] == 0
|
||
assert body["data"] == {"accepted": 2}
|
||
|
||
rows = await _fetch_events(auth["user_id"])
|
||
assert len(rows) == 2
|
||
assert rows[0].event_name == "page_view"
|
||
assert rows[0].props == {"page": "home"} # phone 已在服务端剔除
|
||
assert rows[0].client_ts is not None
|
||
assert rows[0].server_ts is not None
|
||
assert rows[1].event_name == "outfit_card.tap"
|
||
assert rows[1].props == {"card_id": 42}
|
||
assert rows[1].client_ts is None
|
||
|
||
|
||
async def test_invalid_batch_is_rejected_atomically(client: Any, auth: dict[str, Any]) -> None:
|
||
before = await _count_events(auth["user_id"])
|
||
|
||
resp = await client.post(
|
||
"/api/v1/events/batch",
|
||
json={
|
||
"events": [
|
||
{"event_name": "page_view"},
|
||
{"event_name": "PageView"}, # 大写 → 形状非法
|
||
]
|
||
},
|
||
headers=auth["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert resp.json()["code"] == 40020
|
||
|
||
# 整批原子：合法的那条也不能半落库
|
||
assert await _count_events(auth["user_id"]) == before
|
||
|
||
|
||
async def test_events_require_authentication(client: Any) -> None:
|
||
resp = await client.post("/api/v1/events/batch", json={"events": [{"event_name": "page_view"}]})
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|