浅念
58 lines
1.7 KiB
Python
58 lines
1.7 KiB
Python
"""/api/analytics/health:E0-lite 冒烟。"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
|
|
from app.db.models import Customer, MsgDailyStats
|
|
|
|
# naive:updated_at 列是无时区 DateTime;fromisoformat 构造不触发 DTZ 检查
|
|
_SEEN = "2026-09-04T12:00:00"
|
|
|
|
|
|
async def test_health_empty(make_client):
|
|
async def seed(session):
|
|
...
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/analytics/health")
|
|
assert r.status_code == 200
|
|
body = r.json()
|
|
assert body["data_as_of"] is None
|
|
assert len(body["entities"]) == 6
|
|
assert all(e["last_seen"] is None for e in body["entities"])
|
|
|
|
|
|
async def test_health_with_stats(make_client):
|
|
async def seed(session):
|
|
session.add(
|
|
MsgDailyStats(
|
|
channel="wechat",
|
|
owner_account_id="wx1",
|
|
peer_kind="friend",
|
|
peer_id="t1",
|
|
day=20260904,
|
|
send_cnt=1,
|
|
recv_cnt=2,
|
|
last_ts=1756900000000,
|
|
last_is_send=False,
|
|
)
|
|
)
|
|
session.add(
|
|
Customer(
|
|
friend_id="f1",
|
|
channel="wechat",
|
|
updated_at=dt.datetime.fromisoformat(_SEEN),
|
|
)
|
|
)
|
|
await session.commit()
|
|
|
|
client = await make_client(seed)
|
|
r = await client.get("/api/analytics/health")
|
|
assert r.status_code == 200
|
|
assert r.json()["data_as_of"] == 1756900000000
|
|
entities = {e["table"]: e["last_seen"] for e in r.json()["entities"]}
|
|
# 前端契约:last_seen 必须是 ISO 字符串(fromisoformat 可解析)
|
|
assert dt.datetime.fromisoformat(entities["customer"]) == dt.datetime.fromisoformat(
|
|
_SEEN
|
|
)
|