浅念
6839cb94d2
- I-1 站点停用全链接线: Site 数据类加 status 字段; DB 源 _load_one 与 文件源 read_tenant 均只输出 active 站点(停用站点域名不进 registry 索引 → 该域名 404;全停用=合法运营状态输出空 sites;文件源无 status 键默认 active 兼容)。E2E: PATCH 站点 disabled → 其域名 404、复启用 恢复;源级直测两源过滤。 - I-3 prefix 跨租户查重补两口: db.other_push_prefixes 共用助手 (MAX(version) 逐租户当前前缀,排除自身);创建向导(出厂档案前缀被占 → 400 提示改 blank/改占用方)与 reset push(重置回出厂撞车 → 400 提示 走 PUT)接入;PUT 原有查重改用同一助手。rollback 维持豁免(历史不可变 语义优先,取舍清单记档)。既有测试第二租户改 blank profile 避撞。 - I-4 默认凭据警示: lifespan 启动检测 ADMIN_PASSWORD 仍为 Settings 字段 默认值(非硬编码第二份,防双源漂移)→ logger.warning 醒目提示; CLAUDE.md Phase 4 部署记档补 ADMIN_PASSWORD 生产要求。 - I-2/M-2 取舍补录: 密钥掩码未实现(GET config/settings 明文返回 api_key/smtp 密码,Phase 3 前须定 __SET__ 哨兵方案);站点管理权限现为 仅超管(Phase 3 对接时决策放开或改 spec);取舍清单第 2 条同步刷新 (rollback 现为唯一豁免口)。 全量回归: 571 passed(基线 566 + 新增 5)。 Co-Authored-By: Claude Code <noreply@anthropic.com>
480 lines
20 KiB
Python
480 lines
20 KiB
Python
"""数据视图 API(Phase 2 Task 8): 会话列表(筛选/分页)/详情/删除 + 推送日志。
|
|
|
|
环境: db_source_app/super_client(DB 源闭环,租户全部经创建 API 进入
|
|
registry)。查询语义自 jc 旧单租户 admin.py 移植(仅至少一条消息的会话进
|
|
列表/创建时间倒序/意向筛选/load_only 免大字段),路由形态多租户化。
|
|
|
|
种子纪律: 租户库五表(conversations/messages/customer_profiles/
|
|
evaluations/lead_push_logs)直接 ORM 插——独立 NullPool engine 直连同一
|
|
db 文件,一次 asyncio.run 内建表(create_all,与 registry ensure_schema
|
|
同构幂等)+插数+dispose,绝不碰 registry 的 pool=2 常驻连接(跨循环
|
|
纪律,同 test_tenant_admin_api 的 _db_fetch 模式)。
|
|
|
|
懒建触发: 经创建 API 进 registry 的租户从未被域名 resolve(engine 仍
|
|
None),数据视图首次查询必须经 ensure_ready 建库不炸(空列表 200,库文件
|
|
落地)——test_lazy_tenant_first_query_builds_db 钉住。
|
|
"""
|
|
import asyncio
|
|
from datetime import datetime, timezone
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.engine.models import (
|
|
Base,
|
|
Conversation,
|
|
ConversationSource,
|
|
ConversationStatus,
|
|
CustomerProfile,
|
|
Evaluation,
|
|
IntentLevel,
|
|
LeadPushLog,
|
|
Message,
|
|
MessageRole,
|
|
)
|
|
|
|
TENANTS = "/api/platform/tenants"
|
|
LOGIN = "/api/platform/auth/login"
|
|
CHANGE_PW = "/api/platform/auth/change-password"
|
|
UTC = timezone.utc
|
|
# 推送日志大字段唯一标记: 列表响应里绝不出现(payload lazy 钉子)
|
|
PAYLOAD_MARKER = "UNISERVICE-PAYLOAD-9f1c2d"
|
|
|
|
|
|
# ---- 助手(同 test_tenant_admin_api 的模式) ----------------------------------------
|
|
|
|
|
|
def _auth(client):
|
|
return {"Authorization": f"Bearer {client.platform_token}"}
|
|
|
|
|
|
def _bearer(token):
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def _create(client, slug="demo", admin=None, password=None, domains=None,
|
|
profile="jiancai"):
|
|
return client.post(TENANTS, headers=_auth(client), json={
|
|
"name": f"{slug} 租户",
|
|
"slug": slug,
|
|
"profile": profile,
|
|
"admin_username": admin or f"{slug}-admin",
|
|
"admin_password": password or f"{slug}-pass-1",
|
|
"domains": domains or [],
|
|
})
|
|
|
|
|
|
def _tenant_admin_token(client, username, old_password, new_password="tn-pass-1"):
|
|
r = client.post(LOGIN, json={"username": username, "password": old_password})
|
|
assert r.status_code == 200
|
|
token = r.json()["token"]
|
|
r = client.post(CHANGE_PW, headers=_bearer(token), json={
|
|
"old_password": old_password, "new_password": new_password})
|
|
assert r.status_code == 200
|
|
return token
|
|
|
|
|
|
# ---- 租户库直连(NullPool 跨循环纪律) ----------------------------------------------
|
|
|
|
|
|
def _tenant_engine(root, slug):
|
|
tenants = root / "tenants"
|
|
tenants.mkdir(parents=True, exist_ok=True)
|
|
return create_async_engine(
|
|
f"sqlite+aiosqlite:///{(tenants / f'{slug}.db').as_posix()}",
|
|
poolclass=NullPool)
|
|
|
|
|
|
def _seed(root, slug, populate):
|
|
"""直连租户库造数: create_all + populate(session) + commit,跑完 dispose。"""
|
|
|
|
async def run():
|
|
engine = _tenant_engine(root, slug)
|
|
try:
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
async with async_sessionmaker(engine, expire_on_commit=False)() as session:
|
|
await populate(session)
|
|
await session.commit()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def _count(root, slug, table):
|
|
"""直连租户库数行(删除级联断言用)。"""
|
|
|
|
async def run():
|
|
engine = _tenant_engine(root, slug)
|
|
try:
|
|
async with AsyncSession(engine) as session:
|
|
return (await session.execute(
|
|
select(func.count()).select_from(table))).scalar()
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
return asyncio.run(run())
|
|
|
|
|
|
# ---- 种子工厂 ---------------------------------------------------------------------
|
|
|
|
|
|
def _conv(cid, at, *, site="main", ip="127.0.0.1", lang=None,
|
|
status=ConversationStatus.completed,
|
|
source=ConversationSource.widget):
|
|
client_info = {"ip": ip, "user_agent": f"UA-{cid}"}
|
|
if site:
|
|
client_info["site"] = site
|
|
if lang:
|
|
client_info["lang"] = lang
|
|
return Conversation(
|
|
id=cid, session_id=f"sess-{cid}", status=status, source=source,
|
|
client_info=client_info, created_at=at, updated_at=at)
|
|
|
|
|
|
def _msg(cid, role, content, at):
|
|
return Message(conversation_id=cid, role=role, content=content, created_at=at)
|
|
|
|
|
|
# ---- 核心 4 项 --------------------------------------------------------------------
|
|
|
|
|
|
def test_conversations_list_with_filters(super_client, tmp_path):
|
|
_create(super_client, slug="demo")
|
|
|
|
async def populate(session):
|
|
at1 = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
|
|
at2 = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
|
|
at4 = datetime(2026, 2, 1, 8, 0, tzinfo=UTC)
|
|
# c1: es 站/interested/2 条消息/已留资已发邮件
|
|
session.add(_conv("c1", at1, site="es", ip="8.8.8.8", lang="en"))
|
|
session.add(_msg("c1", MessageRole.user, "您好，我想买切割机", at1))
|
|
session.add(_msg("c1", MessageRole.assistant, "您好！请问要切什么材料？", at1))
|
|
session.add(Evaluation(
|
|
conversation_id="c1", intent_level=IntentLevel.interested,
|
|
reasoning="明确采购意向", contact_info={"email": "buy@example.com"},
|
|
email_sent=True, evaluated_at=at1))
|
|
# c2: main 站/not_interested/1 条消息
|
|
session.add(_conv("c2", at2, site="main", ip="127.0.0.1"))
|
|
session.add(_msg("c2", MessageRole.user, "just browsing", at2))
|
|
session.add(Evaluation(
|
|
conversation_id="c2", intent_level=IntentLevel.not_interested,
|
|
contact_info={}, evaluated_at=at2))
|
|
# c3: 零消息——jc 语义: 不进列表(哪怕 intent 命中)
|
|
session.add(_conv("c3", at2, site="es"))
|
|
session.add(Evaluation(
|
|
conversation_id="c3", intent_level=IntentLevel.interested,
|
|
evaluated_at=at2))
|
|
# c4: 2 月的会话(日期区间排除用)
|
|
session.add(_conv("c4", at4, site="main"))
|
|
session.add(_msg("c4", MessageRole.user, "february lead", at4))
|
|
session.add(Evaluation(
|
|
conversation_id="c4", intent_level=IntentLevel.pending,
|
|
evaluated_at=at4))
|
|
|
|
_seed(tmp_path, "demo", populate)
|
|
|
|
def _list(query=""):
|
|
r = super_client.get(f"{TENANTS}/demo/conversations{query}",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200, r.text
|
|
return r.json()
|
|
|
|
# 默认: 只有带消息的会话,创建时间倒序
|
|
body = _list()
|
|
assert body["total"] == 3
|
|
assert [i["id"] for i in body["items"]] == ["c4", "c2", "c1"]
|
|
row = next(i for i in body["items"] if i["id"] == "c1")
|
|
assert row["session_id"] == "sess-c1"
|
|
assert row["status"] == "completed"
|
|
assert row["source"] == "widget"
|
|
assert row["site"] == "es"
|
|
assert row["ip"] == "8.8.8.8"
|
|
assert row["country"] == "美国" # ip_service 归国复用
|
|
assert row["intent_level"] == "interested"
|
|
assert row["reasoning"] == "明确采购意向"
|
|
assert row["contact_info"]["email"] == "buy@example.com"
|
|
assert row["email_sent"] is True
|
|
assert row["message_count"] == 2
|
|
assert row["preview"] == "您好，我想买切割机"
|
|
assert row["created_at"].startswith("2026-01-05")
|
|
row2 = next(i for i in body["items"] if i["id"] == "c2")
|
|
assert row2["country"] == "本地" # 127.0.0.1
|
|
assert row2["email_sent"] is False
|
|
|
|
# intent 筛选(c3 同意向但零消息,不得混入)
|
|
body = _list("?intent=interested")
|
|
assert body["total"] == 1
|
|
assert [i["id"] for i in body["items"]] == ["c1"]
|
|
|
|
# site 筛选(client_info.site)
|
|
body = _list("?site=es")
|
|
assert body["total"] == 1
|
|
assert [i["id"] for i in body["items"]] == ["c1"]
|
|
|
|
# 日期区间,边界含: from=精确时刻 → c1(等于边界)在内
|
|
body = _list("?from=2026-01-05T12:00:00")
|
|
assert body["total"] == 3
|
|
# to=日期(整天含) → 只 c1(12:00 ≤ 当日 23:59:59.999999),c2 次日出局
|
|
body = _list("?to=2026-01-05")
|
|
assert [i["id"] for i in body["items"]] == ["c1"]
|
|
# 闭区间 [01-06, 01-06] → 只 c2
|
|
body = _list("?from=2026-01-06&to=2026-01-06")
|
|
assert [i["id"] for i in body["items"]] == ["c2"]
|
|
# 跨月区间排除 c4
|
|
body = _list("?from=2026-01-01&to=2026-01-31")
|
|
assert [i["id"] for i in body["items"]] == ["c2", "c1"]
|
|
|
|
# 分页
|
|
body = _list("?page_size=2&page=1")
|
|
assert body["total"] == 3
|
|
assert [i["id"] for i in body["items"]] == ["c4", "c2"]
|
|
body = _list("?page_size=2&page=2")
|
|
assert [i["id"] for i in body["items"]] == ["c1"]
|
|
body = _list("?page_size=2&page=3")
|
|
assert body["items"] == []
|
|
|
|
# 上限与非法参数
|
|
assert super_client.get(
|
|
f"{TENANTS}/demo/conversations?page_size=101",
|
|
headers=_auth(super_client)).status_code == 400
|
|
assert super_client.get(
|
|
f"{TENANTS}/demo/conversations?page=0",
|
|
headers=_auth(super_client)).status_code == 400
|
|
assert super_client.get(
|
|
f"{TENANTS}/demo/conversations?intent=bogus",
|
|
headers=_auth(super_client)).status_code == 400
|
|
assert super_client.get(
|
|
f"{TENANTS}/demo/conversations?from=not-a-date",
|
|
headers=_auth(super_client)).status_code == 400
|
|
|
|
|
|
def test_conversation_detail_with_messages(super_client, tmp_path):
|
|
_create(super_client, slug="demo")
|
|
|
|
async def populate(session):
|
|
at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
|
|
session.add(_conv("c1", at, site="es", ip="8.8.8.8", lang="en"))
|
|
session.add(_msg("c1", MessageRole.user, "您好，我想买切割机", at))
|
|
session.add(_msg("c1", MessageRole.assistant,
|
|
"您好！请问要切什么材料？", at))
|
|
# 空内容消息: jc 详情语义不收
|
|
session.add(_msg("c1", MessageRole.assistant, "", at))
|
|
session.add(CustomerProfile(
|
|
conversation_id="c1",
|
|
raw_info={"姓名": "John", "产品": "切割机",
|
|
"_cc": {"whatsapp": "+8613800000000"}},
|
|
summary="尼日利亚买家,切割机"))
|
|
session.add(Evaluation(
|
|
conversation_id="c1", intent_level=IntentLevel.interested,
|
|
reasoning="明确采购意向", contact_info={"email": "buy@example.com"},
|
|
email_sent=True, correction_count=1, evaluated_at=at))
|
|
|
|
_seed(tmp_path, "demo", populate)
|
|
|
|
r = super_client.get(f"{TENANTS}/demo/conversations/c1",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200, r.text
|
|
d = r.json()
|
|
# 会话元数据 + client_info 原样
|
|
assert d["id"] == "c1"
|
|
assert d["session_id"] == "sess-c1"
|
|
assert d["status"] == "completed"
|
|
assert d["source"] == "widget"
|
|
assert d["client_info"] == {"ip": "8.8.8.8", "user_agent": "UA-c1",
|
|
"site": "es", "lang": "en"}
|
|
assert d["site"] == "es"
|
|
assert d["country"] == "美国"
|
|
assert d["created_at"].startswith("2026-01-05")
|
|
assert d["updated_at"].startswith("2026-01-05")
|
|
# 画像 chips: raw_info 去 _ 前缀键(_cc 内部态不上桌)
|
|
assert d["profile"]["chips"] == {"姓名": "John", "产品": "切割机"}
|
|
assert d["profile"]["summary"] == "尼日利亚买家,切割机"
|
|
# 意向块
|
|
assert d["evaluation"]["intent_level"] == "interested"
|
|
assert d["evaluation"]["email_sent"] is True
|
|
assert d["evaluation"]["correction_count"] == 1
|
|
# 消息全量(按时间升序,空内容被滤)
|
|
assert [(m["role"], m["content"]) for m in d["messages"]] == [
|
|
("user", "您好，我想买切割机"),
|
|
("assistant", "您好！请问要切什么材料？")]
|
|
assert all(m["time"] for m in d["messages"])
|
|
|
|
# 未知会话 / 未知租户
|
|
assert super_client.get(f"{TENANTS}/demo/conversations/nope",
|
|
headers=_auth(super_client)).status_code == 404
|
|
assert super_client.get(f"{TENANTS}/ghost/conversations/c1",
|
|
headers=_auth(super_client)).status_code == 404
|
|
|
|
|
|
def test_push_logs_payload_lazy(super_client, tmp_path):
|
|
_create(super_client, slug="demo")
|
|
|
|
payload = {"customer": {"name": "John"},
|
|
"messages": [f"msg-{i}" for i in range(40)],
|
|
"marker": PAYLOAD_MARKER}
|
|
|
|
async def populate(session):
|
|
at1 = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
|
|
at2 = datetime(2026, 1, 6, 9, 0, tzinfo=UTC)
|
|
session.add(_conv("c1", at1))
|
|
session.add(_msg("c1", MessageRole.user, "您好", at1))
|
|
session.add(CustomerProfile(conversation_id="c1", raw_info={}))
|
|
ev_ok = Evaluation(
|
|
conversation_id="c1", intent_level=IntentLevel.interested,
|
|
contact_info={"email": "buy@example.com"}, evaluated_at=at1)
|
|
session.add(ev_ok)
|
|
await session.flush()
|
|
session.add(LeadPushLog(
|
|
evaluation_id=ev_ok.id, conversation_id="c1", push_type="lead",
|
|
status="success", attempts=1, request_payload=payload,
|
|
response_status=200, outcome="accepted", created_at=at1,
|
|
updated_at=at1))
|
|
session.add(LeadPushLog(
|
|
evaluation_id=ev_ok.id, conversation_id="c1", push_type="correction",
|
|
status="failed", attempts=2, request_payload=payload,
|
|
response_status=500, error_message="upstream 500", created_at=at2,
|
|
updated_at=at2))
|
|
|
|
_seed(tmp_path, "demo", populate)
|
|
|
|
r = super_client.get(f"{TENANTS}/demo/push-logs", headers=_auth(super_client))
|
|
assert r.status_code == 200, r.text
|
|
body = r.json()
|
|
assert body["total"] == 2
|
|
# 大字段 lazy: 响应绝不含 payload 内容与键名,但有长度标记
|
|
assert PAYLOAD_MARKER not in r.text
|
|
assert "request_payload" not in r.text
|
|
by_status = {i["status"]: i for i in body["items"]}
|
|
assert set(by_status) == {"success", "failed"}
|
|
ok = by_status["success"]
|
|
assert ok["conversation_id"] == "c1"
|
|
assert ok["push_type"] == "lead"
|
|
assert ok["outcome"] == "accepted"
|
|
assert ok["response_status"] == 200
|
|
assert ok["attempts"] == 1
|
|
assert isinstance(ok["payload_size"], int) and ok["payload_size"] > 0
|
|
assert by_status["failed"]["error_message"] == "upstream 500"
|
|
|
|
# status 筛选 / conversation_id 筛选 / 非法值
|
|
r = super_client.get(f"{TENANTS}/demo/push-logs?status=failed",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert [i["status"] for i in r.json()["items"]] == ["failed"]
|
|
r = super_client.get(f"{TENANTS}/demo/push-logs?conversation_id=c2",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert r.json()["total"] == 0
|
|
assert super_client.get(f"{TENANTS}/demo/push-logs?status=bogus",
|
|
headers=_auth(super_client)).status_code == 400
|
|
# 分页
|
|
r = super_client.get(f"{TENANTS}/demo/push-logs?page_size=1&page=2",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert len(r.json()["items"]) == 1
|
|
|
|
|
|
def test_tenant_scope_isolation(super_client, tmp_path):
|
|
_create(super_client, slug="jca", admin="jca-admin",
|
|
password="jca-pass-1")
|
|
# blank: 第二个租户不撞 jiancai 出厂 push.prefix(终审 I-3)
|
|
_create(super_client, slug="jcb", admin="jcb-admin",
|
|
password="jcb-pass-1", profile="blank")
|
|
|
|
async def populate(session):
|
|
at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
|
|
session.add(_conv("b1", at))
|
|
session.add(_msg("b1", MessageRole.user, "hello", at))
|
|
|
|
_seed(tmp_path, "jcb", populate)
|
|
|
|
token = _tenant_admin_token(super_client, "jca-admin", "jca-pass-1")
|
|
# jca 管理员看 jcb 的数据 → 403(列表/详情/删除/推送日志全挡)
|
|
assert super_client.get(
|
|
f"{TENANTS}/jcb/conversations",
|
|
headers=_bearer(token)).status_code == 403
|
|
assert super_client.get(
|
|
f"{TENANTS}/jcb/conversations/b1",
|
|
headers=_bearer(token)).status_code == 403
|
|
assert super_client.delete(
|
|
f"{TENANTS}/jcb/conversations/b1",
|
|
headers=_bearer(token)).status_code == 403
|
|
assert super_client.get(
|
|
f"{TENANTS}/jcb/push-logs",
|
|
headers=_bearer(token)).status_code == 403
|
|
# 自己的 → 200(空库)
|
|
r = super_client.get(f"{TENANTS}/jca/conversations",
|
|
headers=_bearer(token))
|
|
assert r.status_code == 200
|
|
assert r.json()["total"] == 0
|
|
# 超管看 jcb → 200 且数据在
|
|
r = super_client.get(f"{TENANTS}/jcb/conversations",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert [i["id"] for i in r.json()["items"]] == ["b1"]
|
|
# 未认证 → 401
|
|
assert super_client.get(
|
|
f"{TENANTS}/jcb/conversations").status_code == 401
|
|
|
|
|
|
# ---- 删除 / 懒建 ------------------------------------------------------------------
|
|
|
|
|
|
def test_delete_conversation_cascades(super_client, tmp_path):
|
|
_create(super_client, slug="demo")
|
|
|
|
async def populate(session):
|
|
at = datetime(2026, 1, 5, 12, 0, tzinfo=UTC)
|
|
session.add(_conv("c1", at))
|
|
session.add(_msg("c1", MessageRole.user, "您好", at))
|
|
session.add(_msg("c1", MessageRole.assistant, "你好", at))
|
|
session.add(CustomerProfile(conversation_id="c1", raw_info={"姓名": "J"}))
|
|
ev = Evaluation(conversation_id="c1",
|
|
intent_level=IntentLevel.interested,
|
|
contact_info={"email": "x@y.com"}, evaluated_at=at)
|
|
session.add(ev)
|
|
await session.flush()
|
|
session.add(LeadPushLog(
|
|
evaluation_id=ev.id, conversation_id="c1", push_type="lead",
|
|
status="success", request_payload={"m": 1}, created_at=at,
|
|
updated_at=at))
|
|
|
|
_seed(tmp_path, "demo", populate)
|
|
|
|
r = super_client.delete(f"{TENANTS}/demo/conversations/c1",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"ok": True}
|
|
# jc 删除语义(级联消息/画像/意向) + 推送日志补齐: 五表全空
|
|
for table in (Conversation, Message, CustomerProfile, Evaluation,
|
|
LeadPushLog):
|
|
assert _count(tmp_path, "demo", table) == 0, table.__tablename__
|
|
# 详情随之 404;重复删 404
|
|
assert super_client.get(f"{TENANTS}/demo/conversations/c1",
|
|
headers=_auth(super_client)).status_code == 404
|
|
assert super_client.delete(f"{TENANTS}/demo/conversations/c1",
|
|
headers=_auth(super_client)).status_code == 404
|
|
|
|
|
|
def test_lazy_tenant_first_query_builds_db(super_client, db_source_app, tmp_path):
|
|
"""未懒建租户(创建后从未被域名 resolve,engine 仍 None)首次查询数据
|
|
视图: ensure_ready 建库建表,空列表 200,不炸。"""
|
|
_create(super_client, slug="fresh")
|
|
registry = db_source_app.state.registry
|
|
rt = registry._by_slug["fresh"]
|
|
assert rt.engine is None # 前提: 确未懒建
|
|
|
|
r = super_client.get(f"{TENANTS}/fresh/conversations",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200, r.text
|
|
assert r.json() == {"total": 0, "page": 1, "page_size": 20, "items": []}
|
|
assert rt.engine is not None # 首次查询触发懒建
|
|
assert (tmp_path / "tenants" / "fresh.db").exists()
|
|
# 推送日志同样触发(复用已建 engine,再查不炸)
|
|
r = super_client.get(f"{TENANTS}/fresh/push-logs",
|
|
headers=_auth(super_client))
|
|
assert r.status_code == 200
|
|
assert r.json()["total"] == 0
|