381 lines
13 KiB
Python
381 lines
13 KiB
Python
"""admin 用户行为看板集成用例（真实 PostgreSQL，FR-ADM 行为分析能力）。
|
||
|
||
覆盖四个端点的字段与分页、X-Admin-Token 鉴权口径与现有 admin 一致
|
||
（错误令牌 → 40100）、union_id 掩码（前 4 后 4，防 PII 直出）、配额快照
|
||
只读不造行。种子数据用专属随机 user_id，断言全部相对种子而非全表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
from datetime import UTC, datetime, timedelta
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
ADMIN_TOKEN = "it-admin-token-behavior"
|
||
AUTH = {"X-Admin-Token": ADMIN_TOKEN}
|
||
BAD_AUTH = {"X-Admin-Token": "wrong-token"}
|
||
|
||
_SEED_EVENT_NAMES = ("page_view", "outfit_card.tap")
|
||
|
||
|
||
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(autouse=True)
|
||
async def _dispose_engine_after_test() -> AsyncIterator[None]:
|
||
"""用例结束即释放全局引擎（理由同 test_admin_console.py）。"""
|
||
yield
|
||
from packages.shared.db import dispose_engine
|
||
|
||
await dispose_engine()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client() -> Any:
|
||
import httpx
|
||
|
||
from apps.admin.app import create_admin_app
|
||
from packages.shared.config import Settings
|
||
|
||
app = create_admin_app(Settings(admin_token=ADMIN_TOKEN))
|
||
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 seed() -> dict[str, Any]:
|
||
"""造 2 个用户 + 身份 + 事件（7 天内/外各一）+ 报告 + 配额规则与用量。"""
|
||
from sqlalchemy import select
|
||
|
||
from packages.domain.billing.quota import period_key
|
||
from packages.domain.ids import next_id
|
||
from packages.shared.db import get_session_factory
|
||
from packages.shared.models import (
|
||
ObEvent,
|
||
Quota,
|
||
QuotaRule,
|
||
RpReport,
|
||
User,
|
||
UserIdentity,
|
||
)
|
||
|
||
suffix = uuid.uuid4().hex[:8]
|
||
user_a, user_b = next_id(), next_id()
|
||
union_id_a = f"email:it-user-{suffix}@example.com"
|
||
now = datetime.now(UTC)
|
||
quota_key = f"it-quota-{suffix}"
|
||
day_key = period_key(now, "day")
|
||
|
||
async with get_session_factory()() as session:
|
||
session.add_all(
|
||
[
|
||
User(
|
||
user_id=user_a,
|
||
status="active",
|
||
nickname=f"it-a-{suffix}",
|
||
register_source="email",
|
||
created_at=now - timedelta(days=3),
|
||
),
|
||
User(
|
||
user_id=user_b,
|
||
status="active",
|
||
nickname=f"it-b-{suffix}",
|
||
register_source="dev",
|
||
created_at=now - timedelta(days=1),
|
||
),
|
||
UserIdentity(
|
||
identity_id=next_id(),
|
||
user_id=user_a,
|
||
provider="email",
|
||
union_id=union_id_a,
|
||
),
|
||
ObEvent(
|
||
event_id=next_id(),
|
||
user_id=user_a,
|
||
event_name=_SEED_EVENT_NAMES[0],
|
||
props={"page": "home"},
|
||
server_ts=now - timedelta(hours=2),
|
||
),
|
||
ObEvent(
|
||
event_id=next_id(),
|
||
user_id=user_a,
|
||
event_name=_SEED_EVENT_NAMES[1],
|
||
props={"card_id": 1},
|
||
server_ts=now - timedelta(days=1),
|
||
),
|
||
# 8 天前：在 7 天窗口外——event_count_7d 不该算它
|
||
ObEvent(
|
||
event_id=next_id(),
|
||
user_id=user_a,
|
||
event_name=_SEED_EVENT_NAMES[0],
|
||
props={"page": "old"},
|
||
server_ts=now - timedelta(days=8),
|
||
),
|
||
ObEvent(
|
||
event_id=next_id(),
|
||
user_id=user_b,
|
||
event_name=_SEED_EVENT_NAMES[0],
|
||
props={"page": "home"},
|
||
server_ts=now - timedelta(hours=1),
|
||
),
|
||
RpReport(
|
||
report_id=next_id(),
|
||
user_id=user_a,
|
||
tier="basic",
|
||
status="ready",
|
||
version=1,
|
||
),
|
||
QuotaRule(
|
||
quota_key=quota_key,
|
||
period_type="day",
|
||
base_limit=10,
|
||
task_bonus_max=0,
|
||
hard_cap=20,
|
||
),
|
||
Quota(
|
||
user_id=user_a,
|
||
quota_key=quota_key,
|
||
period_key=day_key,
|
||
used=3,
|
||
limit_value=10,
|
||
bonus_granted=0,
|
||
),
|
||
]
|
||
)
|
||
await session.commit()
|
||
rules_before = len(
|
||
(await session.execute(select(QuotaRule).where(QuotaRule.quota_key == quota_key)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
return {
|
||
"user_a": user_a,
|
||
"user_b": user_b,
|
||
"quota_key": quota_key,
|
||
"day_key": day_key,
|
||
"rules_before": rules_before,
|
||
"union_id_a": union_id_a,
|
||
}
|
||
|
||
|
||
async def test_wrong_token_rejected_like_existing_admin(client: Any) -> None:
|
||
"""鉴权口径与现有 admin 一致：错误令牌 → 40100。"""
|
||
resp = await client.get("/admin/api/v1/users", headers=BAD_AUTH)
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
|
||
resp = await client.get("/admin/api/v1/events", headers=BAD_AUTH)
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
|
||
resp = await client.get("/admin/api/v1/stats/overview", headers=BAD_AUTH)
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
|
||
# 无令牌头：同样 40100（compare_digest 对空串，不炸 500）
|
||
resp = await client.get("/admin/api/v1/users")
|
||
assert resp.status_code == 401
|
||
assert resp.json()["code"] == 40100
|
||
|
||
|
||
async def test_users_list_fields_and_filters(client: Any, seed: dict[str, Any]) -> None:
|
||
resp = await client.get("/admin/api/v1/users", headers=AUTH, params={"query": "email"})
|
||
assert resp.status_code == 200
|
||
body = resp.json()["data"]
|
||
assert body["page"] == 1
|
||
assert body["page_size"] == 20
|
||
assert body["total"] >= 1
|
||
assert all(item["register_source"] == "email" for item in body["items"])
|
||
|
||
# query 按 user_id 精确：只命中种子用户 A——字段契约全在它身上断言，
|
||
# 不受库里其它用户数据/分页位置影响
|
||
resp = await client.get(
|
||
"/admin/api/v1/users", headers=AUTH, params={"query": str(seed["user_a"])}
|
||
)
|
||
body = resp.json()["data"]
|
||
assert [item["user_id"] for item in body["items"]] == [str(seed["user_a"])]
|
||
row_a = body["items"][0]
|
||
assert row_a["register_source"] == "email"
|
||
assert row_a["created_at"] is not None
|
||
assert row_a["last_active_at"] is not None
|
||
assert row_a["event_count_7d"] == 2 # 8 天外那条不算
|
||
assert row_a["report_count"] == 1
|
||
|
||
|
||
async def test_users_list_pagination(client: Any, seed: dict[str, Any]) -> None:
|
||
resp = await client.get(
|
||
"/admin/api/v1/users",
|
||
headers=AUTH,
|
||
params={"query": "dev", "page": 1, "page_size": 1},
|
||
)
|
||
body = resp.json()["data"]
|
||
assert body["page_size"] == 1
|
||
assert len(body["items"]) == 1
|
||
assert body["total"] >= 1
|
||
|
||
|
||
async def test_user_detail_masks_identity_and_snapshots_quota(
|
||
client: Any, seed: dict[str, Any]
|
||
) -> None:
|
||
from sqlalchemy import select
|
||
|
||
from packages.shared.db import get_session_factory
|
||
from packages.shared.models import Quota, UserIdentity
|
||
|
||
user_id = seed["user_a"]
|
||
resp = await client.get(f"/admin/api/v1/users/{user_id}", headers=AUTH)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
|
||
# 基本信息
|
||
assert data["user"]["user_id"] == str(user_id)
|
||
assert data["user"]["register_source"] == "email"
|
||
|
||
# 身份：provider 明文、union_id 只留前 4 后 4（PII 直出防护，逐字符断言）
|
||
assert len(data["identities"]) == 1
|
||
ident = data["identities"][0]
|
||
assert ident["provider"] == "email"
|
||
raw = seed["union_id_a"]
|
||
expected = f"{raw[:4]}{'*' * (len(raw) - 8)}{raw[-4:]}"
|
||
assert ident["union_id"] == expected
|
||
assert raw not in str(data["identities"])
|
||
|
||
# 配额快照：种子用量 used=3，limit=base 10，remaining=7
|
||
quota_rows = [q for q in data["quotas"] if q["quota_key"] == seed["quota_key"]]
|
||
assert len(quota_rows) == 1
|
||
assert quota_rows[0]["used"] == 3
|
||
assert quota_rows[0]["limit"] == 10
|
||
assert quota_rows[0]["remaining"] == 7
|
||
assert quota_rows[0]["period_key"] == seed["day_key"]
|
||
|
||
# 只读断言：看板不产生写副作用（身份与计量行数不变）
|
||
async with get_session_factory()() as session:
|
||
idents = (
|
||
(await session.execute(select(UserIdentity).where(UserIdentity.user_id == user_id)))
|
||
.scalars()
|
||
.all()
|
||
)
|
||
quotas = (
|
||
(
|
||
await session.execute(
|
||
select(Quota).where(
|
||
Quota.user_id == user_id, Quota.quota_key == seed["quota_key"]
|
||
)
|
||
)
|
||
)
|
||
.scalars()
|
||
.all()
|
||
)
|
||
assert len(idents) == 1
|
||
assert len(quotas) == 1
|
||
assert quotas[0].used == 3
|
||
|
||
# 最近事件/报告：各落一条种子数据
|
||
assert len(data["recent_events"]) == 3 # 2 条 7 天内 + 1 条 8 天前（窗口不按详情截）
|
||
assert data["recent_events"][0]["server_ts"] >= data["recent_events"][-1]["server_ts"]
|
||
assert len(data["recent_reports"]) == 1
|
||
report = data["recent_reports"][0]
|
||
assert report["tier"] == "basic"
|
||
assert report["status"] == "ready"
|
||
assert report["version"] == 1
|
||
assert report["report_id"]
|
||
|
||
|
||
async def test_user_detail_not_found(client: Any) -> None:
|
||
resp = await client.get("/admin/api/v1/users/0", headers=AUTH)
|
||
assert resp.status_code == 404
|
||
assert resp.json()["code"] == 40400
|
||
|
||
|
||
async def test_events_stream_filters_and_pagination(client: Any, seed: dict[str, Any]) -> None:
|
||
user_id = seed["user_a"]
|
||
# user_id 过滤：只出种子用户的事件，server_ts 倒序
|
||
resp = await client.get(
|
||
"/admin/api/v1/events",
|
||
headers=AUTH,
|
||
params={"user_id": user_id, "page_size": 2, "page": 1},
|
||
)
|
||
assert resp.status_code == 200
|
||
body = resp.json()["data"]
|
||
assert body["total"] == 3
|
||
assert len(body["items"]) == 2
|
||
assert all(item["user_id"] == str(user_id) for item in body["items"])
|
||
first, second = body["items"][0]["server_ts"], body["items"][1]["server_ts"]
|
||
assert first >= second
|
||
|
||
# 翻第二页拿到最早那条（8 天前），证明分页生效
|
||
resp = await client.get(
|
||
"/admin/api/v1/events",
|
||
headers=AUTH,
|
||
params={"user_id": user_id, "page_size": 2, "page": 2},
|
||
)
|
||
page2 = resp.json()["data"]
|
||
assert len(page2["items"]) == 1
|
||
|
||
# event_name 过滤 + ISO 时间范围：8 天前那条被 start 挡在窗外
|
||
start = (datetime.now(UTC) - timedelta(days=7)).isoformat()
|
||
resp = await client.get(
|
||
"/admin/api/v1/events",
|
||
headers=AUTH,
|
||
params={"user_id": user_id, "event_name": _SEED_EVENT_NAMES[0], "start": start},
|
||
)
|
||
body = resp.json()["data"]
|
||
assert body["total"] == 1
|
||
assert body["items"][0]["props"] == {"page": "home"}
|
||
|
||
# epoch 秒也接受（与 ISO 等价窗口）
|
||
epoch_start = str(int((datetime.now(UTC) - timedelta(days=7)).timestamp()))
|
||
resp = await client.get(
|
||
"/admin/api/v1/events",
|
||
headers=AUTH,
|
||
params={"user_id": user_id, "event_name": _SEED_EVENT_NAMES[0], "start": epoch_start},
|
||
)
|
||
assert resp.json()["data"]["total"] == 1
|
||
|
||
# page_size 超上限 → 422（契约上限 200）
|
||
resp = await client.get("/admin/api/v1/events", headers=AUTH, params={"page_size": 201})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
async def test_stats_overview_pivot_and_series(client: Any, seed: dict[str, Any]) -> None:
|
||
resp = await client.get("/admin/api/v1/stats/overview", headers=AUTH, params={"days": 7})
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["days"] == 7
|
||
assert data["registered_total"] >= 2
|
||
|
||
# 透视表：种子用户的两个事件名都出现，PV 至少为种子条数
|
||
pv = {(row["event_name"]) for row in data["daily_event_pv"]}
|
||
assert _SEED_EVENT_NAMES[0] in pv
|
||
assert _SEED_EVENT_NAMES[1] in pv
|
||
page_view_pv = sum(
|
||
row["pv"] for row in data["daily_event_pv"] if row["event_name"] == _SEED_EVENT_NAMES[0]
|
||
)
|
||
assert page_view_pv >= 2 # 种子 A、B 各至少一条（库内其它数据只会更多）
|
||
|
||
# 活跃序列：至少一天有活跃用户
|
||
assert data["daily_active_users"]
|
||
assert max(row["active_users"] for row in data["daily_active_users"]) >= 1
|
||
|
||
# days 超上限 → 422
|
||
resp = await client.get("/admin/api/v1/stats/overview", headers=AUTH, params={"days": 91})
|
||
assert resp.status_code == 422
|