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 已说明），未私开
回退通道。
223 lines
8.7 KiB
Python
223 lines
8.7 KiB
Python
"""kb 域纯逻辑单测（不连库）：状态机全跃迁矩阵、12 类 Literal、色板格式校验。
|
||
|
||
repo / service 的 DB 路径由 ``tests/integration/test_kb_entity.py`` 覆盖，
|
||
这里只钉死「规则本身」：哪条跃迁合法、哪类实体名合法、什么 hex 合法。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from itertools import product
|
||
from typing import get_args
|
||
|
||
import pytest
|
||
from pydantic import ValidationError
|
||
|
||
from packages.kb.schemas import (
|
||
ENTITY_STATUSES,
|
||
ENTITY_TYPES,
|
||
MAX_GRAPH_HOPS,
|
||
ColorRole,
|
||
EntityCreate,
|
||
EntityUpdate,
|
||
GraphHop,
|
||
GraphNode,
|
||
GraphPath,
|
||
RelationCreate,
|
||
)
|
||
from packages.kb.service import (
|
||
CONTENT_EDITABLE_STATUSES,
|
||
LEGAL_TRANSITIONS,
|
||
assert_transition,
|
||
can_transition,
|
||
ensure_content_editable,
|
||
ensure_valid_color_role,
|
||
ensure_valid_hex,
|
||
ensure_valid_hops,
|
||
)
|
||
from packages.shared.errors import AppError, ErrorCode
|
||
|
||
# 12 类本体（知识库文档 §3.1）。这里刻意不引用 ENTITY_TYPES 常量本身，
|
||
# 而是手写清单——常量抄错时测试必须能抓住。
|
||
_EXPECTED_12_TYPES = (
|
||
"item",
|
||
"color",
|
||
"style",
|
||
"body_shape",
|
||
"face_hair",
|
||
"occasion",
|
||
"material",
|
||
"accessory",
|
||
"makeup",
|
||
"rule",
|
||
"outfit",
|
||
"trend",
|
||
)
|
||
|
||
#: 状态机合法跃迁（draft→reviewing→published→published 后归档 + 撤稿支线）。
|
||
#: published→draft 不在其中：改已发布内容 = 撤稿归档后重录新草稿。
|
||
_LEGAL_PAIRS = {
|
||
("draft", "reviewing"),
|
||
("reviewing", "published"),
|
||
("published", "archived"),
|
||
("draft", "archived"),
|
||
("reviewing", "archived"),
|
||
}
|
||
|
||
|
||
class TestStateMachineMatrix:
|
||
"""全跃迁矩阵：4 态 × 4 态共 16 组合，合法 5 条，其余 11 条全部拦下。"""
|
||
|
||
@pytest.mark.parametrize(("current", "target"), list(product(ENTITY_STATUSES, repeat=2)))
|
||
def test_matrix(self, current: str, target: str) -> None:
|
||
expected = (current, target) in _LEGAL_PAIRS
|
||
assert can_transition(current, target) is expected
|
||
if expected:
|
||
assert_transition(current, target) # 不抛即合法
|
||
else:
|
||
with pytest.raises(AppError) as exc_info:
|
||
assert_transition(current, target)
|
||
assert exc_info.value.code == ErrorCode.STATE_CONFLICT
|
||
assert exc_info.value.http_status == 409
|
||
assert exc_info.value.data == {"current": current, "target": target}
|
||
|
||
def test_matrix_definition_is_exact(self) -> None:
|
||
"""合法性来源是 LEGAL_TRANSITIONS 常量：常量与期望集合互为镜像。"""
|
||
assert frozenset(_LEGAL_PAIRS) == LEGAL_TRANSITIONS
|
||
assert len(_LEGAL_PAIRS) == 5
|
||
|
||
@pytest.mark.parametrize("status", ["draft", "reviewing"])
|
||
def test_content_editable_in(self, status: str) -> None:
|
||
ensure_content_editable(status) # 不抛即可编辑
|
||
|
||
@pytest.mark.parametrize("status", ["published", "archived"])
|
||
def test_content_frozen_outside_draft_reviewing(self, status: str) -> None:
|
||
with pytest.raises(AppError) as exc_info:
|
||
ensure_content_editable(status)
|
||
assert exc_info.value.code == ErrorCode.STATE_CONFLICT
|
||
assert frozenset({"draft", "reviewing"}) == CONTENT_EDITABLE_STATUSES
|
||
|
||
|
||
class TestEntitySchemas:
|
||
@pytest.mark.parametrize("entity_type", _EXPECTED_12_TYPES)
|
||
def test_all_12_entity_types_accepted(self, entity_type: str) -> None:
|
||
payload = EntityCreate(entity_type=entity_type, name="白衬衫", attrs={"sleeve": "long"})
|
||
assert payload.entity_type == entity_type
|
||
assert ENTITY_TYPES == _EXPECTED_12_TYPES
|
||
assert len(_EXPECTED_12_TYPES) == 12
|
||
|
||
@pytest.mark.parametrize("entity_type", ["garbage", "Item", "items", "", "user"])
|
||
def test_invalid_entity_type_rejected(self, entity_type: str) -> None:
|
||
with pytest.raises(ValidationError):
|
||
EntityCreate(entity_type=entity_type, name="白衬衫") # type: ignore[arg-type]
|
||
|
||
@pytest.mark.parametrize("name", ["", "x" * 129])
|
||
def test_name_length_enforced(self, name: str) -> None:
|
||
with pytest.raises(ValidationError):
|
||
EntityCreate(entity_type="item", name=name)
|
||
|
||
def test_attrs_default_empty_and_optional_fields(self) -> None:
|
||
payload = EntityCreate(entity_type="style", name="通勤风")
|
||
assert payload.attrs == {}
|
||
assert payload.source is None
|
||
assert payload.license is None
|
||
|
||
def test_entity_update_defaults_and_status_literal(self) -> None:
|
||
assert EntityUpdate().content_fields() == {}
|
||
assert EntityUpdate(attrs={"a": 1}).content_fields() == {"attrs": {"a": 1}}
|
||
# status 不是内容字段：纯跃迁请求不该引起 version 变化
|
||
assert EntityUpdate(status="published").content_fields() == {}
|
||
with pytest.raises(ValidationError):
|
||
EntityUpdate(status="frozen") # type: ignore[arg-type]
|
||
|
||
def test_entity_update_distinguishes_unset_and_null(self) -> None:
|
||
"""PATCH 语义：显式传 null 清空字段，未传不动字段。"""
|
||
payload = EntityUpdate.model_validate({"source": None})
|
||
assert payload.content_fields() == {"source": None}
|
||
assert EntityUpdate.model_validate({}).content_fields() == {}
|
||
|
||
def test_status_literal_matches_state_machine(self) -> None:
|
||
assert set(ENTITY_STATUSES) == {"draft", "reviewing", "published", "archived"}
|
||
|
||
|
||
class TestRelationAndGraphSchemas:
|
||
def test_relation_create_defaults(self) -> None:
|
||
payload = RelationCreate(from_entity_id=1, to_entity_id=2, rel_type="suits")
|
||
assert payload.weight == payload.confidence # 默认 1
|
||
assert str(payload.weight) == "1"
|
||
|
||
@pytest.mark.parametrize(
|
||
("rel_type", "weight"),
|
||
[("", 1), ("x" * 65, 1), ("suits", -0.1), ("suits", 10)],
|
||
)
|
||
def test_relation_create_rejects_bad_shape(self, rel_type: str, weight: float) -> None:
|
||
with pytest.raises(ValidationError):
|
||
RelationCreate(from_entity_id=1, to_entity_id=2, rel_type=rel_type, weight=weight)
|
||
|
||
def test_graph_path_shape_invariants(self) -> None:
|
||
"""可解释路径的结构不变式：len(hops) == len(nodes) - 1，逐跳有 rel_type。"""
|
||
path = GraphPath(
|
||
nodes=[
|
||
GraphNode(entity_id="1", entity_type="item", name="白衬衫", status="published"),
|
||
GraphNode(entity_id="2", entity_type="style", name="通勤风", status="published"),
|
||
GraphNode(entity_id="3", entity_type="occasion", name="面试", status="published"),
|
||
],
|
||
hops=[
|
||
GraphHop(from_entity_id="1", rel_type="suits", to_entity_id="2"),
|
||
GraphHop(from_entity_id="2", rel_type="fits", to_entity_id="3"),
|
||
],
|
||
)
|
||
assert len(path.hops) == len(path.nodes) - 1
|
||
assert [hop.rel_type for hop in path.hops] == ["suits", "fits"]
|
||
|
||
|
||
class TestColorValidation:
|
||
@pytest.mark.parametrize("hex_code", ["#A1B2C3", "#a1b2c3", "#000000", "#FFFFFF"])
|
||
def test_valid_hex(self, hex_code: str) -> None:
|
||
ensure_valid_hex(hex_code) # 不抛即合法
|
||
|
||
@pytest.mark.parametrize(
|
||
"hex_code",
|
||
[
|
||
"A1B2C3", # 缺 #
|
||
"#A1B2C", # 5 位
|
||
"#A1B2C3D", # 7 位
|
||
"#G1H2J3", # 非 hex 字符
|
||
" #A1B2C3", # 前导空格
|
||
"#A1B2C3\n", # 尾部换行：fullmatch 必须拦
|
||
"#Z9",
|
||
"",
|
||
],
|
||
)
|
||
def test_invalid_hex(self, hex_code: str) -> None:
|
||
with pytest.raises(AppError) as exc_info:
|
||
ensure_valid_hex(hex_code)
|
||
assert exc_info.value.code == ErrorCode.BAD_REQUEST
|
||
assert exc_info.value.http_status == 400
|
||
|
||
@pytest.mark.parametrize("role", ["base", "accent", "avoid"])
|
||
def test_valid_roles(self, role: str) -> None:
|
||
ensure_valid_color_role(role)
|
||
|
||
@pytest.mark.parametrize("role", ["primary", "BASE", "", "main"])
|
||
def test_invalid_roles(self, role: str) -> None:
|
||
with pytest.raises(AppError) as exc_info:
|
||
ensure_valid_color_role(role)
|
||
assert exc_info.value.code == ErrorCode.BAD_REQUEST
|
||
|
||
def test_color_role_literal_matches_guard(self) -> None:
|
||
assert set(get_args(ColorRole)) == {"base", "accent", "avoid"}
|
||
|
||
|
||
class TestHopsBound:
|
||
"""1~3 跳：超界 400（HTTP 入口由 Query ge/le 先行拦成 422，service 兜底）。"""
|
||
|
||
@pytest.mark.parametrize("max_hops", [1, 2, MAX_GRAPH_HOPS])
|
||
def test_valid_hops(self, max_hops: int) -> None:
|
||
ensure_valid_hops(max_hops)
|
||
|
||
@pytest.mark.parametrize("max_hops", [0, -1, 4, 99])
|
||
def test_out_of_range_hops(self, max_hops: int) -> None:
|
||
with pytest.raises(AppError) as exc_info:
|
||
ensure_valid_hops(max_hops)
|
||
assert exc_info.value.code == ErrorCode.BAD_REQUEST
|