DadaoTask/backend/tests/test_knowledge.py

"""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"]


@pytest.mark.asyncio
async def test_draft_timeline_reads_like_a_story(client: AsyncClient) -> None:
    """过程段要写「待办 → 已完成」，只写"状态变更"等于没写。

    这条用例是有来历的：时间线最初只查了 action，没带 payload，
    结果生成的草稿里全是"状态变更"，看不出事情是怎么走的。
    """
    owner = await login(client)
    ws = await create_workspace(client, owner, "时间线")
    task = await _task(client, owner, ws["id"], "写清楚每一步")
    for status in ("todo", "doing", "done"):
        task = unwrap(
            await client.post(
                f"{V1}/workspaces/{ws['id']}/items/{task['id']}/status",
                json={"status": status, "version": task["version"]},
                headers=owner["headers"],
            )
        )
    doc = unwrap(
        await client.post(
            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/knowledge", headers=owner["headers"]
        )
    )
    body = doc["body_md"]
    assert "待办 → 进行中" in body
    assert "进行中 → 已完成" in body