dadao
e9addeed25
- knowledge_doc 与 item 1:1，除 source_kind=manual 外必须有来源（库级约束） - 状态机 draft/in_review/published/archived，发布要 owner/admin；非法动作回 allowed_from 与 allowed_actions - 工单验收通过自动落 postmortem 草稿（generated_by=rule），同一来源只生成一次 - 接口：列表/详情/手记/改内容/状态动作/从事项沉淀；已发布禁改，改内容带乐观锁 - Web 知识页 + 详情抽屉；任务与工单详情里可「沉淀成知识」 - 8 个后端用例 + e2e 7 项断言（沉淀 → 待核对 → 发布）
343 lines
12 KiB
Python
343 lines
12 KiB
Python
"""M3 知识域测试：草稿、来源、审议发布、隔离。
|
||
|
||
重点不是"接口能通"，而是这几条容易做错的：
|
||
|
||
1. **知识的来源不可丢**：从工单沉淀出来的知识必须指回那张工单（`source_kind='ticket'`），
|
||
否则半年后没人能核对这条结论从哪来；
|
||
2. **自动沉淀只落草稿**：验收通过后知识是 `draft`，不是 `published`；
|
||
3. **同一件事只沉淀一次**：重复点"沉淀成知识"要报 7002，而不是悄悄生成两份；
|
||
4. **发布要管理员**：普通成员的发布请求必须是 403（7004）；
|
||
5. **跨租户不可见**：别人的知识列表为空、直接按 id 取是 7001。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from httpx import AsyncClient
|
||
|
||
from .conftest import create_workspace, login, unwrap
|
||
|
||
V1 = "/api/v1"
|
||
|
||
|
||
async def _member(
|
||
client: AsyncClient, owner: dict[str, Any], ws: dict[str, Any], role: str = "member"
|
||
) -> dict[str, Any]:
|
||
user = await login(client)
|
||
inv = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/invitations", json={"role": role}, headers=owner["headers"]
|
||
)
|
||
)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/join", json={"code": inv["code"]}, headers=user["headers"]
|
||
)
|
||
)
|
||
return user
|
||
|
||
|
||
async def _task(client: AsyncClient, user: dict[str, Any], ws_id: str, title: str) -> dict:
|
||
return unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items",
|
||
json={"title": title, "status": "todo"},
|
||
headers=user["headers"],
|
||
)
|
||
)
|
||
|
||
|
||
async def _finish_ticket(
|
||
client: AsyncClient, owner: dict[str, Any], ws_id: str, helper: dict[str, Any], task: dict
|
||
) -> dict:
|
||
"""把任务升级成工单并走完到验收通过，返回工单。"""
|
||
ticket = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)["ticket"]
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items/{ticket['id']}/ticket/accept", headers=helper["headers"]
|
||
)
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items/{ticket['id']}/ticket/start", headers=helper["headers"]
|
||
)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items/{ticket['id']}/ticket/submit",
|
||
json={"note": "已经开了只读账号"},
|
||
headers=helper["headers"],
|
||
)
|
||
)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_id}/items/{ticket['id']}/ticket/verify",
|
||
json={"approve": True, "resolution_md": "账号权限已开通，走的是 A 方案"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
return ticket
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_create_manual_knowledge_is_draft(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "知识手记")
|
||
|
||
data = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
json={
|
||
"title": "报销单要贴发票原件",
|
||
"body_md": "财务只认纸质发票，扫描件不行。",
|
||
"doc_type": "note",
|
||
"tag_names": ["财务"],
|
||
},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert data["kind"] == "knowledge"
|
||
assert data["status"] == "draft"
|
||
assert data["knowledge"]["source_kind"] == "manual"
|
||
assert data["knowledge"]["source_item_id"] is None
|
||
assert [t["name"] for t in data["tags"]] == ["财务"]
|
||
|
||
listed = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/knowledge", headers=owner["headers"])
|
||
)
|
||
assert [i["id"] for i in listed["items"]] == [data["id"]]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_approval_creates_draft_with_source(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "闭环沉淀")
|
||
helper = await _member(client, owner, ws)
|
||
task = await _task(client, owner, ws["id"], "开通测试账号权限")
|
||
ticket = await _finish_ticket(client, owner, ws["id"], helper, task)
|
||
|
||
listed = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
params={"status": "draft"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert listed["has_more"] is False
|
||
docs = [i for i in listed["items"] if i["knowledge"]["source_item_id"] == ticket["id"]]
|
||
assert len(docs) == 1, "验收通过应当恰好沉淀一份草稿"
|
||
doc = docs[0]
|
||
assert doc["status"] == "draft", "自动沉淀只能是草稿，不能直接发布"
|
||
assert doc["knowledge"]["source_kind"] == "ticket"
|
||
assert doc["knowledge"]["generated_by"] == "rule"
|
||
# 草稿正文必须带结论和"请先核对"的提示，否则用户会把它当已确认结论转发
|
||
assert "账号权限已开通" in doc["body_md"]
|
||
assert "请先核对再发布" in doc["body_md"]
|
||
assert doc["source"]["kind"] == "ticket"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_promote_twice_conflicts(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "重复沉淀")
|
||
task = await _task(client, owner, ws["id"], "写周报")
|
||
|
||
first = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/knowledge",
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert first["knowledge"]["source_kind"] == "task"
|
||
assert first["knowledge"]["generated_by"] == "human"
|
||
|
||
resp = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/knowledge", headers=owner["headers"]
|
||
)
|
||
assert resp.json()["code"] == 7002
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_publish_requires_admin(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "发布权限")
|
||
member = await _member(client, owner, ws)
|
||
doc = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
json={"title": "值班手册", "body_md": "先重启再上报。"},
|
||
headers=member["headers"],
|
||
)
|
||
)
|
||
|
||
reviewed = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "submit_review", "version": doc["version"]},
|
||
headers=member["headers"],
|
||
)
|
||
)
|
||
assert reviewed["status"] == "in_review"
|
||
|
||
denied = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "publish", "version": reviewed["version"]},
|
||
headers=member["headers"],
|
||
)
|
||
assert denied.json()["code"] == 7004
|
||
|
||
published = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "publish", "version": reviewed["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert published["status"] == "published"
|
||
assert published["knowledge"]["published_at"] is not None
|
||
|
||
# 已发布的内容不许直接改，先撤回
|
||
blocked = await client.patch(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}",
|
||
json={"title": "值班手册 v2", "version": published["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
assert blocked.json()["code"] == 7003
|
||
|
||
archived = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "archive", "version": published["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert archived["status"] == "archived"
|
||
assert archived["knowledge"]["archived_at"] is not None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_illegal_action_reports_allowed(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "非法动作")
|
||
doc = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
json={"title": "草稿"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "archive", "version": doc["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
resp = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}/status",
|
||
json={"action": "publish", "version": doc["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
body = resp.json()
|
||
assert body["code"] == 7003
|
||
assert body["data"]["allowed_from"] == ["draft", "in_review"]
|
||
assert "restore" in body["data"]["allowed_actions"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_knowledge_is_tenant_scoped(client: AsyncClient) -> None:
|
||
alice = await login(client)
|
||
ws_a = await create_workspace(client, alice, "A 的知识")
|
||
doc = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_a['id']}/knowledge",
|
||
json={"title": "A 的内部结论"},
|
||
headers=alice["headers"],
|
||
)
|
||
)
|
||
|
||
bob = await login(client)
|
||
ws_b = await create_workspace(client, bob, "B 的工作区")
|
||
|
||
listed = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws_b['id']}/knowledge", headers=bob["headers"])
|
||
)
|
||
assert listed["items"] == []
|
||
|
||
cross = await client.get(
|
||
f"{V1}/workspaces/{ws_b['id']}/knowledge/{doc['id']}", headers=bob["headers"]
|
||
)
|
||
assert cross.json()["code"] == 7001
|
||
|
||
not_member = await client.get(f"{V1}/workspaces/{ws_a['id']}/knowledge", headers=bob["headers"])
|
||
assert not_member.json()["code"] == 3003
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_search_and_filters(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "检索")
|
||
for title, doc_type in (
|
||
("周报模板", "howto"),
|
||
("客户投诉复盘", "postmortem"),
|
||
("周报提醒", "note"),
|
||
):
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
json={"title": title, "body_md": f"{title} 的正文", "doc_type": doc_type},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
hit = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
params={"q": "周报"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert sorted(i["title"] for i in hit["items"]) == ["周报提醒", "周报模板"]
|
||
|
||
by_type = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
params={"doc_type": "postmortem"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert [i["title"] for i in by_type["items"]] == ["客户投诉复盘"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_patch_uses_optimistic_lock(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "并发改")
|
||
doc = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge",
|
||
json={"title": "初稿", "version": 0},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
updated = unwrap(
|
||
await client.patch(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}",
|
||
json={"body_md": "补一段", "version": doc["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert updated["version"] == doc["version"] + 1
|
||
|
||
stale = await client.patch(
|
||
f"{V1}/workspaces/{ws['id']}/knowledge/{doc['id']}",
|
||
json={"body_md": "又一段", "version": doc["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
assert stale.json()["code"] == 4009
|
||
assert stale.json()["data"]["current_version"] == updated["version"]
|