dadao
30caaaf1d9
- item.kind='ticket' + ticket_ext 1:1，迁移 0003 同时重写 item 状态约束为任务∪工单并集 - 工单状态机与动作权限：受理/开工/提交限办理人，验收限需求人（办理人不能自验） - 升级时原任务转 blocked，验收通过后自动回流 todo，事件流两侧都留痕 - item_link（父子/相关）与 notification（只发当事人），三张表均开 RLS - 抽出 serializers.py 复用任务序列化，修复工单响应缺 tags 导致前端白屏 - 评论与 PATCH 放开给工单；conftest 清库前终止残留事务，避免 TRUNCATE 死等锁 - tests/test_tickets.py 11 个用例，全量 57 passed
494 lines
17 KiB
Python
494 lines
17 KiB
Python
"""M2 工单域测试：升级、指派、办理、验收、回流、权限、通知。
|
||
|
||
重点不是"能跑通"，而是这几条容易做错的：
|
||
|
||
1. 升级后**原任务必须变 blocked**（否则用户以为这事还在自己手上）；
|
||
2. 验收通过后**原任务自动回到 todo**（闭环回流，这是整个产品的卖点）；
|
||
3. 驳回**回 in_progress 而不是 open**，且 `reject_count` 加一；
|
||
4. **办理人不能自己验收**（否则工单退化成打勾）；
|
||
5. 通知发给对的人：提交 → 提需求的人；通过/驳回 → 办理人。
|
||
"""
|
||
|
||
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"],
|
||
)
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_promote_creates_ticket_and_blocks_task(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"])
|
||
|
||
data = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
# 原任务：还是任务，但被卡住了
|
||
assert data["task"]["id"] == task["id"]
|
||
assert data["task"]["status"] == "blocked"
|
||
assert data["task"]["kind"] == "task"
|
||
|
||
# 工单：新的一条 item，指回原任务
|
||
ticket = data["ticket"]
|
||
assert ticket["id"] != task["id"]
|
||
assert ticket["kind"] == "ticket"
|
||
assert ticket["status"] == "open"
|
||
assert ticket["ticket"]["requester_id"] == owner["user_id"]
|
||
assert ticket["ticket"]["assignee_id"] == helper["user_id"]
|
||
assert ticket["ticket"]["parent_item_id"] == task["id"]
|
||
|
||
# 工单不出现在任务列表里（两条流各看各的）
|
||
tasks = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/items?view=all", headers=owner["headers"])
|
||
)
|
||
assert [t["id"] for t in tasks["items"]] == [task["id"]]
|
||
|
||
# 被指派人收到通知
|
||
notes = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
|
||
)
|
||
assert [n["kind"] for n in notes["notifications"]] == ["ticket_assigned"]
|
||
assert notes["notifications"][0]["item_id"] == ticket["id"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_full_flow_reopens_task_on_approval(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 = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)["ticket"]
|
||
|
||
base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
|
||
assert (
|
||
unwrap(await client.post(f"{base}/accept", headers=helper["headers"]))["status"]
|
||
== "accepted"
|
||
)
|
||
assert (
|
||
unwrap(await client.post(f"{base}/start", headers=helper["headers"]))["status"]
|
||
== "in_progress"
|
||
)
|
||
submitted = unwrap(
|
||
await client.post(f"{base}/submit", json={"note": "权限已开通"}, headers=helper["headers"])
|
||
)
|
||
assert submitted["status"] == "pending_verify"
|
||
|
||
# 提需求的人收到验收通知
|
||
notes = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=owner["headers"])
|
||
)
|
||
assert notes["notifications"][0]["kind"] == "ticket_submitted"
|
||
|
||
done = unwrap(
|
||
await client.post(
|
||
f"{base}/verify",
|
||
json={"approve": True, "resolution_md": "权限已开通，验证通过"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert done["status"] == "done"
|
||
assert done["ticket"]["verified_by"] == owner["user_id"]
|
||
assert done["ticket"]["resolution_md"] == "权限已开通，验证通过"
|
||
|
||
# 闭环回流：原任务从 blocked 回到 todo，并且能看到回流事件
|
||
reopened = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
|
||
)
|
||
assert reopened["status"] == "todo"
|
||
assert reopened["completed_at"] is None
|
||
actions = [a["action"] for a in reopened["activities"]]
|
||
assert "promoted_to_ticket" in actions
|
||
assert "ticket_done" in actions
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_reject_goes_back_to_in_progress_and_counts(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 = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)["ticket"]
|
||
base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
|
||
await client.post(f"{base}/accept", headers=helper["headers"])
|
||
await client.post(f"{base}/start", headers=helper["headers"])
|
||
await client.post(f"{base}/submit", json={}, headers=helper["headers"])
|
||
|
||
rejected = unwrap(
|
||
await client.post(
|
||
f"{base}/verify",
|
||
json={"approve": False, "note": "截图里没看到生效"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
assert rejected["status"] == "in_progress"
|
||
assert rejected["ticket"]["reject_count"] == 1
|
||
assert rejected["ticket"]["verified_at"] is None
|
||
|
||
# 原任务仍在 blocked（驳回不算办完）
|
||
still = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
|
||
)
|
||
assert still["status"] == "blocked"
|
||
|
||
# 办理人收到驳回通知
|
||
notes = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
|
||
)
|
||
kinds = [n["kind"] for n in notes["notifications"]]
|
||
assert "ticket_rejected" in kinds
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_assignee_cannot_self_approve(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 = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)["ticket"]
|
||
base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
|
||
await client.post(f"{base}/accept", headers=helper["headers"])
|
||
await client.post(f"{base}/start", headers=helper["headers"])
|
||
await client.post(f"{base}/submit", json={}, headers=helper["headers"])
|
||
|
||
resp = await client.post(f"{base}/verify", json={"approve": True}, headers=helper["headers"])
|
||
assert resp.json()["code"] == 6004 # TICKET_FORBIDDEN
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_illegal_transition_reports_allowed(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 = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)["ticket"]
|
||
|
||
resp = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket/submit",
|
||
json={},
|
||
headers=helper["headers"],
|
||
)
|
||
body = resp.json()
|
||
assert body["code"] == 6003
|
||
assert body["data"]["from"] == "open"
|
||
# allowed 回答"现在还能去哪儿"，allowed_from 回答"这个动作要什么前置状态"
|
||
assert body["data"]["allowed"] == ["accepted", "canceled"]
|
||
assert body["data"]["allowed_from"] == ["in_progress"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_promote_requires_member_assignee_and_no_double(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "升级校验")
|
||
task = await _task(client, owner, ws["id"])
|
||
stranger = await login(client) # 不在工作区
|
||
|
||
bad = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": stranger["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
assert bad.json()["code"] == 1001
|
||
|
||
helper = await _member(client, owner, ws)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
twice = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
assert twice.json()["code"] == 6002
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_list_scopes(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"], "等设计稿")
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
mine_as_helper = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/tickets?role=assigned", headers=helper["headers"]
|
||
)
|
||
)
|
||
assert len(mine_as_helper["tickets"]) == 1
|
||
assert mine_as_helper["tickets"][0]["title"] == "等设计稿"
|
||
|
||
requested = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/tickets?role=requested", headers=owner["headers"]
|
||
)
|
||
)
|
||
assert len(requested["tickets"]) == 1
|
||
|
||
# 只按状态过滤
|
||
empty = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/tickets?role=assigned&status=done",
|
||
headers=helper["headers"],
|
||
)
|
||
)
|
||
assert empty["tickets"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cross_tenant_ticket_invisible(client: AsyncClient) -> None:
|
||
alice = await login(client)
|
||
bob = await login(client)
|
||
ws_a = await create_workspace(client, alice, "A 公司")
|
||
ws_b = await create_workspace(client, bob, "B 公司")
|
||
|
||
helper = await _member(client, alice, ws_a)
|
||
task = await _task(client, alice, ws_a["id"])
|
||
ticket = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws_a['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=alice["headers"],
|
||
)
|
||
)["ticket"]
|
||
|
||
# 用 B 工作区去访问 A 的工单：一律"不存在"，不泄漏存在性
|
||
detail = await client.get(
|
||
f"{V1}/workspaces/{ws_b['id']}/items/{ticket['id']}", headers=bob["headers"]
|
||
)
|
||
assert detail.json()["code"] in (5001, 3003)
|
||
act = await client.post(
|
||
f"{V1}/workspaces/{ws_b['id']}/items/{ticket['id']}/ticket/accept", headers=bob["headers"]
|
||
)
|
||
assert act.json()["code"] in (5001, 3003)
|
||
|
||
# A 的工单列表在 B 里是空的
|
||
listing = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws_b['id']}/tickets?role=all", headers=bob["headers"])
|
||
)
|
||
assert listing["tickets"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_links_parent_child_and_related(client: AsyncClient) -> None:
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "关系")
|
||
parent = await _task(client, owner, ws["id"], "上线新版")
|
||
child = await _task(client, owner, ws["id"], "写发布说明")
|
||
other = await _task(client, owner, ws["id"], "相关：监控告警")
|
||
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
|
||
json={"to_item_id": child["id"], "link_type": "parent"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
|
||
json={"to_item_id": other["id"], "link_type": "related"},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
links = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links", headers=owner["headers"]
|
||
)
|
||
)
|
||
assert [c["item"]["id"] for c in links["children"]] == [child["id"]]
|
||
assert [r["item"]["id"] for r in links["related"]] == [other["id"]]
|
||
assert links["parents"] == []
|
||
|
||
# 从子项看，父项在 parents 里
|
||
from_child = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/items/{child['id']}/links", headers=owner["headers"]
|
||
)
|
||
)
|
||
assert [p["item"]["id"] for p in from_child["parents"]] == [parent["id"]]
|
||
|
||
# 重复挂同一条关系 → 冲突
|
||
dup = await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
|
||
json={"to_item_id": child["id"], "link_type": "parent"},
|
||
headers=owner["headers"],
|
||
)
|
||
assert dup.json()["code"] == 5005
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_notifications_read_flow(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"])
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
count = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/notifications/unread-count", headers=helper["headers"]
|
||
)
|
||
)
|
||
assert count["unread"] == 1
|
||
|
||
notes = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
|
||
)
|
||
note_id = notes["notifications"][0]["id"]
|
||
unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/notifications/{note_id}/read", headers=helper["headers"]
|
||
)
|
||
)
|
||
after = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/notifications/unread-count", headers=helper["headers"]
|
||
)
|
||
)
|
||
assert after["unread"] == 0
|
||
|
||
# 别人的通知看不到（同工作区的另一个人）
|
||
other = await _member(client, owner, ws)
|
||
stranger_notes = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/notifications?unread_only=false", headers=other["headers"]
|
||
)
|
||
)
|
||
assert stranger_notes["notifications"] == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_ticket_payload_always_has_tags(client: AsyncClient) -> None:
|
||
"""回归：工单接口曾经自己拼 item 字典，漏了 `tags`，
|
||
前端 `item.tags.map(...)` 直接白屏。任何返回 item 的接口都得带 tags 数组。"""
|
||
owner = await login(client)
|
||
ws = await create_workspace(client, owner, "标签回归")
|
||
helper = await _member(client, owner, ws)
|
||
task = await _task(client, owner, ws["id"], "带标签的任务")
|
||
|
||
tagged = unwrap(
|
||
await client.patch(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}",
|
||
json={"tag_names": ["权限", "运维"], "version": task["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
|
||
data = unwrap(
|
||
await client.post(
|
||
f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
|
||
json={"assignee_id": helper["user_id"], "version": tagged["version"]},
|
||
headers=owner["headers"],
|
||
)
|
||
)
|
||
# 工单是独立的一条事项，标签不自动继承；关键是字段必须在、而且必须是数组
|
||
assert isinstance(data["task"]["tags"], list)
|
||
assert isinstance(data["ticket"]["tags"], list)
|
||
assert sorted(t["name"] for t in tagged["tags"]) == ["权限", "运维"]
|
||
|
||
# 列表与详情也要一致
|
||
listing = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/tickets?role=all", headers=owner["headers"])
|
||
)
|
||
assert isinstance(listing["tickets"][0]["tags"], list)
|
||
|
||
# 详情接口（任务与工单走同一个入口）都必须带 tags
|
||
task_detail = unwrap(
|
||
await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
|
||
)
|
||
assert sorted(t["name"] for t in task_detail["tags"]) == ["权限", "运维"]
|
||
ticket_detail = unwrap(
|
||
await client.get(
|
||
f"{V1}/workspaces/{ws['id']}/items/{data['ticket']['id']}", headers=owner["headers"]
|
||
)
|
||
)
|
||
assert isinstance(ticket_detail["tags"], list)
|