From 3961b6fbfa9c0064f60e3dc5fd51439e571bed4b Mon Sep 17 00:00:00 2001 From: xsl Date: Fri, 14 Aug 2026 21:19:31 +0800 Subject: [PATCH] feat(admin): minimal console for KB content operations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit apps/admin 独立应用：X-Admin-Token 鉴权（空=全部路由不注册=404）、实体/规则 维护与状态机流转、发布编排（publish/archive → 规则缓存 invalidate + retrieval invalidate_cache 双失效）、素材登记与审核（tryon_ready 门禁透传）、 缺口看板（T-P2-09，FR-ADM-01~05）。V1 共享令牌鉴权，无角色无审计字段， 操作留证靠 structlog request_id。 测试：单测 tests/unit/test_admin_app.py（10 例：空令牌 404、错/缺/非 ASCII 令牌 40100、正确令牌 200 假服务、发布编排接线、幂等重放无副作用）； 集成 tests/integration/test_admin_console.py（5 例：顾问全流程不碰客户端 API、双失效副作用可观察、非法跃迁 40900、素材审核门禁、缺口看板排序、 令牌 401）。483 单测 + 43 集成全绿；ruff/mypy/lint-imports 全绿。 偏差说明： 1. packages/retrieval.repo.list_gaps 有缺陷：对 select(KbGap) 的结果调 .mappings()，行键是实体名而非列名，GapSummary(row) 取 row.gap_id 必然 AttributeError（该路径无既有测试覆盖，接真库即暴露）。packages/** 本 任务禁改，故缺口查询在 apps/admin/routers/gaps.py 以等价 SQL 自实现 （hit_count 降序、同频次按 last_seen_at 降序、limit/offset 语义一致）； 建议合并阶段修复上游后切回 retrieval.list_gaps。 2. 实体 transition 的 target 收全量四态（任务文档示例仅列 reviewing/ published/archived）：按任务测试要求 published→draft 必须返回 40900 （状态机语义冲突）而非 schema 422；draft 作为目标由服务层跃迁矩阵恒拒。 3. 规则路由以 rule_key 为路径参数（任务文档写 /rules/{id}，但 RuleService 全程以 rule_key 寻址，无按数字 id 的接口，以代码为准）。 4. 发布编排触发面含 archived（撤稿）：依据 retrieval.invalidate_cache 的 自身契约「发布/归档知识后调用」，比文档字面的「发布成功后」多了撤稿 方向；reviewing 等其余状态不触发。 5. updated_by/author 记 "admin-console"：V1 共享令牌无操作者身份，字段 留渠道来源；运营账号体系属 V2（见包 docstring 的升级路径）。 --- apps/admin/__init__.py | 20 +- apps/admin/app.py | 81 +++++ apps/admin/deps.py | 90 ++++++ apps/admin/publishing.py | 35 +++ apps/admin/routers/__init__.py | 1 + apps/admin/routers/assets.py | 71 +++++ apps/admin/routers/gaps.py | 58 ++++ apps/admin/routers/kb.py | 132 ++++++++ apps/admin/routers/rules.py | 82 +++++ apps/admin/schemas.py | 78 +++++ tests/integration/test_admin_console.py | 334 ++++++++++++++++++++ tests/unit/test_admin_app.py | 393 ++++++++++++++++++++++++ 12 files changed, 1374 insertions(+), 1 deletion(-) create mode 100644 apps/admin/app.py create mode 100644 apps/admin/deps.py create mode 100644 apps/admin/publishing.py create mode 100644 apps/admin/routers/__init__.py create mode 100644 apps/admin/routers/assets.py create mode 100644 apps/admin/routers/gaps.py create mode 100644 apps/admin/routers/kb.py create mode 100644 apps/admin/routers/rules.py create mode 100644 apps/admin/schemas.py create mode 100644 tests/integration/test_admin_console.py create mode 100644 tests/unit/test_admin_app.py diff --git a/apps/admin/__init__.py b/apps/admin/__init__.py index 3ee4f89..fb2518e 100644 --- a/apps/admin/__init__.py +++ b/apps/admin/__init__.py @@ -1 +1,19 @@ -"""极简运营后台（占位，FR-ADM-01~05）。""" +"""极简运营后台（T-P2-09，FR-ADM-01~05）：知识内容的维护 / 审核 / 发布 / 回滚入口。 + +与 ``apps/api`` 平级的独立 FastAPI 应用（部署上分离）：顾问从这里完成实体、 +规则、素材与缺口看板的全部操作，不再需要工程介入——这是 M2 出口的成败标志。 + +V1 极简鉴权的边界与升级路径： + +- **共享令牌，无角色**：全后台只有一把 ``X-Admin-Token``（``ADMIN_TOKEN``）， + 建库与审核是同一个人。没有「编辑 / 审核分离」的角色体系。 +- **无审计日志字段**：谁在什么时间改了什么，V1 只能靠 structlog 的 + request_id 关联访问日志（service 层的每个写操作都有结构化事件兜底）， + 没有独立的审计表。令牌泄露 = 全权限泄露，轮换令牌是唯一的处置手段。 +- **升级路径**：运营账号体系（登录、角色、操作审计）属 V2；届时把 + ``apps.admin.deps.admin_auth`` 换成账号会话依赖即可，路由与服务层不动。 +- **空令牌 = 未部署**：``ADMIN_TOKEN`` 为空时应用装配期不注册任何业务路由 + （docs/openapi 一并关闭），所有 admin 端点 404——避免带病上线。 + +应用工厂见 :func:`apps.admin.app.create_admin_app`。 +""" diff --git a/apps/admin/app.py b/apps/admin/app.py new file mode 100644 index 0000000..150b011 --- /dev/null +++ b/apps/admin/app.py @@ -0,0 +1,81 @@ +"""admin 应用工厂（T-P2-09）：极简运营后台的装配层。 + +与客户端 API 的装配差异只有三点：路由前缀 ``/admin/api/v1``（区分流量）、 +鉴权换成 ``X-Admin-Token``、**空令牌时一个业务路由都不注册**——「未部署」 +的语义用装配期决定，而不是每个请求里判空（404 更干净，扫描器也探不到面）。 +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from apps.admin.routers import assets, gaps, kb, rules +from packages.shared.config import Settings, get_settings +from packages.shared.db import dispose_engine, get_session_factory +from packages.shared.handlers import install_exception_handlers +from packages.shared.logging import configure_logging, get_logger +from packages.shared.middleware import RequestContextMiddleware +from packages.shared.observability import setup_tracing + +logger = get_logger(__name__) + +#: 与客户端 API（/api/v1）区分的固定前缀；不提供 /v1 别名（后台没有历史包袱）。 +ADMIN_API_PREFIX = "/admin/api/v1" + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncIterator[None]: + settings: Settings = app.state.admin_settings + # 启动即触碰共享 engine（懒加载单例，复用 packages.shared.db）： + # 连接配置有问题应该在部署的那一刻暴露，而不是等顾问点第一个按钮。 + get_session_factory() + logger.info( + "admin_started", + env=settings.app_env, + routes_registered=bool(settings.admin_token), + ) + try: + yield + finally: + await dispose_engine() + logger.info("admin_stopped") + + +def create_admin_app(settings: Settings | None = None) -> FastAPI: + cfg = settings or get_settings() + configure_logging( + level=cfg.log_level, json_output=cfg.log_json, extra_redact_keys=cfg.extra_redact_keys + ) + + # 空令牌 = 等价未部署：业务路由、openapi、docs 全部不装配。 + deployed = bool(cfg.admin_token) + app = FastAPI( + title="美搭运营后台", + version="0.1.0", + description=( + "极简运营后台：知识实体/规则维护、素材登记审核、发布编排与缺口看板" + "（T-P2-09）。鉴权：请求头 X-Admin-Token。" + ), + lifespan=lifespan, + openapi_url=f"{ADMIN_API_PREFIX}/openapi.json" if deployed else None, + docs_url=f"{ADMIN_API_PREFIX}/docs" if deployed and cfg.is_local else None, + redoc_url=None, + ) + app.state.admin_settings = cfg + + app.add_middleware(RequestContextMiddleware) + install_exception_handlers(app) + setup_tracing(app, cfg) + + if deployed: + for router in (kb.router, rules.router, assets.router, gaps.router): + app.include_router(router, prefix=ADMIN_API_PREFIX) + else: + logger.warning( + "admin_routes_disabled", + reason="ADMIN_TOKEN 为空：全部 admin 端点不注册（等价未部署）", + ) + return app diff --git a/apps/admin/deps.py b/apps/admin/deps.py new file mode 100644 index 0000000..795a2f1 --- /dev/null +++ b/apps/admin/deps.py @@ -0,0 +1,90 @@ +"""admin 应用的依赖装配：鉴权 + session + 各域 service。 + +**不 import ``apps.api`` 的任何东西**（import-linter：apps 之间互不依赖）—— +session 依赖在这里按同样语义自建一份；``ok``/``AppError`` 等统一信封件来自 +``packages.shared``，是允许的共享层。 +""" + +from __future__ import annotations + +import hmac +from collections.abc import AsyncIterator +from typing import Annotated + +from fastapi import Depends, Request +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.kb.service import KbAssetService, KbService +from packages.rules.service import RuleService +from packages.shared.config import Settings +from packages.shared.db import get_session_factory +from packages.shared.errors import AppError, ErrorCode + +#: 鉴权请求头。共享令牌方案（见包 docstring 的 V1 边界说明）。 +ADMIN_TOKEN_HEADER = "X-Admin-Token" + + +def admin_settings(request: Request) -> Settings: + """取应用工厂装配时挂上的 settings。 + + 鉴权口径必须与应用装配口径同源：工厂收到的 settings 决定路由是否注册， + 也必须决定令牌是否匹配——两处各读一遍 ``get_settings()`` 会给「换令牌 + 重启了一半」的窗口留缝。 + """ + settings: Settings = request.app.state.admin_settings + return settings + + +async def admin_auth(request: Request) -> None: + """``X-Admin-Token`` 精确匹配校验，失败 → 40100（UNAUTHENTICATED）。 + + 用 ``hmac.compare_digest`` 而非 ``==``：令牌比较不该泄露「对到了第几位」。 + 先 encode 成 bytes 再比较——str 版本的 compare_digest 遇到非 ASCII 头会 + 直接 TypeError 变 500，那等于给攻击者一个免费的服务端异常。 + """ + settings = admin_settings(request) + expected = settings.admin_token.encode() + if not expected: + # 正常到不了这里（空令牌时路由根本没注册）；留作纵深防御： + # 万一未来有人在别的 app 里复用这个依赖，也不许空配置放行。 + raise AppError(ErrorCode.UNAUTHENTICATED, message="后台未配置访问令牌") + provided = request.headers.get(ADMIN_TOKEN_HEADER, "").encode() + if not hmac.compare_digest(provided, expected): + raise AppError( + ErrorCode.UNAUTHENTICATED, + message="管理令牌缺失或不正确", + hint=f"请在请求头携带 {ADMIN_TOKEN_HEADER}", + ) + + +async def session_dep() -> AsyncIterator[AsyncSession]: + """请求级事务作用域：成功统一 commit、异常统一 rollback。 + + 与 ``apps.api`` 的同名依赖语义一致（包内自建，不跨 app import）。 + """ + factory = get_session_factory() + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + +def kb_service_dep(session: SessionDep) -> KbService: + return KbService(session) + + +def kb_asset_service_dep(session: SessionDep) -> KbAssetService: + return KbAssetService(session) + + +def rule_service_dep(session: SessionDep) -> RuleService: + return RuleService(session) + + +SessionDep = Annotated[AsyncSession, Depends(session_dep)] +KbServiceDep = Annotated[KbService, Depends(kb_service_dep)] +KbAssetServiceDep = Annotated[KbAssetService, Depends(kb_asset_service_dep)] +RuleServiceDep = Annotated[RuleService, Depends(rule_service_dep)] diff --git a/apps/admin/publishing.py b/apps/admin/publishing.py new file mode 100644 index 0000000..4b759e7 --- /dev/null +++ b/apps/admin/publishing.py @@ -0,0 +1,35 @@ +"""发布编排：知识变更后的下游缓存「自然失效」（T-P2-10 的入口）。 + +运营后台是唯一有「发布 / 撤稿」动作的地方，因此失效编排长在这里而不是 +service 层：``KbService``/``RuleService`` 管状态机，跨域的缓存联动（规则缓存 ++ 检索缓存）是应用层的编排职责。T-P2-10 要求的「发布后自然失效」——顾问点 +发布，检索立刻能看见新知识，不需要任何人记得去刷缓存。 +""" + +from __future__ import annotations + +from packages.retrieval import service as retrieval_service +from packages.rules import service as rules_service +from packages.shared.logging import get_logger + +logger = get_logger(__name__) + +#: 落地后必须让下游缓存失效的目标状态：``published`` 让新知识对检索可见， +#: ``archived`` 让旧知识从检索消失——两个方向都是线上知识集合的真实变更。 +#: （依据 ``packages.retrieval.service.invalidate_cache`` 的契约：「发布/归档 +#: 知识后调用」。） +CACHE_AFFECTING_TARGETS: frozenset[str] = frozenset({"published", "archived"}) + + +def invalidate_downstream_caches(*, domain: str, key: str, target: str) -> None: + """规则进程内缓存 + 检索结果缓存整体失效。 + + - 规则缓存：``RuleService.publish/archive`` 内部已会失效一次，这里显式 + 再调是编排契约的一部分（实体发布同样要失效规则缓存——实体变更可能 + 改变规则命中的知识基础），幂等无副作用； + - 检索缓存：当前是失效版本号桩，T14 会在背后接真正的进程内缓存实现， + 调用点无需再改。 + """ + rules_service.invalidate_published_cache() + retrieval_service.invalidate_cache() + logger.info("admin_knowledge_invalidated", domain=domain, key=key, target=target) diff --git a/apps/admin/routers/__init__.py b/apps/admin/routers/__init__.py new file mode 100644 index 0000000..6e41ccb --- /dev/null +++ b/apps/admin/routers/__init__.py @@ -0,0 +1 @@ +"""admin 路由包：每域一个 router 文件，全部经 ``admin_auth`` 鉴权。""" diff --git a/apps/admin/routers/assets.py b/apps/admin/routers/assets.py new file mode 100644 index 0000000..b89d930 --- /dev/null +++ b/apps/admin/routers/assets.py @@ -0,0 +1,71 @@ +"""素材路由（FR-ADM-03）：图像资产登记与审核。 + +素材由运营流水线高频登记（T-P2-07/15），审核是唯一能把 ``tryon_ready`` +打开的入口——服务层已强制「tryon_ready 只能随 approved 打开」，这里只做 +透传，约束在 HTTP 层的表现是 400（BAD_REQUEST）。 +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Query + +from apps.admin.deps import KbAssetServiceDep, admin_auth +from packages.kb.schemas import AssetCreate, AssetOut, AssetReview +from packages.shared.models import KbAsset +from packages.shared.response import ok + +router = APIRouter(prefix="/assets", tags=["admin:assets"], dependencies=[Depends(admin_auth)]) + + +def _serialize(asset: KbAsset) -> dict[str, Any]: + return AssetOut( + asset_id=str(asset.asset_id), + entity_id=str(asset.entity_id), + asset_type=asset.asset_type, + cos_key=asset.cos_key, + spec_check=dict(asset.spec_check), + tryon_ready=asset.tryon_ready, + review_status=asset.review_status, + license=asset.license, + created_at=asset.created_at, + ).model_dump() + + +@router.post("", summary="登记素材（登记时 tryon_ready 恒为 False，与请求体取值无关）") +async def register_asset(payload: AssetCreate, assets: KbAssetServiceDep) -> dict[str, Any]: + asset = await assets.register_asset(payload) + return ok(_serialize(asset)) + + +@router.get("", summary="素材列表：entity_id / tryon_ready 过滤 + 分页") +async def list_assets( + assets: KbAssetServiceDep, + entity_id: int | None = Query(default=None, description="按实体过滤（雪花 ID）"), + tryon_ready: bool | None = Query(default=None, description="true=仅可换装素材"), + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), +) -> dict[str, Any]: + rows, total = await assets.list_assets( + entity_id=entity_id, tryon_ready=tryon_ready, limit=limit, offset=offset + ) + return ok( + { + "items": [_serialize(row) for row in rows], + "total": total, + "limit": limit, + "offset": offset, + } + ) + + +@router.post( + "/{asset_id}/review", + summary="审核素材（tryon_ready=true 仅在 review_status=approved 时被接受）", +) +async def review_asset( + asset_id: int, payload: AssetReview, assets: KbAssetServiceDep +) -> dict[str, Any]: + asset = await assets.review_asset(asset_id, payload) + return ok(_serialize(asset)) diff --git a/apps/admin/routers/gaps.py b/apps/admin/routers/gaps.py new file mode 100644 index 0000000..7083a08 --- /dev/null +++ b/apps/admin/routers/gaps.py @@ -0,0 +1,58 @@ +"""知识缺口看板路由（FR-ADM-05）：被问到却没有答案的查询，按热度排序。 + +数据来自检索无结果时的自动埋点（T-P2-06），看板只读：它回答「下一批内容 +该生产什么」，是内容排期的输入，不是待办队列。 + +偏差说明（详见提交信息）：协调提交提供的 ``packages.retrieval.list_gaps`` +对 ``select(KbGap)`` 的结果调 ``.mappings()``，行键是实体名而非列名， +``GapSummary(row)`` 取 ``row.gap_id`` 必然 ``AttributeError``（该路径无既有 +测试覆盖，真库一跑即暴露）。``packages/**`` 本任务禁改，故此处以等价查询 +在 admin 包内自实现；上游修复后应切回 retrieval 的 ``list_gaps``。 +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Query +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from apps.admin.deps import SessionDep, admin_auth +from apps.admin.schemas import GapOut +from packages.shared.models import KbGap +from packages.shared.response import ok + +router = APIRouter(prefix="/kb", tags=["admin:gaps"], dependencies=[Depends(admin_auth)]) + + +async def list_gap_rows(session: AsyncSession, *, limit: int, offset: int) -> list[KbGap]: + """kb_gap 看板查询：hit_count 降序，同频次按最近出现降序（与 retrieval 语义一致）。""" + result = await session.execute( + select(KbGap) + .order_by(KbGap.hit_count.desc(), KbGap.last_seen_at.desc()) + .limit(limit) + .offset(offset) + ) + return list(result.scalars().all()) + + +@router.get("/gaps", summary="知识缺口列表（hit_count 降序：最常被问到的排最前）") +async def list_gaps( + session: SessionDep, + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), +) -> dict[str, Any]: + rows = await list_gap_rows(session, limit=limit, offset=offset) + items = [ + GapOut( + gap_id=str(row.gap_id), + query_text=row.query_text, + filter_ctx=dict(row.filter_ctx), + hit_count=row.hit_count, + first_seen_at=row.first_seen_at, + last_seen_at=row.last_seen_at, + ).model_dump() + for row in rows + ] + return ok({"items": items, "total": len(items), "limit": limit, "offset": offset}) diff --git a/apps/admin/routers/kb.py b/apps/admin/routers/kb.py new file mode 100644 index 0000000..12e9cd4 --- /dev/null +++ b/apps/admin/routers/kb.py @@ -0,0 +1,132 @@ +"""知识实体路由（FR-ADM-01）：维护 + 状态机流转 + 图谱查看。 + +与客户端 ``apps.api`` 的 kb 路由关键差异： + +- 鉴权走 ``X-Admin-Token``（共享令牌），操作者没有用户身份，``updated_by`` + 固定记录为 ``admin-console``——V1 的操作留证靠 request_id 关联访问日志； +- 多出 ``POST /items/{id}/transition``：把状态机流转从 PATCH 的「内容+状态 + 可混用」里独立出来，发布/撤稿动作在这里挂上跨域缓存失效编排（T-P2-10 + 的「自然失效」入口）。 +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Query + +from apps.admin.deps import KbServiceDep, admin_auth +from apps.admin.publishing import CACHE_AFFECTING_TARGETS, invalidate_downstream_caches +from apps.admin.schemas import TransitionRequest +from packages.kb.schemas import ( + MAX_GRAPH_HOPS, + EntityCreate, + EntityOut, + EntityStatus, + EntityType, + EntityUpdate, +) +from packages.shared.models import KbEntity +from packages.shared.response import ok + +router = APIRouter(prefix="/kb", tags=["admin:kb"], dependencies=[Depends(admin_auth)]) + +#: V1 共享令牌鉴权没有操作者身份，字段统一记渠道来源（见模块 docstring）。 +ADMIN_OPERATOR = "admin-console" + + +def _serialize(entity: KbEntity) -> dict[str, Any]: + 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( + 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, Any]: + 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_item(payload: EntityCreate, kb: KbServiceDep) -> dict[str, Any]: + entity = await kb.create_entity(payload=payload, updated_by=ADMIN_OPERATOR) + return ok(_serialize(entity)) + + +@router.get("/items/{entity_id}", summary="实体详情") +async def get_item(entity_id: int, kb: KbServiceDep) -> dict[str, Any]: + entity = await kb.get_entity(entity_id=entity_id) + return ok(_serialize(entity)) + + +@router.patch("/items/{entity_id}", summary="修改内容（draft/reviewing；published 只读）") +async def update_item(entity_id: int, payload: EntityUpdate, kb: KbServiceDep) -> dict[str, Any]: + entity = await kb.update_entity(entity_id=entity_id, payload=payload, updated_by=ADMIN_OPERATOR) + return ok(_serialize(entity)) + + +@router.post( + "/items/{entity_id}/transition", + summary="状态机流转：draft→reviewing→published→archived（发布/撤稿触发缓存失效编排）", +) +async def transition_item( + entity_id: int, payload: TransitionRequest, kb: KbServiceDep +) -> dict[str, Any]: + # 先读当前状态：与目标相同的重放是幂等 no-op（与 service 语义一致）， + # 不触发失效编排——缓存重建是全量动作，重放不该有副作用。 + current = await kb.get_entity(entity_id=entity_id) + if current.status == payload.target: + return ok(_serialize(current)) + + entity = await kb.update_entity( + entity_id=entity_id, + payload=EntityUpdate(status=payload.target), + updated_by=ADMIN_OPERATOR, + ) + if payload.target in CACHE_AFFECTING_TARGETS: + invalidate_downstream_caches( + domain="kb_entity", key=str(entity.entity_id), target=payload.target + ) + return ok(_serialize(entity)) + + +@router.get("/items/{entity_id}/graph", summary="从该实体出发的 1~3 跳可解释路径") +async def get_item_graph( + entity_id: int, + kb: KbServiceDep, + max_hops: int = Query( + default=MAX_GRAPH_HOPS, ge=1, le=MAX_GRAPH_HOPS, description="遍历深度 1~3，超界 422" + ), +) -> dict[str, Any]: + result = await kb.traverse_graph(entity_id=entity_id, max_hops=max_hops) + return ok(result.model_dump()) diff --git a/apps/admin/routers/rules.py b/apps/admin/routers/rules.py new file mode 100644 index 0000000..de912d4 --- /dev/null +++ b/apps/admin/routers/rules.py @@ -0,0 +1,82 @@ +"""规则路由（FR-ADM-02）：决策表维护 + 发布/下线状态机。 + +规则以 ``rule_key``（命中链锚点）为地址，不是数字主键——路径参数即 rule_key。 +HTTP 载荷直接复用 ``packages.rules.dsl.Rule``（封闭词汇表在入口就拦住 typo +键名），service 层落库前还会再校验一遍。发布与下线都走缓存失效编排：规则 +既影响建议输出（规则缓存），也影响检索加权（rule_boost）。 +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Query + +from apps.admin.deps import RuleServiceDep, admin_auth +from apps.admin.publishing import invalidate_downstream_caches +from apps.admin.schemas import RuleOut, RuleTransitionRequest, RuleUpsert +from packages.shared.models import KbRule +from packages.shared.response import ok + +router = APIRouter(prefix="/rules", tags=["admin:rules"], dependencies=[Depends(admin_auth)]) + +#: V1 共享令牌鉴权没有操作者身份，author 记渠道来源。 +ADMIN_OPERATOR = "admin-console" + + +def _serialize(row: KbRule) -> dict[str, Any]: + return RuleOut( + rule_id=str(row.rule_id), + rule_key=row.rule_key, + category=row.category, + version=row.version, + status=row.status, + priority=row.priority, + conditions=dict(row.conditions), + effects=dict(row.effects), + is_hard=row.is_hard, + confidence=float(row.confidence), + evidence=row.evidence, + author=row.author, + reviewer=row.reviewer, + published_at=row.published_at, + created_at=row.created_at, + ).model_dump() + + +@router.get("", summary="规则列表（可选 status 过滤）") +async def list_rules( + rules: RuleServiceDep, + status: str | None = Query(default=None, description="draft / published / archived"), +) -> dict[str, Any]: + rows = await rules.list(status) + return ok({"items": [_serialize(row) for row in rows], "total": len(rows)}) + + +@router.post("", summary="新建规则（draft / version=1 起步，DSL 先校验后落库）") +async def create_rule(payload: RuleUpsert, rules: RuleServiceDep) -> dict[str, Any]: + row = await rules.create(payload.model_dump(), author=ADMIN_OPERATOR) + return ok(_serialize(row)) + + +@router.patch("/{rule_key}", summary="修改规则内容（仅 draft；rule_key 不可改）") +async def update_rule(rule_key: str, payload: RuleUpsert, rules: RuleServiceDep) -> dict[str, Any]: + row = await rules.update(rule_key, payload.model_dump()) + return ok(_serialize(row)) + + +@router.post( + "/{rule_key}/transition", + summary="发布 / 下线规则（均触发规则缓存与检索缓存的失效编排）", +) +async def transition_rule( + rule_key: str, payload: RuleTransitionRequest, rules: RuleServiceDep +) -> dict[str, Any]: + # service 的 publish/archive 自带非法跃迁校验（40900）与规则缓存失效； + # 检索缓存的联动在这里补上（跨域编排属应用层职责，见 publishing 模块）。 + if payload.target == "published": + row = await rules.publish(rule_key) + else: + row = await rules.archive(rule_key) + invalidate_downstream_caches(domain="kb_rule", key=rule_key, target=payload.target) + return ok(_serialize(row)) diff --git a/apps/admin/schemas.py b/apps/admin/schemas.py new file mode 100644 index 0000000..48b9995 --- /dev/null +++ b/apps/admin/schemas.py @@ -0,0 +1,78 @@ +"""admin 域请求/响应模型（FR-ADM-01~05）。 + +实体与素材的载荷复用 ``packages/kb/schemas.py``（服务层契约同源）；本模块只 +放后台特有的形状：状态机流转请求、规则对外视图、缺口看板行。雪花 ID 一律 +以字符串出现在载荷里（与 kb 契约一致：JS Number 只有 2^53 精度）。 +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal + +from pydantic import BaseModel, Field + +from packages.kb.schemas import EntityStatus +from packages.rules.dsl import Rule as RuleUpsert + +__all__ = [ + "GapOut", + "RuleOut", + "RuleTransitionRequest", + "RuleUpsert", + "TransitionRequest", +] + + +class TransitionRequest(BaseModel): + """``POST /kb/items/{id}/transition`` 载荷。 + + ``target`` 收全量四态而不是只收三个「前进方向」：published→draft 这类非法 + 跃迁应该是**状态机语义冲突（40900）**，而不是 schema 层的 422——顾问在 + 界面上看到的错误要能告诉他「为什么不行」（矩阵见 ``packages.kb.service``）， + 而不是「这个值不合法」。draft 作为目标在矩阵里永远非法，自然会被拦下。 + """ + + target: EntityStatus + + +class RuleTransitionRequest(BaseModel): + """``POST /rules/{rule_key}/transition`` 载荷。 + + 规则状态机里 draft 永远不是合法目标（发布与下线是仅有的两个前进动作， + 与实体不同，规则没有 reviewing 中间态），直接在 schema 层收窄。 + """ + + target: Literal["published", "archived"] + + +class RuleOut(BaseModel): + """规则对外视图。conditions/effects 保持入库的 DSL 原样（决策表是数据， + 后台表单按 DSL 渲染，不做二次加工）。""" + + rule_id: str + rule_key: str + category: str + version: int = Field(description="内容版本号，每次修改 +1") + status: str + priority: int + conditions: dict[str, Any] + effects: dict[str, Any] + is_hard: bool + confidence: float + evidence: str | None + author: str | None + reviewer: str | None + published_at: datetime | None + created_at: datetime + + +class GapOut(BaseModel): + """缺口看板一行（kb_gap，T-P2-06）。hit_count 是排序依据也是优先级信号。""" + + gap_id: str + query_text: str = Field(description="用户真实查询原文——内容生产的起点") + filter_ctx: dict[str, Any] = Field(description="缺口出现时的过滤条件快照") + hit_count: int = Field(description="同一缺口累计出现次数") + first_seen_at: datetime + last_seen_at: datetime diff --git a/tests/integration/test_admin_console.py b/tests/integration/test_admin_console.py new file mode 100644 index 0000000..084cce7 --- /dev/null +++ b/tests/integration/test_admin_console.py @@ -0,0 +1,334 @@ +"""极简运营后台集成用例（真实 PostgreSQL，T-P2-09 / FR-ADM-01~05）。 + +核心断言是「顾问全流程不碰客户端 API」：建实体 → 评审 → 发布 → 建规则 → +发布，全部只走 ``/admin/api/v1``；发布编排的副作用（规则缓存失效 + +检索缓存失效版本号递增）直接可观察。其余覆盖：非法跃迁 40900、素材登记 +审核约束、缺口看板排序、令牌校验。 +""" + +from __future__ import annotations + +import os +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +import pytest + +ADMIN_TOKEN = "it-admin-token-t12" +AUTH = {"X-Admin-Token": ADMIN_TOKEN} + + +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]: + """用例结束即释放全局引擎（理由同 test_kb_entity.py：每用例独立事件循环）。""" + yield + from packages.shared.db import dispose_engine + + await dispose_engine() + + +@pytest.fixture +async def client() -> Any: + import httpx + + from apps.admin.app import create_admin_app + from packages.shared.config import Settings + + # 显式给非空令牌（.env 默认为空 = 未部署语义，那一条在单测覆盖） + app = create_admin_app(Settings(admin_token=ADMIN_TOKEN)) + transport = httpx.ASGITransport(app=app) + async with ( + httpx.AsyncClient(transport=transport, base_url="http://test") as http, + app.router.lifespan_context(app), + ): + yield http + + +@asynccontextmanager +async def _db() -> AsyncIterator[Any]: + """直连 session（只读断言与 kb_gap 预插数据），复用全局懒加载工厂。""" + from packages.shared.db import get_session_factory + + async with get_session_factory()() as session: + yield session + + +def _rule_payload(rule_key: str) -> dict[str, Any]: + return { + "rule_key": rule_key, + "category": "body_shape", + "priority": 90, + "confidence": 0.9, + "conditions": {"body_shape": "A型"}, + "effects": { + "recommend": [{"slot": "top", "attrs": {"shoulder": ["T12垫肩"]}}], + "explanation": "T12 后台发布编排集成规则", + }, + } + + +async def _create_published_entity(client: Any, suffix: str) -> str: + """经后台 HTTP 走完整评审线：draft → reviewing → published。""" + created = await client.post( + "/admin/api/v1/kb/items", + json={"entity_type": "item", "name": f"风衣-{suffix}", "attrs": {"length": "long"}}, + headers=AUTH, + ) + assert created.status_code == 200 + entity_id = created.json()["data"]["entity_id"] + reviewing = await client.post( + f"/admin/api/v1/kb/items/{entity_id}/transition", json={"target": "reviewing"}, headers=AUTH + ) + assert reviewing.status_code == 200 + published = await client.post( + f"/admin/api/v1/kb/items/{entity_id}/transition", json={"target": "published"}, headers=AUTH + ) + assert published.status_code == 200 + return entity_id + + +async def test_consultant_full_flow_publishes_without_engineering(client: Any) -> None: + """顾问全流程只打后台 API；发布编排的两个失效副作用都可观察。""" + from packages.retrieval import repo as retrieval_repo + from packages.rules import service as rules_service + from packages.shared.models import KbEntity + + suffix = uuid.uuid4().hex[:8] + rule_key = f"T12-IT-{suffix}" + + # -- 建实体（draft 起步，操作者记为 admin-console）----------------------- + created = await client.post( + "/admin/api/v1/kb/items", + json={ + "entity_type": "item", + "name": f"西装外套-{suffix}", + "attrs": {"lapel": "peak"}, + "source": "搭配手册", + }, + headers=AUTH, + ) + assert created.status_code == 200 + entity = created.json()["data"] + entity_id = entity["entity_id"] + assert entity["status"] == "draft" + assert entity["version"] == 1 + assert entity["updated_by"] == "admin-console" + + # -- reviewing：不是知识可见性变更，不触发失效编排 ----------------------- + v0 = retrieval_repo.INVALIDATION_VERSION + reviewing = await client.post( + f"/admin/api/v1/kb/items/{entity_id}/transition", + json={"target": "reviewing"}, + headers=AUTH, + ) + assert reviewing.status_code == 200 + assert reviewing.json()["data"]["status"] == "reviewing" + assert v0 == retrieval_repo.INVALIDATION_VERSION + + # -- published：双失效（规则缓存 + 检索缓存）的副作用可观察 -------------- + v1 = retrieval_repo.INVALIDATION_VERSION + published = await client.post( + f"/admin/api/v1/kb/items/{entity_id}/transition", + json={"target": "published"}, + headers=AUTH, + ) + assert published.status_code == 200 + assert published.json()["data"]["status"] == "published" + assert v1 + 1 == retrieval_repo.INVALIDATION_VERSION + + # 「检索新内容可见」的简化断言：直接再查库 + async with _db() as session: + row = await session.get(KbEntity, int(entity_id)) + assert row is not None + assert row.status == "published" + + # -- 建规则（draft）→ 发布 → PublishedRules 缓存能载入 ------------------- + rule_created = await client.post( + "/admin/api/v1/rules", json=_rule_payload(rule_key), headers=AUTH + ) + assert rule_created.status_code == 200 + rule_body = rule_created.json()["data"] + assert rule_body["status"] == "draft" + assert rule_body["author"] == "admin-console" + + # 预热进程内缓存：发布前不可见 + async with _db() as session: + pre = await rules_service._published_rules.load(session) + assert rule_key not in {r.rule_key for r in pre} + + v2 = retrieval_repo.INVALIDATION_VERSION + rule_published = await client.post( + f"/admin/api/v1/rules/{rule_key}/transition", + json={"target": "published"}, + headers=AUTH, + ) + assert rule_published.status_code == 200 + assert rule_published.json()["data"]["published_at"] is not None + assert v2 + 1 == retrieval_repo.INVALIDATION_VERSION + + # 发布 + 失效后重载：新规则可被 PublishedRules 载入 + async with _db() as session: + post = await rules_service._published_rules.load(session) + assert rule_key in {r.rule_key for r in post} + + # -- 撤稿规则（下线同样触发失效编排），给环境留干净状态 ------------------- + v3 = retrieval_repo.INVALIDATION_VERSION + archived = await client.post( + f"/admin/api/v1/rules/{rule_key}/transition", + json={"target": "archived"}, + headers=AUTH, + ) + assert archived.status_code == 200 + assert archived.json()["data"]["status"] == "archived" + assert v3 + 1 == retrieval_repo.INVALIDATION_VERSION + async with _db() as session: + after = await rules_service._published_rules.load(session) + assert rule_key not in {r.rule_key for r in after} + + +async def test_illegal_transitions_surface_as_40900(client: Any) -> None: + suffix = uuid.uuid4().hex[:8] + entity_id = await _create_published_entity(client, suffix) + + # published → draft：状态机语义冲突 40900（而不是 schema 422） + backward = await client.post( + f"/admin/api/v1/kb/items/{entity_id}/transition", + json={"target": "draft"}, + headers=AUTH, + ) + assert backward.status_code == 409 + assert backward.json()["code"] == 40900 + + # 规则侧：published 再发布（自跃迁）同样 40900 + rule_key = f"T12-SM-{suffix}" + created = await client.post("/admin/api/v1/rules", json=_rule_payload(rule_key), headers=AUTH) + assert created.status_code == 200 + published = await client.post( + f"/admin/api/v1/rules/{rule_key}/transition", + json={"target": "published"}, + headers=AUTH, + ) + assert published.status_code == 200 + republish = await client.post( + f"/admin/api/v1/rules/{rule_key}/transition", + json={"target": "published"}, + headers=AUTH, + ) + assert republish.status_code == 409 + assert republish.json()["code"] == 40900 + + # 收尾下线，不留 published 规则影响其他用例 + cleanup = await client.post( + f"/admin/api/v1/rules/{rule_key}/transition", + json={"target": "archived"}, + headers=AUTH, + ) + assert cleanup.status_code == 200 + + +async def test_asset_register_review_and_tryon_ready_gate(client: Any) -> None: + suffix = uuid.uuid4().hex[:8] + entity_id = await _create_published_entity(client, suffix) + + # 登记时即使请求体给 tryon_ready=true，服务端也强制 False（审核是唯一开关） + registered = await client.post( + "/admin/api/v1/assets", + json={ + "entity_id": entity_id, + "asset_type": "display", + "cos_key": f"kb/it/t12/{suffix}/coat.jpg", + "spec_check": {"bg": "white", "size": "1024x1536"}, + "tryon_ready": True, + "license": "CC-BY-4.0", + }, + headers=AUTH, + ) + assert registered.status_code == 200 + asset = registered.json()["data"] + asset_id = asset["asset_id"] + assert asset["tryon_ready"] is False + assert asset["review_status"] == "pending" + + # 审核不通过却想打开 tryon_ready → 服务层约束透传成 400 + rejected = await client.post( + f"/admin/api/v1/assets/{asset_id}/review", + json={"review_status": "rejected", "tryon_ready": True}, + headers=AUTH, + ) + assert rejected.status_code == 400 + assert rejected.json()["code"] == 40000 + + # 审核通过 + tryon_ready 一起打开 + approved = await client.post( + f"/admin/api/v1/assets/{asset_id}/review", + json={"review_status": "approved", "tryon_ready": True}, + headers=AUTH, + ) + assert approved.status_code == 200 + assert approved.json()["data"]["tryon_ready"] is True + assert approved.json()["data"]["review_status"] == "approved" + + # 列表过滤：tryon_ready=true 命中，false 视图为空 + ready = await client.get( + f"/admin/api/v1/assets?entity_id={entity_id}&tryon_ready=true", headers=AUTH + ) + assert ready.status_code == 200 + ready_items = ready.json()["data"]["items"] + assert [item["asset_id"] for item in ready_items] == [asset_id] + not_ready = await client.get( + f"/admin/api/v1/assets?entity_id={entity_id}&tryon_ready=false", headers=AUTH + ) + assert not_ready.json()["data"]["total"] == 0 + + +async def test_gap_dashboard_orders_by_hit_count_desc(client: Any) -> None: + from packages.domain.ids import next_id + from packages.shared.models import KbGap + + suffix = uuid.uuid4().hex[:8] + low, high = f"T12-缺口低频-{suffix}", f"T12-缺口高频-{suffix}" + async with _db() as session: + session.add(KbGap(gap_id=next_id(), query_text=low, filter_ctx={}, hit_count=2)) + session.add( + KbGap( + gap_id=next_id(), + query_text=high, + filter_ctx={"body_shape": "A型"}, + hit_count=51, + ) + ) + await session.commit() + + resp = await client.get("/admin/api/v1/kb/gaps?limit=200", headers=AUTH) + assert resp.status_code == 200 + items = resp.json()["data"]["items"] + texts = [item["query_text"] for item in items] + assert low in texts and high in texts + # hit_count 降序：高频缺口的位次必须靠前（历史数据可以穿插在中间） + assert texts.index(high) < texts.index(low) + high_row = next(item for item in items if item["query_text"] == high) + assert high_row["hit_count"] == 51 + assert high_row["filter_ctx"] == {"body_shape": "A型"} + assert high_row["gap_id"].isdigit() # 雪花 ID 以字符串出现 + + +async def test_missing_or_wrong_token_is_401(client: Any) -> None: + for headers in (None, {"X-Admin-Token": "wrong"}): + resp = await client.get("/admin/api/v1/rules", headers=headers) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 diff --git a/tests/unit/test_admin_app.py b/tests/unit/test_admin_app.py new file mode 100644 index 0000000..abd3a27 --- /dev/null +++ b/tests/unit/test_admin_app.py @@ -0,0 +1,393 @@ +"""admin 应用装配层单测（不连库，FR-ADM 鉴权语义，T-P2-09）。 + +覆盖三件事： + +1. ``ADMIN_TOKEN=""`` 时全部业务路由不注册（404 = 等价未部署）； +2. 令牌缺失 / 错误 / 非 ASCII → 40100（比较必须走 bytes：str 版 + ``compare_digest`` 遇非 ASCII 会 TypeError 变 500）； +3. 正确令牌 → 200：服务依赖用假实现替换（单测不碰 DB），顺带验证 + 发布编排（publish → 规则缓存 + 检索缓存双失效）在路由层的接线。 +真实的 40900 / 缓存副作用语义由 ``tests/integration/test_admin_console.py`` +用真库验证。 +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from datetime import UTC, datetime +from types import SimpleNamespace + +import httpx +import pytest +from fastapi import FastAPI + +from apps.admin.app import ADMIN_API_PREFIX, create_admin_app +from apps.admin.deps import ( + kb_asset_service_dep, + kb_service_dep, + rule_service_dep, + session_dep, +) +from apps.admin.publishing import invalidate_downstream_caches +from packages.kb.schemas import EntityUpdate +from packages.shared.config import Settings + +TOKEN = "unit-admin-token" +P = ADMIN_API_PREFIX + +#: 全部业务路由的代表性样本——空令牌时一个都不该存在。 +_ALL_ROUTES: tuple[tuple[str, str], ...] = ( + ("GET", f"{P}/kb/items"), + ("POST", f"{P}/kb/items"), + ("GET", f"{P}/kb/items/1"), + ("PATCH", f"{P}/kb/items/1"), + ("POST", f"{P}/kb/items/1/transition"), + ("GET", f"{P}/kb/items/1/graph"), + ("GET", f"{P}/kb/gaps"), + ("GET", f"{P}/rules"), + ("POST", f"{P}/rules"), + ("PATCH", f"{P}/rules/demo-rule"), + ("POST", f"{P}/rules/demo-rule/transition"), + ("GET", f"{P}/assets"), + ("POST", f"{P}/assets"), + ("POST", f"{P}/assets/1/review"), +) + + +def _build_app(token: str) -> FastAPI: + return create_admin_app(Settings(admin_token=token)) + + +@asynccontextmanager +async def _client(app: FastAPI) -> AsyncIterator[httpx.AsyncClient]: + """不进 lifespan：单测不建 DB engine，只测装配与鉴权层。""" + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as http: + yield http + + +# --- 假实现：够走通序列化即可，不模拟 ORM 行为 -------------------------------- + + +def _fake_entity(status: str) -> SimpleNamespace: + now = datetime.now(UTC) + return SimpleNamespace( + entity_id=123456789012345678, + entity_type="item", + name="白衬衫", + attrs={"sleeve": "long"}, + status=status, + version=1, + source=None, + license=None, + updated_by=None, + created_at=now, + updated_at=now, + ) + + +def _fake_rule(status: str) -> SimpleNamespace: + now = datetime.now(UTC) + return SimpleNamespace( + rule_id=987654321098765432, + rule_key="demo-rule", + category="body_shape", + version=1, + status=status, + priority=90, + conditions={"body_shape": "A型"}, + effects={"avoid": [{"slot": "top", "attrs": {"shoulder": ["落肩"]}}]}, + is_hard=False, + confidence=0.9, + evidence=None, + author=None, + reviewer=None, + published_at=None, + created_at=now, + ) + + +class _ListKbService: + async def search_entities(self, **_: object) -> tuple[list[object], int]: + return [], 0 + + +class _ListRuleService: + async def list(self, status: str | None = None) -> list[object]: + return [] + + +class _ListAssetService: + async def list_assets(self, **_: object) -> tuple[list[object], int]: + return [], 0 + + +class _TransitionKbService: + """记录 update 调用的假实体服务（transition 路由专用）。""" + + def __init__(self, entity: SimpleNamespace) -> None: + self._entity = entity + self.update_calls: list[dict[str, object]] = [] + + async def get_entity(self, *, entity_id: int) -> SimpleNamespace: + return self._entity + + async def update_entity( + self, + *, + entity_id: int, + payload: EntityUpdate, + updated_by: str | None = None, + ) -> SimpleNamespace: + self.update_calls.append({"entity_id": entity_id, "target": payload.status}) + self._entity.status = payload.status + return self._entity + + +class _TransitionRuleService: + def __init__(self, row: SimpleNamespace) -> None: + self._row = row + self.calls: list[str] = [] + + async def publish(self, rule_key: str) -> SimpleNamespace: + self.calls.append(f"publish:{rule_key}") + self._row.status = "published" + return self._row + + async def archive(self, rule_key: str) -> SimpleNamespace: + self.calls.append(f"archive:{rule_key}") + self._row.status = "archived" + return self._row + + +class _FakeScalars: + def all(self) -> list[object]: + return [] + + +class _FakeExecResult: + def scalars(self) -> _FakeScalars: + return _FakeScalars() + + +class _FakeSession: + async def execute(self, stmt: object) -> _FakeExecResult: + return _FakeExecResult() + + +async def _fake_session_dep() -> AsyncIterator[_FakeSession]: + yield _FakeSession() + + +# --- 1. 空令牌 = 未部署 -------------------------------------------------------- + + +async def test_empty_token_registers_no_routes_at_all() -> None: + app = create_admin_app(Settings(admin_token="")) + async with _client(app) as client: + for method, path in _ALL_ROUTES: + resp = await client.request( + method, path, json={}, headers={"X-Admin-Token": "anything"} + ) + assert resp.status_code == 404, (method, path) + # 404 也保持统一信封（经 StarletteHTTPException 处理器转 40400） + assert resp.json()["code"] == 40400 + # openapi / docs 一并关闭，扫描器探不到任何面 + assert (await client.get(f"{P}/openapi.json")).status_code == 404 + assert (await client.get(f"{P}/docs")).status_code == 404 + + +# --- 2. 令牌校验 -------------------------------------------------------------- + + +async def test_missing_or_wrong_token_is_40100() -> None: + app = _build_app(TOKEN) + async with _client(app) as client: + cases: list[dict[str, str] | None] = [ + None, # 无头 + {"X-Admin-Token": "wrong-token"}, + {"X-Admin-Token": ""}, # 空头与缺失同罪 + {"X-Admin-Token": TOKEN + "x"}, # 前缀对也不行（compare_digest 全长比较） + ] + for headers in cases: + resp = await client.get(f"{P}/rules", headers=headers) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + +async def test_non_ascii_token_does_not_blow_up_into_500() -> None: + """str 版 compare_digest 遇非 ASCII 会 TypeError；实现必须走 bytes 比较。 + + httpx 不许 str 头带非 ASCII，但线上恶意客户端可以发任意字节——这里用 + bytes 头值模拟（starlette 会以 latin-1 解出非 ASCII str）。 + """ + app = _build_app(TOKEN) + async with _client(app) as client: + resp = await client.get(f"{P}/rules", headers={"X-Admin-Token": "非ASCII令牌".encode()}) + assert resp.status_code == 401 + assert resp.json()["code"] == 40100 + + +# --- 3. 正确令牌 → 200（假服务） ---------------------------------------------- + + +async def test_valid_token_reaches_all_read_handlers() -> None: + app = _build_app(TOKEN) + app.dependency_overrides[kb_service_dep] = lambda: _ListKbService() + app.dependency_overrides[rule_service_dep] = lambda: _ListRuleService() + app.dependency_overrides[kb_asset_service_dep] = lambda: _ListAssetService() + app.dependency_overrides[session_dep] = _fake_session_dep + headers = {"X-Admin-Token": TOKEN} + async with _client(app) as client: + for path in (f"{P}/kb/items", f"{P}/rules", f"{P}/assets", f"{P}/kb/gaps"): + resp = await client.get(path, headers=headers) + assert resp.status_code == 200, path + body = resp.json() + assert body["code"] == 0 + assert body["data"]["items"] == [] + assert body["request_id"] # 中间件注入，V1 操作留证的依托 + + +# --- 4. 发布编排接线 ---------------------------------------------------------- + + +def test_invalidate_downstream_caches_calls_both_domains( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from apps.admin import publishing + + calls: list[str] = [] + monkeypatch.setattr( + publishing.rules_service, "invalidate_published_cache", lambda: calls.append("rules") + ) + monkeypatch.setattr( + publishing.retrieval_service, "invalidate_cache", lambda: calls.append("retrieval") + ) + invalidate_downstream_caches(domain="kb_entity", key="1", target="published") + assert calls == ["rules", "retrieval"] + + +async def test_entity_publish_transition_triggers_double_invalidation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + app = _build_app(TOKEN) + svc = _TransitionKbService(_fake_entity(status="reviewing")) + app.dependency_overrides[kb_service_dep] = lambda: svc + triggered: list[dict[str, str]] = [] + monkeypatch.setattr( + "apps.admin.routers.kb.invalidate_downstream_caches", + lambda **kw: triggered.append(kw), + ) + async with _client(app) as client: + resp = await client.post( + f"{P}/kb/items/1/transition", + json={"target": "published"}, + headers={"X-Admin-Token": TOKEN}, + ) + assert resp.status_code == 200 + assert resp.json()["data"]["status"] == "published" + assert svc.update_calls == [{"entity_id": 1, "target": "published"}] + assert triggered == [ + {"domain": "kb_entity", "key": "123456789012345678", "target": "published"} + ] + + +async def test_entity_transition_replay_is_noop_without_invalidation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """与当前状态相同的重放幂等成功，但不触发缓存重建（重试不该有副作用）。""" + app = _build_app(TOKEN) + svc = _TransitionKbService(_fake_entity(status="published")) + app.dependency_overrides[kb_service_dep] = lambda: svc + triggered: list[dict[str, str]] = [] + monkeypatch.setattr( + "apps.admin.routers.kb.invalidate_downstream_caches", + lambda **kw: triggered.append(kw), + ) + async with _client(app) as client: + resp = await client.post( + f"{P}/kb/items/1/transition", + json={"target": "published"}, + headers={"X-Admin-Token": TOKEN}, + ) + assert resp.status_code == 200 + assert resp.json()["data"]["status"] == "published" + assert svc.update_calls == [] + assert triggered == [] + + +async def test_entity_transition_accepts_draft_target_at_schema_level( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """target 收全量四态：draft 能通过 body 校验到达处理器（证明不是 422 拦截）。 + + 这里用「draft→draft 的幂等重放分支」验证到达；真实的 published→draft + 40900 语义由集成测试对真状态机验证。 + """ + app = _build_app(TOKEN) + svc = _TransitionKbService(_fake_entity(status="draft")) + app.dependency_overrides[kb_service_dep] = lambda: svc + triggered: list[dict[str, str]] = [] + monkeypatch.setattr( + "apps.admin.routers.kb.invalidate_downstream_caches", + lambda **kw: triggered.append(kw), + ) + async with _client(app) as client: + resp = await client.post( + f"{P}/kb/items/1/transition", + json={"target": "draft"}, + headers={"X-Admin-Token": TOKEN}, + ) + assert resp.status_code == 200 + assert resp.json()["data"]["status"] == "draft" + assert svc.update_calls == [] + + +async def test_rule_transition_publish_and_archive_both_invalidate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + app = _build_app(TOKEN) + svc = _TransitionRuleService(_fake_rule(status="draft")) + app.dependency_overrides[rule_service_dep] = lambda: svc + triggered: list[dict[str, str]] = [] + monkeypatch.setattr( + "apps.admin.routers.rules.invalidate_downstream_caches", + lambda **kw: triggered.append(kw), + ) + headers = {"X-Admin-Token": TOKEN} + async with _client(app) as client: + published = await client.post( + f"{P}/rules/demo-rule/transition", json={"target": "published"}, headers=headers + ) + assert published.status_code == 200 + assert published.json()["data"]["status"] == "published" + + archived = await client.post( + f"{P}/rules/demo-rule/transition", json={"target": "archived"}, headers=headers + ) + assert archived.status_code == 200 + assert archived.json()["data"]["status"] == "archived" + + # 规则状态机没有「回 draft」动作，schema 层直接收窄成 422 + bad = await client.post( + f"{P}/rules/demo-rule/transition", json={"target": "draft"}, headers=headers + ) + assert bad.status_code == 422 + assert bad.json()["code"] == 40001 + + assert svc.calls == ["publish:demo-rule", "archive:demo-rule"] + assert [t["target"] for t in triggered] == ["published", "archived"] + + +async def test_entity_transition_rejects_unknown_target() -> None: + """未知目标值（不在四态里）在 schema 层拦成 422。""" + app = _build_app(TOKEN) + async with _client(app) as client: + resp = await client.post( + f"{P}/kb/items/1/transition", + json={"target": "frozen"}, + headers={"X-Admin-Token": TOKEN}, + ) + assert resp.status_code == 422 + assert resp.json()["code"] == 40001