xsl
d9249400af
12 类实体统一表 + 状态机 + version 递增 + kb_relation 递归 CTE 1~3 跳
可解释路径（防环）+ kb_color 色板（T-P2-01/05，TD-09/18）。
实现：
- packages/kb/schemas.py：12 类 EntityType 与状态 Literal 校验、
EntityCreate/Update/Out、RelationCreate、GraphNode/Hop/Path/QueryResult、
ColorOut；雪花 ID 全部字符串出参（JS 精度）。
- packages/kb/repo.py：实体 CRUD/按名查重/过滤分页搜索；traverse 递归
CTE（path 数组 + NOT @> 防环，锚点与递归段 varchar[] 显转型）；色板读写。
- packages/kb/service.py：状态机矩阵（draft→reviewing→published→archived
+ draft/reviewing→archived 撤稿，非法跃迁 40900，矩阵单测 16 组全覆盖）、
内容修改仅 draft/reviewing 且 version+1、published 只读；关系写入
（自环/三元组重复 400、端点缺失 404）；色板 role/hex(#RRGGBB) 校验。
- apps/api/routers/kb.py：GET/POST /kb/items、GET/PATCH /kb/items/{id}、
GET /kb/items/{id}/graph?max_hops=1..3（超界 422）；KbService 装配就近
放路由文件（apps/api/deps.py 本轮禁改）。
测试：单测 77 例（跃迁矩阵、12 类 Literal、非法类型拒绝、hex/role/hops
校验）；集成 5 例（真实 PG：生命周期/改 published 被拒/版本递增、关系
约束、2 跳可解释路径+防环、列表过滤分页、色板读写）。全套门禁绿：
ruff check/format、mypy（指定包列表）、lint-imports、单测 315 通过、
集成 21 通过。
偏差说明：
1. 新增 tests/unit/__init__.py 与 tests/integration/__init__.py（仅注释
的包标记）：任务文档要求单测/集成同名 test_kb_entity.py，pytest
prepend 导入模式按 basename 注册模块会撞名，加包标记唯一化。
2. 关系与色板未挂 HTTP 路由：文档 §3.4 路由清单仅列 items+graph，
录入入口属运营后台 T-P2-09，本批在 service 层落地并集成直连验证。
3. published→draft 按文档字面矩阵（其余跃迁非法）拦为 40900：改已发布
内容的既定路径是撤稿归档后重录新草稿（错误 hint 已说明），未私开
回退通道。
407 lines
14 KiB
Python
407 lines
14 KiB
Python
"""kb 实体底座集成用例（真实 PostgreSQL，T-P2-01/05）.
|
||
|
||
覆盖：实体生命周期（draft→reviewing→published、改 published 被拒、
|
||
draft 改 version+1）、关系写入约束（自环/重复/端点缺失）、递归 CTE
|
||
1~3 跳可解释路径与防环、列表过滤分页、自建色板读写。
|
||
实体/关系录入没有 HTTP 路由（运营后台属 T-P2-09），这些路径直接驱动
|
||
service；读路径走 HTTP，顺带验证路由装配与统一信封。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import uuid
|
||
from collections.abc import AsyncIterator
|
||
from contextlib import asynccontextmanager
|
||
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(autouse=True)
|
||
async def _dispose_engine_after_test() -> AsyncIterator[None]:
|
||
"""用例结束即释放全局引擎。
|
||
|
||
引擎是进程级懒加载单例，而 pytest-asyncio 每个用例一个新 event loop：
|
||
走 HTTP 的用例靠 client 的 lifespan teardown 顺手 dispose；本文件的
|
||
service 直连用例没有这层兜底，不主动释放会把绑定旧 loop 的连接池
|
||
泄漏给后续用例（表现为 Event loop is closed / sqlalchemy 报错）。
|
||
"""
|
||
yield
|
||
from packages.shared.db import dispose_engine
|
||
|
||
await dispose_engine()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client() -> Any:
|
||
import httpx
|
||
|
||
from apps.api.main import create_app
|
||
from packages.shared.config import get_settings
|
||
|
||
# kb 链路不碰 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]:
|
||
resp = await client.post(
|
||
"/api/v1/auth/dev-token", json={"device_key": f"t6-{uuid.uuid4().hex[:8]}"}
|
||
)
|
||
assert resp.json()["code"] == 0
|
||
data = resp.json()["data"]
|
||
return {
|
||
"headers": {"Authorization": f"Bearer {data['access_token']}"},
|
||
"user_id": data["user_id"],
|
||
}
|
||
|
||
|
||
@asynccontextmanager
|
||
async def service_scope() -> AsyncIterator[Any]:
|
||
"""直连 service 的事务作用域：成功 commit、异常 rollback。"""
|
||
from packages.kb.service import KbService
|
||
from packages.shared.db import get_session_factory
|
||
|
||
async with get_session_factory()() as session:
|
||
try:
|
||
yield KbService(session)
|
||
await session.commit()
|
||
except Exception:
|
||
await session.rollback()
|
||
raise
|
||
|
||
|
||
async def _mk_entity(
|
||
kb: Any, *, entity_type: str, name: str, attrs: dict[str, Any] | None = None
|
||
) -> Any:
|
||
from packages.kb.schemas import EntityCreate
|
||
|
||
return await kb.create_entity(
|
||
payload=EntityCreate(entity_type=entity_type, name=name, attrs=attrs or {}), # type: ignore[arg-type]
|
||
updated_by="it",
|
||
)
|
||
|
||
|
||
async def test_entity_lifecycle_publish_freezes_and_version_bumps(
|
||
client: Any, auth: dict[str, Any]
|
||
) -> None:
|
||
headers = auth["headers"]
|
||
suffix = uuid.uuid4().hex[:8]
|
||
|
||
# 新建 → draft / version=1
|
||
created = await client.post(
|
||
"/api/v1/kb/items",
|
||
json={"entity_type": "item", "name": f"白衬衫-{suffix}", "attrs": {"sleeve": "long"}},
|
||
headers=headers,
|
||
)
|
||
assert created.status_code == 200
|
||
body = created.json()
|
||
assert body["code"] == 0
|
||
entity = body["data"]
|
||
entity_id = entity["entity_id"]
|
||
assert entity["status"] == "draft"
|
||
assert entity["version"] == 1
|
||
assert entity["attrs"] == {"sleeve": "long"}
|
||
|
||
# 详情可读
|
||
detail = (await client.get(f"/api/v1/kb/items/{entity_id}", headers=headers)).json()
|
||
assert detail["code"] == 0
|
||
assert detail["data"]["name"] == f"白衬衫-{suffix}"
|
||
|
||
# 非法跃迁：draft 不能直接 published
|
||
shortcut = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "published"}, headers=headers
|
||
)
|
||
assert shortcut.status_code == 409
|
||
assert shortcut.json()["code"] == 40900
|
||
|
||
# draft → reviewing（纯跃迁，version 不变）→ published
|
||
reviewing = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "reviewing"}, headers=headers
|
||
)
|
||
assert reviewing.json()["data"]["status"] == "reviewing"
|
||
assert reviewing.json()["data"]["version"] == 1
|
||
published = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "published"}, headers=headers
|
||
)
|
||
assert published.json()["data"]["status"] == "published"
|
||
|
||
# 改 published 被拒（内容与回 draft 都不行）
|
||
frozen = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"attrs": {"sleeve": "short"}}, headers=headers
|
||
)
|
||
assert frozen.status_code == 409
|
||
assert frozen.json()["code"] == 40900
|
||
backdoor = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "draft"}, headers=headers
|
||
)
|
||
assert backdoor.status_code == 409
|
||
assert backdoor.json()["code"] == 40900
|
||
|
||
# published → archived 收尾合法；archived 终态
|
||
archived = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "archived"}, headers=headers
|
||
)
|
||
assert archived.json()["data"]["status"] == "archived"
|
||
resurrect = await client.patch(
|
||
f"/api/v1/kb/items/{entity_id}", json={"status": "reviewing"}, headers=headers
|
||
)
|
||
assert resurrect.status_code == 409
|
||
|
||
# draft 实体改内容 → version+1；同名同类目拒绝
|
||
draft2 = await client.post(
|
||
"/api/v1/kb/items",
|
||
json={"entity_type": "item", "name": f"牛仔裤-{suffix}"},
|
||
headers=headers,
|
||
)
|
||
draft2_id = draft2.json()["data"]["entity_id"]
|
||
edited = await client.patch(
|
||
f"/api/v1/kb/items/{draft2_id}",
|
||
json={"attrs": {"fit": "slim"}, "source": "搭配手册"},
|
||
headers=headers,
|
||
)
|
||
assert edited.json()["data"]["version"] == 2
|
||
assert edited.json()["data"]["attrs"] == {"fit": "slim"}
|
||
duplicate = await client.post(
|
||
"/api/v1/kb/items",
|
||
json={"entity_type": "item", "name": f"牛仔裤-{suffix}"},
|
||
headers=headers,
|
||
)
|
||
assert duplicate.status_code == 400
|
||
assert duplicate.json()["code"] == 40000
|
||
|
||
# 非法 entity_type / 超界 max_hops 由入口校验拦成 422
|
||
bad_type = await client.post(
|
||
"/api/v1/kb/items", json={"entity_type": "user", "name": "x"}, headers=headers
|
||
)
|
||
assert bad_type.status_code == 422
|
||
assert bad_type.json()["code"] == 40001
|
||
bad_hops = await client.get(f"/api/v1/kb/items/{draft2_id}/graph?max_hops=4", headers=headers)
|
||
assert bad_hops.status_code == 422
|
||
|
||
|
||
async def test_relation_rules_self_loop_duplicate_and_missing() -> None:
|
||
from packages.kb.schemas import RelationCreate
|
||
from packages.shared.errors import AppError, ErrorCode
|
||
|
||
suffix = uuid.uuid4().hex[:8]
|
||
async with service_scope() as kb:
|
||
a = await _mk_entity(kb, entity_type="item", name=f"selfloop-{suffix}")
|
||
b = await _mk_entity(kb, entity_type="style", name=f"duppair-{suffix}")
|
||
|
||
# 自环拒绝
|
||
with pytest.raises(AppError) as loop_err:
|
||
await kb.add_relation(
|
||
payload=RelationCreate(
|
||
from_entity_id=a.entity_id, to_entity_id=a.entity_id, rel_type="suits"
|
||
)
|
||
)
|
||
assert loop_err.value.code == ErrorCode.BAD_REQUEST
|
||
|
||
# 端点缺失 → 40400
|
||
with pytest.raises(AppError) as missing_err:
|
||
await kb.add_relation(
|
||
payload=RelationCreate(from_entity_id=a.entity_id, to_entity_id=1, rel_type="suits")
|
||
)
|
||
assert missing_err.value.code == ErrorCode.NOT_FOUND
|
||
|
||
# 三元组重复拒绝
|
||
await kb.add_relation(
|
||
payload=RelationCreate(
|
||
from_entity_id=a.entity_id, to_entity_id=b.entity_id, rel_type="suits"
|
||
)
|
||
)
|
||
with pytest.raises(AppError) as dup_err:
|
||
await kb.add_relation(
|
||
payload=RelationCreate(
|
||
from_entity_id=a.entity_id, to_entity_id=b.entity_id, rel_type="suits"
|
||
)
|
||
)
|
||
assert dup_err.value.code == ErrorCode.BAD_REQUEST
|
||
|
||
|
||
async def test_graph_two_hops_explainable_paths_and_cycle_guard(
|
||
client: Any, auth: dict[str, Any]
|
||
) -> None:
|
||
headers = auth["headers"]
|
||
suffix = uuid.uuid4().hex[:8]
|
||
async with service_scope() as kb:
|
||
from packages.kb.schemas import RelationCreate
|
||
|
||
shirt = await _mk_entity(kb, entity_type="item", name=f"衬衫-{suffix}")
|
||
commut = await _mk_entity(kb, entity_type="style", name=f"通勤风-{suffix}")
|
||
interview = await _mk_entity(kb, entity_type="occasion", name=f"面试-{suffix}")
|
||
for from_e, to_e, rel in (
|
||
(shirt, commut, "suits"),
|
||
(commut, interview, "fits"),
|
||
(interview, shirt, "matches"), # 成环：面试 → 衬衫 → 通勤风 → …
|
||
):
|
||
await kb.add_relation(
|
||
payload=RelationCreate(
|
||
from_entity_id=from_e.entity_id, to_entity_id=to_e.entity_id, rel_type=rel
|
||
)
|
||
)
|
||
|
||
graph = (
|
||
await client.get(f"/api/v1/kb/items/{shirt.entity_id}/graph?max_hops=2", headers=headers)
|
||
).json()
|
||
assert graph["code"] == 0
|
||
data = graph["data"]
|
||
assert data["from_entity_id"] == str(shirt.entity_id)
|
||
assert data["max_hops"] == 2
|
||
|
||
def _signature(path: dict[str, Any]) -> tuple[tuple[str, ...], tuple[str, ...]]:
|
||
return (
|
||
tuple(node["entity_id"] for node in path["nodes"]),
|
||
tuple(hop["rel_type"] for hop in path["hops"]),
|
||
)
|
||
|
||
signatures = {_signature(p) for p in data["paths"]}
|
||
assert (
|
||
(str(shirt.entity_id), str(commut.entity_id)),
|
||
("suits",),
|
||
) in signatures, "1 跳路径应包含每跳 rel_type"
|
||
assert (
|
||
(str(shirt.entity_id), str(commut.entity_id), str(interview.entity_id)),
|
||
("suits", "fits"),
|
||
) in signatures, "2 跳路径应可解释：节点序列 + 逐跳 rel_type"
|
||
# 节点摘要可读（不是光秃秃的 id）
|
||
first = next(p for p in data["paths"] if len(p["hops"]) == 1)
|
||
assert first["nodes"][0]["name"] == f"衬衫-{suffix}"
|
||
assert first["nodes"][0]["entity_type"] == "item"
|
||
|
||
# 防环：3 跳也不会绕回起点或无限展开
|
||
deep = (
|
||
await client.get(f"/api/v1/kb/items/{shirt.entity_id}/graph?max_hops=3", headers=headers)
|
||
).json()["data"]
|
||
for path in deep["paths"]:
|
||
ids = [node["entity_id"] for node in path["nodes"]]
|
||
assert len(ids) == len(set(ids)), "路径内不得重复出现同一实体"
|
||
assert ids[0] == str(shirt.entity_id)
|
||
# 环上只有 2 条不回头路径：衬衫→通勤风、衬衫→通勤风→面试（→衬衫 被防环剪掉）
|
||
assert len(deep["paths"]) == 2
|
||
|
||
# 起点 404 语义
|
||
missing = await client.get("/api/v1/kb/items/1/graph", headers=headers)
|
||
assert missing.status_code == 404
|
||
assert missing.json()["code"] == 40400
|
||
|
||
|
||
async def test_list_items_filter_and_pagination(client: Any, auth: dict[str, Any]) -> None:
|
||
headers = auth["headers"]
|
||
suffix = uuid.uuid4().hex[:8]
|
||
for i in range(3):
|
||
resp = await client.post(
|
||
"/api/v1/kb/items",
|
||
json={"entity_type": "style", "name": f"pag-{suffix}-{i}"},
|
||
headers=headers,
|
||
)
|
||
assert resp.json()["code"] == 0
|
||
await client.post(
|
||
"/api/v1/kb/items",
|
||
json={"entity_type": "material", "name": f"pag-{suffix}-denim"},
|
||
headers=headers,
|
||
)
|
||
|
||
# entity_type 过滤 + 名称模糊（把别的历史数据隔离掉）
|
||
base = "/api/v1/kb/items"
|
||
first = (
|
||
await client.get(
|
||
f"{base}?entity_type=style&name=pag-{suffix}&limit=2&offset=0", headers=headers
|
||
)
|
||
).json()["data"]
|
||
assert first["total"] == 3
|
||
assert len(first["items"]) == 2
|
||
second = (
|
||
await client.get(
|
||
f"{base}?entity_type=style&name=pag-{suffix}&limit=2&offset=2", headers=headers
|
||
)
|
||
).json()["data"]
|
||
assert len(second["items"]) == 1
|
||
ids_first = {item["entity_id"] for item in first["items"]}
|
||
ids_second = {item["entity_id"] for item in second["items"]}
|
||
assert ids_first | ids_second and not ids_first & ids_second
|
||
|
||
# status 过滤：归档一条后，draft 视图少一行
|
||
archived_id = first["items"][0]["entity_id"]
|
||
await client.patch(
|
||
f"/api/v1/kb/items/{archived_id}", json={"status": "archived"}, headers=headers
|
||
)
|
||
drafts = (
|
||
await client.get(
|
||
f"{base}?entity_type=style&name=pag-{suffix}&status=draft", headers=headers
|
||
)
|
||
).json()["data"]
|
||
assert drafts["total"] == 2
|
||
|
||
# 非法 status 枚举 → 422
|
||
bad_status = await client.get(f"{base}?status=frozen", headers=headers)
|
||
assert bad_status.status_code == 422
|
||
|
||
|
||
async def test_color_palette_roundtrip() -> None:
|
||
from packages.shared.errors import AppError, ErrorCode
|
||
|
||
# season_code 是 varchar(16)：uuid 后缀只能取 8 位防超长
|
||
season = f"sw{uuid.uuid4().hex[:8]}"
|
||
async with service_scope() as kb:
|
||
color = await kb.add_color(
|
||
season_code=season,
|
||
role="base",
|
||
lab_l=82.35,
|
||
lab_a=1.2,
|
||
lab_b=9.87,
|
||
hex_code="#F2D8C2",
|
||
name_cn="杏色",
|
||
)
|
||
assert color.role == "base"
|
||
assert color.hex == "#F2D8C2"
|
||
assert color.lab_l == pytest.approx(82.35)
|
||
|
||
with pytest.raises(AppError) as hex_err:
|
||
await kb.add_color(
|
||
season_code=season,
|
||
role="accent",
|
||
lab_l=50,
|
||
lab_a=10,
|
||
lab_b=-10,
|
||
hex_code="F2D8C2", # 缺 #
|
||
name_cn="坏色",
|
||
)
|
||
assert hex_err.value.code == ErrorCode.BAD_REQUEST
|
||
|
||
with pytest.raises(AppError) as role_err:
|
||
await kb.add_color(
|
||
season_code=season,
|
||
role="primary", # 不在 base/accent/avoid
|
||
lab_l=50,
|
||
lab_a=10,
|
||
lab_b=-10,
|
||
hex_code="#112233",
|
||
name_cn="坏角色",
|
||
)
|
||
assert role_err.value.code == ErrorCode.BAD_REQUEST
|
||
|
||
rows = await kb.list_colors(season_code=season)
|
||
assert [row.hex for row in rows] == ["#F2D8C2"]
|