diff --git a/apps/api/routers/kb.py b/apps/api/routers/kb.py index 598d560..8910c5b 100644 --- a/apps/api/routers/kb.py +++ b/apps/api/routers/kb.py @@ -1,10 +1,122 @@ -"""知识库实体路由（骨架，T6 填充实现）。 +"""kb 实体路由（FR-KB-01/02/05）：实体 CRUD + 状态机 + 1~3 跳图谱查询。 -GET /kb/items 等；实现契约见 docs/开发任务包/T6-知识库实体与图谱.md。 +写接口暂不挂鉴权以外的前置（后台运营角色属 T-P2-09，先 ``CurrentUserId``， +``updated_by`` 记录操作者）。统一信封见架构文档 §9.2；跨域响应模型在 +``packages/kb/schemas.py``（``apps/api/schemas.py`` 本轮禁改）。 """ from __future__ import annotations -from fastapi import APIRouter +from typing import Annotated + +from fastapi import APIRouter, Depends, Query + +from apps.api.deps import CurrentUserId, SessionDep +from packages.kb.schemas import ( + MAX_GRAPH_HOPS, + EntityCreate, + EntityOut, + EntityStatus, + EntityType, + EntityUpdate, +) +from packages.kb.service import KbService +from packages.shared.models import KbEntity +from packages.shared.response import ok router = APIRouter(prefix="/kb", tags=["kb"]) + + +def kb_service_dep(session: SessionDep) -> KbService: + """``apps/api/deps.py`` 本轮禁改，kb 的服务装配放路由文件内就近完成。""" + return KbService(session) + + +KbServiceDep = Annotated[KbService, Depends(kb_service_dep)] + + +def _serialize(entity: KbEntity) -> dict[str, object]: + return EntityOut( + entity_id=str(entity.entity_id), + entity_type=entity.entity_type, + name=entity.name, + attrs=dict(entity.attrs), + status=entity.status, + version=entity.version, + source=entity.source, + license=entity.license, + updated_by=entity.updated_by, + created_at=entity.created_at, + updated_at=entity.updated_at, + ).model_dump() + + +@router.get("/items", summary="实体列表：entity_type / status 过滤 + 名称模糊 + 分页") +async def list_items( + user_id: CurrentUserId, + kb: KbServiceDep, + entity_type: EntityType | None = Query(default=None, description="12 类本体之一"), + status: EntityStatus | None = Query(default=None), + name: str | None = Query(default=None, max_length=128, description="名称模糊匹配"), + limit: int = Query(default=20, ge=1, le=100), + offset: int = Query(default=0, ge=0), +) -> dict[str, object]: + rows, total = await kb.search_entities( + entity_type=entity_type, + status=status, + name_ilike=name, + limit=limit, + offset=offset, + ) + return ok( + { + "items": [_serialize(row) for row in rows], + "total": total, + "limit": limit, + "offset": offset, + } + ) + + +@router.post("/items", summary="新建实体（draft / version=1 起步）") +async def create_entity( + payload: EntityCreate, + user_id: CurrentUserId, + kb: KbServiceDep, +) -> dict[str, object]: + entity = await kb.create_entity(payload=payload, updated_by=str(user_id)) + return ok(_serialize(entity)) + + +@router.get("/items/{entity_id}", summary="实体详情") +async def get_entity( + entity_id: int, + user_id: CurrentUserId, + kb: KbServiceDep, +) -> dict[str, object]: + entity = await kb.get_entity(entity_id=entity_id) + return ok(_serialize(entity)) + + +@router.patch("/items/{entity_id}", summary="修改内容（draft/reviewing）或推进状态（状态机校验）") +async def update_entity( + entity_id: int, + payload: EntityUpdate, + user_id: CurrentUserId, + kb: KbServiceDep, +) -> dict[str, object]: + entity = await kb.update_entity(entity_id=entity_id, payload=payload, updated_by=str(user_id)) + return ok(_serialize(entity)) + + +@router.get("/items/{entity_id}/graph", summary="从该实体出发的 1~3 跳可解释路径（T-P2-05）") +async def get_entity_graph( + entity_id: int, + user_id: CurrentUserId, + kb: KbServiceDep, + max_hops: int = Query( + default=MAX_GRAPH_HOPS, ge=1, le=MAX_GRAPH_HOPS, description="遍历深度 1~3，超界 422" + ), +) -> dict[str, object]: + result = await kb.traverse_graph(entity_id=entity_id, max_hops=max_hops) + return ok(result.model_dump()) diff --git a/packages/kb/README.md b/packages/kb/README.md index 08ef25e..24e5500 100644 --- a/packages/kb/README.md +++ b/packages/kb/README.md @@ -1,14 +1,27 @@ -# `packages/kb/` — 美学知识库（占位） +# `packages/kb/` — 美学知识库 **职责**（架构 §4.1）：知识实体/关系/规则/素材的读写与发布。 **不负责**：检索算法（调 `packages/retrieval`）。 -## 待实现 +## 本批已落地（T6，T-P2-01/05） -- 表：`kb_entity` `kb_relation` `kb_rule` `kb_chunk`(vector 1024) `kb_asset` - `kb_color` `kb_gap` -- FR-KB-01~12：本体 Schema、自建色板、内容录入、规则库、白底图流水线、 - 草稿→审核→发布→归档流程、知识缺口埋点 +- `schemas.py`：12 类 `EntityType` Literal、`EntityStatus`、CRUD 与图谱 + 响应模型（雪花 ID 一律字符串出参，防 JS 精度丢失） +- `repo.py`：`kb_entity`/`kb_relation`/`kb_color` 数据访问； + `traverse` 为递归 CTE 1~3 跳有界遍历（防环：`path` 数组 + `NOT @>`） +- `service.py`：状态机（draft→reviewing→published→archived，撤稿支线 + draft/reviewing→archived，非法跃迁 40900）、内容修改 version+1 且 + published 只读（要改必须撤稿归档后重录新草稿）、关系写入（自环/重复/ + 端点缺失校验）、色板 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` + +## 待实现（后续批次） + +- `kb_rule` 决策表 CRUD（T7 规则引擎） +- `kb_chunk` 向量+全文混合检索（T8） +- `kb_asset` 白底图流水线、`kb_gap` 知识缺口埋点（T-P2-06+） +- 关系/色板的运营 HTTP 入口（后台角色 T-P2-09 就绪后挂路由） ## 为什么它是数据不是代码 diff --git a/packages/kb/__init__.py b/packages/kb/__init__.py index b8f4b54..caa08d6 100644 --- a/packages/kb/__init__.py +++ b/packages/kb/__init__.py @@ -1 +1,5 @@ -"""kb 域：知识实体/关系/规则/素材的读写与发布。占位。""" +"""kb 域：知识实体/关系/规则/素材的读写与发布。 + +本批已落地实体底座（schemas / repo / service，见本目录 README）； +规则（kb_rule）归 packages/rules + T7，检索（kb_chunk）归 T8。 +""" diff --git a/packages/kb/repo.py b/packages/kb/repo.py new file mode 100644 index 0000000..53aadb0 --- /dev/null +++ b/packages/kb/repo.py @@ -0,0 +1,168 @@ +"""kb 域数据访问层。**只有本模块允许触碰 kb_entity / kb_relation / kb_color。** + +``traverse`` 用递归 CTE 而不是在 Python 里逐跳发查询：TD-09 的结论是 +知识图谱的查询都是「从某实体出发的浅层有界遍历」（1~3 跳），邻接表 + +CTE 在 PG 内一次往返就能完成，不值得为此引入图数据库的第二套存储与运维。 +防环在 CTE 里做（``path`` 数组 + ``@>`` 包含判断），保证有环数据 +（A→B→A）也不会让递归发散。 +""" + +from __future__ import annotations + +from collections.abc import Iterable +from dataclasses import dataclass + +from sqlalchemy import ColumnElement, func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.shared.models import KbColor, KbEntity, KbRelation + +#: 递归遍历 SQL（1~3 跳）。``path`` 记录已经过的实体 id，``rels`` 与逐跳 +#: rel_type 平行——每行结果本身就是一条完整可解释路径，不需要在 Python +#: 侧再拼接。深度上限由参数 ``:max_hops`` 控制（service 层已校验 1~3）。 +#: ``::varchar[]`` 显转型不可省：锚点项推得 varchar(64)[]、递归项推得 +#: varchar[]，PG 要求两段类型完全一致，不转型直接 DatatypeMismatchError。 +_TRAVERSE_SQL = text( + """ + WITH RECURSIVE walk(from_id, to_id, rel_type, hop, path, rels) AS ( + SELECT r.from_id, r.to_id, r.rel_type, 1, + ARRAY[r.from_id, r.to_id]::bigint[], ARRAY[r.rel_type]::varchar[] + FROM kb_relation r + WHERE r.from_id = :start_id + UNION ALL + SELECT r.from_id, r.to_id, r.rel_type, w.hop + 1, + w.path || r.to_id, w.rels || ARRAY[r.rel_type]::varchar[] + FROM kb_relation r + JOIN walk w ON r.from_id = w.to_id + WHERE w.hop < :max_hops + AND NOT w.path @> ARRAY[r.to_id] + ) + SELECT hop, path, rels FROM walk ORDER BY hop, path, rels + """ +) + + +@dataclass(frozen=True, slots=True) +class RawPath: + """CTE 直接吐出的一条路径（防环已由 SQL 保证，len(rel_types)==len(node_ids)-1）。""" + + node_ids: tuple[int, ...] + rel_types: tuple[str, ...] + + +# --- kb_entity --------------------------------------------------------------- + + +async def get_entity(session: AsyncSession, entity_id: int) -> KbEntity | None: + return await session.get(KbEntity, entity_id) + + +async def get_entity_for_update(session: AsyncSession, entity_id: int) -> KbEntity | None: + result = await session.execute( + select(KbEntity).where(KbEntity.entity_id == entity_id).with_for_update() + ) + return result.scalar_one_or_none() + + +async def get_entities_by_ids(session: AsyncSession, ids: Iterable[int]) -> dict[int, KbEntity]: + id_list = list(ids) + if not id_list: + return {} + result = await session.execute(select(KbEntity).where(KbEntity.entity_id.in_(id_list))) + return {entity.entity_id: entity for entity in result.scalars()} + + +async def find_entity_by_name( + session: AsyncSession, *, entity_type: str, name: str +) -> KbEntity | None: + # UNIQUE(entity_type, name) 保证至多一行 + result = await session.execute( + select(KbEntity).where(KbEntity.entity_type == entity_type, KbEntity.name == name) + ) + return result.scalar_one_or_none() + + +def add_entity(session: AsyncSession, entity: KbEntity) -> KbEntity: + session.add(entity) + return entity + + +async def search_entities( + session: AsyncSession, + *, + entity_type: str | None = None, + status: str | None = None, + name_ilike: str | None = None, + limit: int = 20, + offset: int = 0, +) -> tuple[list[KbEntity], int]: + """过滤 + 分页。返回 ``(当页行, 满足过滤条件的总数)``。 + + ``name_ilike`` 不转义 ``%``/``_``：这是后台运营的模糊搜索框， + 通配符被当通配符用符合直觉，也不是注入面（参数化绑定）。 + """ + conds: list[ColumnElement[bool]] = [] + if entity_type is not None: + conds.append(KbEntity.entity_type == entity_type) + if status is not None: + conds.append(KbEntity.status == status) + if name_ilike: + conds.append(KbEntity.name.ilike(f"%{name_ilike}%")) + + total = int(await session.scalar(select(func.count()).select_from(KbEntity).where(*conds)) or 0) + result = await session.execute( + select(KbEntity) + .where(*conds) + .order_by(KbEntity.created_at.desc(), KbEntity.entity_id.desc()) + .limit(limit) + .offset(offset) + ) + return list(result.scalars().all()), total + + +# --- kb_relation ------------------------------------------------------------- + + +async def find_relation( + session: AsyncSession, *, from_id: int, rel_type: str, to_id: int +) -> KbRelation | None: + result = await session.execute( + select(KbRelation).where( + KbRelation.from_id == from_id, + KbRelation.rel_type == rel_type, + KbRelation.to_id == to_id, + ) + ) + return result.scalar_one_or_none() + + +def add_relation(session: AsyncSession, relation: KbRelation) -> KbRelation: + session.add(relation) + return relation + + +async def traverse(session: AsyncSession, *, from_id: int, max_hops: int) -> list[RawPath]: + """从 ``from_id`` 出发沿 from→to 方向做 ``max_hops`` 跳的有界遍历。 + + 深度合法性（1~3）由 service 层校验，这里信任调用方。返回按 + （跳数, 路径, 关系序列）排序的全部路径——同一终点多条路径都保留， + 「为什么推荐」恰恰需要把每条依据都摆出来。 + """ + result = await session.execute(_TRAVERSE_SQL, {"start_id": from_id, "max_hops": max_hops}) + return [RawPath(node_ids=tuple(row.path), rel_types=tuple(row.rels)) for row in result] + + +# --- kb_color ---------------------------------------------------------------- + + +async def list_colors(session: AsyncSession, *, season_code: str | None = None) -> list[KbColor]: + stmt = select(KbColor).order_by(KbColor.season_code, KbColor.color_id) + if season_code is not None: + stmt = stmt.where(KbColor.season_code == season_code) + result = await session.execute(stmt) + return list(result.scalars().all()) + + +def add_color(session: AsyncSession, color: KbColor) -> KbColor: + session.add(color) + return color diff --git a/packages/kb/schemas.py b/packages/kb/schemas.py new file mode 100644 index 0000000..57b897e --- /dev/null +++ b/packages/kb/schemas.py @@ -0,0 +1,166 @@ +"""kb 域请求/响应模型（架构文档 §6.2，FR-KB-01/02/05）. + +``apps/api/schemas.py`` 是全局共享契约文件，本轮并行开发禁改；kb 接口的 +请求/响应模型统一放本包，由路由层装配进 ``ok(...)`` 信封（NFR-MAINT-01 +「契约先行」：客户端类型从 OpenAPI 生成）。 + +雪花 ID 一律以字符串出现在载荷里：JS Number 只有 2^53 精度，直接给数字 +会被四舍五入成另一个实体。 +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any, Literal, get_args + +from pydantic import BaseModel, Field + +#: 12 类实体（知识库文档 §3.1 本体）。写死成 Literal 而不是普通 str： +#: 新增类目是需要评审的低频动作，宁可发版，也不给运营侧随手写错值留口子。 +EntityType = Literal[ + "item", + "color", + "style", + "body_shape", + "face_hair", + "occasion", + "material", + "accessory", + "makeup", + "rule", + "outfit", + "trend", +] + +ENTITY_TYPES: tuple[str, ...] = get_args(EntityType) + +#: 生命周期四态。合法跃迁矩阵见 :mod:`packages.kb.service`（非法跃迁 40900）。 +EntityStatus = Literal["draft", "reviewing", "published", "archived"] + +ENTITY_STATUSES: tuple[str, ...] = get_args(EntityStatus) + +#: kb_color.role：基础色 / 点缀色 / 回避色（TD-18 自建色板）。 +ColorRole = Literal["base", "accent", "avoid"] + +#: 图谱遍历深度上限（TD-09：浅层有界遍历，1~3 跳，不上图数据库）。 +#: 放在这里是因为路由 Query 与 service 校验必须引用同一个数字，两处硬编码必然漂移。 +MAX_GRAPH_HOPS = 3 + + +class EntityCreate(BaseModel): + """``POST /kb/items`` 载荷。新实体固定从 draft/v1 起步，不收客户端指定状态。""" + + entity_type: EntityType = Field(description="12 类本体之一，见 EntityType") + name: str = Field(min_length=1, max_length=128, description="同类目内唯一（UNIQUE 守护）") + attrs: dict[str, Any] = Field( + default_factory=dict, description="类目差异化字段，由 attrs JSONB 承载（架构 §6.2）" + ) + source: str | None = Field(default=None, max_length=255, description="知识来源，版权审计用") + license: str | None = Field(default=None, max_length=64, description="内容许可证标识") + + +class EntityUpdate(BaseModel): + """``PATCH /kb/items/{id}`` 载荷：内容字段与状态可单独或同时提交。 + + - 只出现内容字段（name/attrs/source/license）→ 内容修改，version+1， + 仅 draft/reviewing 状态允许（published 只读，见 service）。 + - 只出现 status → 纯状态跃迁，走状态机校验，version 不变。 + - 与当前状态相同的 status → 幂等重放，视为 no-op 成功（客户端重试友好）。 + """ + + name: str | None = Field(default=None, min_length=1, max_length=128) + attrs: dict[str, Any] | None = None + source: str | None = Field(default=None, max_length=255) + license: str | None = Field(default=None, max_length=64) + status: EntityStatus | None = None + + def content_fields(self) -> dict[str, Any]: + """本次请求显式提供的内容字段（PATCH 语义：未提供的字段不动）。 + + ``exclude_unset`` 保证显式传 ``null``（清空 source/license）与 + 「未传」可区分——这是 PATCH 覆盖语义的最低要求。 + """ + return self.model_dump(exclude_unset=True, exclude={"status"}) + + +class EntityOut(BaseModel): + """实体对外视图（``data`` 载荷，单条或列表元素）。""" + + entity_id: str = Field(description="雪花 ID，字符串形式") + entity_type: str + name: str + attrs: dict[str, Any] + status: str + version: int = Field(description="内容版本号，每次内容修改 +1") + source: str | None + license: str | None + updated_by: str | None + created_at: datetime + updated_at: datetime + + +class RelationCreate(BaseModel): + """``kb_relation`` 三元组写入载荷。 + + 本批次未暴露 HTTP 路由（关系录入属运营后台 T-P2-09 的范围），但写入 + 校验（自环/重复/端点存在）在 service 层先行落地，集成测试直接驱动。 + """ + + from_entity_id: int + to_entity_id: int + rel_type: str = Field(min_length=1, max_length=64, description="关系类型，如 suits/fits") + weight: Decimal = Field(default=Decimal("1"), ge=0, le=Decimal("9.999"), description="权重") + confidence: Decimal = Field( + default=Decimal("1"), ge=0, le=Decimal("9.999"), description="知识置信度" + ) + evidence: str | None = Field(default=None, max_length=255, description="证据/出处") + + +class GraphNode(BaseModel): + """路径上的实体摘要：图谱路径要能「读给人听」，只有 ID 不可解释。""" + + entity_id: str + entity_type: str + name: str + status: str + + +class GraphHop(BaseModel): + """一跳：从哪个实体经什么关系到哪个实体（T-P2-05 可解释路径的最小单位）。""" + + from_entity_id: str + rel_type: str + to_entity_id: str + + +class GraphPath(BaseModel): + """一条从起点出发的完整路径：节点序列 + 每跳 rel_type。 + + ``len(rel_types) == len(nodes) - 1``；同一终点可能有多条路径， + 遍历结果不去重合并——「为什么推荐」恰恰需要把每条依据都摆出来。 + """ + + nodes: list[GraphNode] + hops: list[GraphHop] + + +class GraphQueryResult(BaseModel): + """``GET /kb/items/{id}/graph`` 载荷。""" + + from_entity_id: str + max_hops: int = Field(description="本次遍历深度（1~3）") + paths: list[GraphPath] = Field(description="按跳数升序的全部可解释路径") + + +class ColorOut(BaseModel): + """色板条目（TD-18 自建色板）。Lab 供后续 ΔE 数值匹配（T8 检索），此处只读展示。""" + + color_id: str + season_code: str = Field(description="季型代码，如 spring_warm") + role: str = Field(description="base=基础色 / accent=点缀色 / avoid=回避色") + lab_l: float + lab_a: float + lab_b: float + hex: str = Field(description="#RRGGBB") + name_cn: str diff --git a/packages/kb/service.py b/packages/kb/service.py new file mode 100644 index 0000000..931044c --- /dev/null +++ b/packages/kb/service.py @@ -0,0 +1,392 @@ +"""kb 域 service：实体生命周期、关系写入、图谱遍历、自建色板（FR-KB-01/02/05）. + +状态机与格式校验拆成模块级纯函数（不碰 DB，单测直接覆盖矩阵）； +``KbService`` 只负责组装、落库与跨表编排。所有写操作只 ``flush`` 不 +``commit``——事务边界交给 ``session_dep`` 的统一提交/回滚，与 task / +billing 域一致。 +""" + +from __future__ import annotations + +import re +from datetime import UTC, datetime +from decimal import Decimal + +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.domain.ids import next_id +from packages.kb import repo +from packages.kb.schemas import ( + MAX_GRAPH_HOPS, + ColorOut, + EntityCreate, + EntityUpdate, + GraphHop, + GraphNode, + GraphPath, + GraphQueryResult, + RelationCreate, +) +from packages.shared.errors import AppError, ErrorCode, NotFoundError +from packages.shared.logging import get_logger +from packages.shared.models import KbColor, KbEntity, KbRelation + +logger = get_logger(__name__) + +#: 合法跃迁矩阵（知识库文档「发布回滚」语义）： +#: draft → reviewing → published → archived 主线； +#: draft/reviewing → archived 撤稿支线； +#: archived 终态。其余（含 published→draft、reviewing→draft、自跃迁） +#: 一律非法——想改已发布内容，就撤稿归档后以新草稿重录，让新版本重走评审。 +LEGAL_TRANSITIONS: frozenset[tuple[str, str]] = frozenset( + { + ("draft", "reviewing"), + ("reviewing", "published"), + ("published", "archived"), + ("draft", "archived"), + ("reviewing", "archived"), + } +) + +#: 内容可编辑的状态。published 内容只读（版本语义的基石），archived 终态。 +CONTENT_EDITABLE_STATUSES: frozenset[str] = frozenset({"draft", "reviewing"}) + +_HEX_COLOR_PATTERN = re.compile(r"^#[0-9A-Fa-f]{6}$") +_COLOR_ROLES: frozenset[str] = frozenset({"base", "accent", "avoid"}) + + +# --- 纯校验函数（单测直接覆盖，不触碰 DB） ---------------------------------- + + +def can_transition(current: str, target: str) -> bool: + return (current, target) in LEGAL_TRANSITIONS + + +def assert_transition(current: str, target: str) -> None: + """非法跃迁 → ``STATE_CONFLICT``（409）。矩阵见 :data:`LEGAL_TRANSITIONS`。""" + if not can_transition(current, target): + raise AppError( + ErrorCode.STATE_CONFLICT, + message="实体状态不允许该跃迁", + data={"current": current, "target": target}, + ) + + +def ensure_content_editable(status: str) -> None: + """published/archived 的内容不可直接修改（version 递增的前提）。""" + if status not in CONTENT_EDITABLE_STATUSES: + raise AppError( + ErrorCode.STATE_CONFLICT, + message=f"{status} 状态的实体内容不可修改", + hint="已发布内容要改：先撤稿归档，再以新草稿重录，新版本重走评审", + data={"status": status}, + ) + + +def ensure_valid_hex(hex_code: str) -> None: + """色板 hex 必须 ``#RRGGBB``（大小写不敏感，入库前不归一化，展示原样保留）。""" + if _HEX_COLOR_PATTERN.fullmatch(hex_code) is None: + raise AppError( + ErrorCode.BAD_REQUEST, + message="色板 hex 格式非法，应为 #RRGGBB", + data={"hex": hex_code}, + ) + + +def ensure_valid_color_role(role: str) -> None: + if role not in _COLOR_ROLES: + raise AppError( + ErrorCode.BAD_REQUEST, + message="色板 role 非法，应为 base/accent/avoid", + data={"role": role}, + ) + + +def ensure_valid_hops(max_hops: int) -> None: + """遍历深度 1~3（TD-09 有界遍历；上限同时也是防环的保险丝）。""" + if not 1 <= max_hops <= MAX_GRAPH_HOPS: + raise AppError( + ErrorCode.BAD_REQUEST, + message=f"max_hops 超出允许范围 1~{MAX_GRAPH_HOPS}", + data={"max_hops": max_hops}, + ) + + +# --- service ----------------------------------------------------------------- + + +class KbService: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + # -- 实体 ----------------------------------------------------------------- + + async def create_entity( + self, *, payload: EntityCreate, updated_by: str | None = None + ) -> KbEntity: + """新建实体：固定 draft / version=1 起步。同名同类目拒绝（400）。""" + existing = await repo.find_entity_by_name( + self._session, entity_type=payload.entity_type, name=payload.name + ) + if existing is not None: + raise AppError( + ErrorCode.BAD_REQUEST, + message="同名同类目的实体已存在", + data={"entity_type": payload.entity_type, "name": payload.name}, + ) + entity = repo.add_entity( + self._session, + KbEntity( + entity_id=next_id(), + entity_type=payload.entity_type, + name=payload.name, + attrs=payload.attrs, + status="draft", + version=1, + source=payload.source, + license=payload.license, + updated_by=updated_by, + ), + ) + try: + await self._session.flush() + except IntegrityError as exc: + # 并发同名是唯一预期内的约束冲突（UNIQUE(entity_type,name) 兜底） + raise AppError( + ErrorCode.BAD_REQUEST, + message="同名同类目的实体已存在", + data={"entity_type": payload.entity_type, "name": payload.name}, + ) from exc + logger.info( + "kb_entity_created", + entity_id=entity.entity_id, + entity_type=entity.entity_type, + name=entity.name, + ) + return entity + + async def get_entity(self, *, entity_id: int) -> KbEntity: + entity = await repo.get_entity(self._session, entity_id) + if entity is None: + raise NotFoundError("实体不存在") + return entity + + async def search_entities( + self, + *, + entity_type: str | None = None, + status: str | None = None, + name_ilike: str | None = None, + limit: int = 20, + offset: int = 0, + ) -> tuple[list[KbEntity], int]: + return await repo.search_entities( + self._session, + entity_type=entity_type, + status=status, + name_ilike=name_ilike, + limit=limit, + offset=offset, + ) + + async def update_entity( + self, *, entity_id: int, payload: EntityUpdate, updated_by: str | None = None + ) -> KbEntity: + """内容修改与状态跃迁共用一个 PATCH 入口，规则： + + - 内容字段仅 draft/reviewing 可改，每次内容修改 version+1； + - 状态变化必须走合法跃迁矩阵；与当前相同的 status 是幂等 no-op； + - 先全部校验、后统一变更——半个请求不会落库。 + """ + entity = await repo.get_entity_for_update(self._session, entity_id) + if entity is None: + raise NotFoundError("实体不存在") + + content = payload.content_fields() + if content: + ensure_content_editable(entity.status) + if payload.status is not None and payload.status != entity.status: + assert_transition(entity.status, payload.status) + + if content: + for field, value in content.items(): + setattr(entity, field, value) + entity.version += 1 + if payload.status is not None: + entity.status = payload.status + if content or payload.status is not None: + entity.updated_by = updated_by or entity.updated_by + entity.updated_at = datetime.now(UTC) + try: + await self._session.flush() + except IntegrityError as exc: + # 改名撞上另一条同类目实体 + raise AppError( + ErrorCode.BAD_REQUEST, + message="同名同类目的实体已存在", + data={"entity_id": str(entity_id)}, + ) from exc + logger.info( + "kb_entity_updated", + entity_id=entity_id, + version=entity.version, + status=entity.status, + content_changed=bool(content), + ) + return entity + + # -- 关系 ----------------------------------------------------------------- + + async def add_relation(self, *, payload: RelationCreate) -> KbRelation: + """写入三元组。自环与重复（UNIQUE 三元组）拒绝（400），端点缺失 404。""" + if payload.from_entity_id == payload.to_entity_id: + raise AppError( + ErrorCode.BAD_REQUEST, + message="不允许自环关系", + data={"entity_id": str(payload.from_entity_id)}, + ) + for end_id in (payload.from_entity_id, payload.to_entity_id): + if await repo.get_entity(self._session, end_id) is None: + raise NotFoundError("关联实体不存在") + duplicate = await repo.find_relation( + self._session, + from_id=payload.from_entity_id, + rel_type=payload.rel_type, + to_id=payload.to_entity_id, + ) + if duplicate is not None: + raise AppError( + ErrorCode.BAD_REQUEST, + message="该 from/rel_type/to 三元组已存在", + data={ + "from_entity_id": str(payload.from_entity_id), + "rel_type": payload.rel_type, + "to_entity_id": str(payload.to_entity_id), + }, + ) + relation = repo.add_relation( + self._session, + KbRelation( + relation_id=next_id(), + from_id=payload.from_entity_id, + rel_type=payload.rel_type, + to_id=payload.to_entity_id, + weight=payload.weight, + confidence=payload.confidence, + evidence=payload.evidence, + ), + ) + await self._session.flush() + logger.info( + "kb_relation_added", + from_id=payload.from_entity_id, + rel_type=payload.rel_type, + to_id=payload.to_entity_id, + ) + return relation + + # -- 图谱 ----------------------------------------------------------------- + + async def traverse_graph( + self, *, entity_id: int, max_hops: int = MAX_GRAPH_HOPS + ) -> GraphQueryResult: + """1~3 跳可解释路径（T-P2-05）。 + + CTE 只回 id 与 rel_type 序列；节点摘要（类型/名称/状态）由第二次 + 查询补齐——遍历有界，涉及的实体集合很小，两次往返远比在递归 SQL + 里 JOIN 实体表好维护。 + """ + ensure_valid_hops(max_hops) + await self.get_entity(entity_id=entity_id) # 起点 404 语义保持一致 + + raw_paths = await repo.traverse(self._session, from_id=entity_id, max_hops=max_hops) + node_ids: set[int] = set() + for raw in raw_paths: + node_ids.update(raw.node_ids) + nodes_map = await repo.get_entities_by_ids(self._session, node_ids) + + paths: list[GraphPath] = [] + for raw in raw_paths: + nodes = [_node_summary(nodes_map, node_id) for node_id in raw.node_ids] + hops = [ + GraphHop( + from_entity_id=str(hop_from), + rel_type=rel_type, + to_entity_id=str(hop_to), + ) + # node_ids 比 rel_types 多一个元素（n 个节点 n-1 跳），首尾 + # 错位 zip 即得「第 i 跳 = nodes[i] --rels[i]--> nodes[i+1]」 + for hop_from, rel_type, hop_to in zip( + raw.node_ids[:-1], raw.rel_types, raw.node_ids[1:], strict=True + ) + ] + paths.append(GraphPath(nodes=nodes, hops=hops)) + return GraphQueryResult(from_entity_id=str(entity_id), max_hops=max_hops, paths=paths) + + # -- 色板（TD-18 自建色板） ------------------------------------------------ + + async def add_color( + self, + *, + season_code: str, + role: str, + lab_l: float, + lab_a: float, + lab_b: float, + hex_code: str, + name_cn: str, + ) -> ColorOut: + """录入色板条目。role/hex 校验在 service 层（运营后台 T-P2-09 直接复用）。""" + ensure_valid_color_role(role) + ensure_valid_hex(hex_code) + color = repo.add_color( + self._session, + KbColor( + color_id=next_id(), + season_code=season_code, + role=role, + lab_l=Decimal(str(lab_l)), + lab_a=Decimal(str(lab_a)), + lab_b=Decimal(str(lab_b)), + hex=hex_code, + name_cn=name_cn, + ), + ) + await self._session.flush() + logger.info("kb_color_added", season_code=season_code, role=role, hex=hex_code) + return _color_out(color) + + async def list_colors(self, *, season_code: str | None = None) -> list[ColorOut]: + rows = await repo.list_colors(self._session, season_code=season_code) + return [_color_out(row) for row in rows] + + +def _node_summary(nodes_map: dict[int, KbEntity], node_id: int) -> GraphNode: + """路径节点摘要。关系端点在写入时校验过存在，理论必中；万一实体行 + 被运维手工删除，退化为占位摘要而不是让整个图查询 500。""" + entity = nodes_map.get(node_id) + if entity is None: + logger.warning("kb_graph_node_missing", entity_id=node_id) + return GraphNode( + entity_id=str(node_id), entity_type="unknown", name="unknown", status="unknown" + ) + return GraphNode( + entity_id=str(entity.entity_id), + entity_type=entity.entity_type, + name=entity.name, + status=entity.status, + ) + + +def _color_out(color: KbColor) -> ColorOut: + return ColorOut( + color_id=str(color.color_id), + season_code=color.season_code, + role=color.role, + lab_l=float(color.lab_l), + lab_a=float(color.lab_a), + lab_b=float(color.lab_b), + hex=color.hex, + name_cn=color.name_cn, + ) diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..1beaa37 --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1,3 @@ +# pytest prepend 导入模式按 basename 注册模块：unit 与 integration 下的 +# 同名测试文件（如 test_kb_entity.py，任务契约指定两端同名）会撞模块名。 +# 加包标记后模块名唯一化为 tests.integration.，与 tests/__init__.py 一致。 diff --git a/tests/integration/test_kb_entity.py b/tests/integration/test_kb_entity.py new file mode 100644 index 0000000..a37d389 --- /dev/null +++ b/tests/integration/test_kb_entity.py @@ -0,0 +1,406 @@ +"""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"] diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..3719c23 --- /dev/null +++ b/tests/unit/__init__.py @@ -0,0 +1,3 @@ +# pytest prepend 导入模式按 basename 注册模块：unit 与 integration 下的 +# 同名测试文件（如 test_kb_entity.py，任务契约指定两端同名）会撞模块名。 +# 加包标记后模块名唯一化为 tests.unit.，与 tests/__init__.py 一致。 diff --git a/tests/unit/test_kb_entity.py b/tests/unit/test_kb_entity.py new file mode 100644 index 0000000..ca43d32 --- /dev/null +++ b/tests/unit/test_kb_entity.py @@ -0,0 +1,222 @@ +"""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