"""微信（clawbot）接入：一条消息进来，变成任务或知识。 测的是**整条链路**：令牌 → 租户上下文 → 建事项 → 留痕 → 重复推送幂等。 webhook 不带登录态，所以这里故意不用 `auth()` 头，只有管理接口才带。 """ from __future__ import annotations from typing import Any from httpx import AsyncClient from .conftest import create_workspace, login, unwrap async def _channel(client: AsyncClient, user: dict[str, Any], ws: dict[str, Any]) -> dict: return unwrap( await client.post( f"/api/v1/workspaces/{ws['id']}/integrations", json={"name": "我的微信"}, headers=user["headers"], ) ) async def _post(client: AsyncClient, token: str, payload: dict[str, Any]): return await client.post(f"/api/v1/integrations/clawbot/{token}", json=payload) async def test_plain_message_becomes_task(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信接入") channel = await _channel(client, user, ws) resp = await _post( client, channel["token"], {"msg_id": "wx-1", "from": "steven", "text": "给客户回电话\n顺便确认下报价"}, ) data = unwrap(resp) assert data["action"] == "task" assert data["duplicate"] is False assert "给客户回电话" in data["reply"] items = unwrap( await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) )["items"] task = next(i for i in items if i["title"] == "给客户回电话") assert task["kind"] == "task" assert task["status"] == "inbox" assert "顺便确认下报价" in task["body_md"] async def test_knowledge_prefix_creates_draft_with_tags(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信知识") channel = await _channel(client, user, ws) data = unwrap( await _post( client, channel["token"], {"text": "知识：回滚前先停灰度 #发布 #回滚\n1. 停流量\n2. 回滚镜像"}, ) ) assert data["action"] == "knowledge" docs = unwrap( await client.get( f"/api/v1/workspaces/{ws['id']}/knowledge?status=draft", headers=user["headers"] ) )["items"] doc = next(d for d in docs if d["title"] == "回滚前先停灰度") assert doc["knowledge"]["status_hint"] == "draft" assert doc["knowledge"]["doc_type"] == "note" assert {t["name"] for t in doc["tags"]} == {"发布", "回滚"} # 标签从标题摘出去以后，正文里不该再留 # 号 assert "#发布" not in doc["body_md"] async def test_help_reply_creates_nothing(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信帮助") channel = await _channel(client, user, ws) data = unwrap(await _post(client, channel["token"], {"text": "帮助"})) assert data["action"] == "help" assert "记成任务" in data["reply"] items = unwrap( await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) )["items"] assert items == [] async def test_unknown_token_is_404(client: AsyncClient) -> None: resp = await client.post("/api/v1/integrations/clawbot/nope", json={"text": "x"}) assert resp.status_code == 404 assert resp.json()["code"] == 9002 async def test_disabled_channel_refuses_messages(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信停用") channel = await _channel(client, user, ws) await client.patch( f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}", json={"status": "disabled"}, headers=user["headers"], ) resp = await _post(client, channel["token"], {"text": "这条不该进去"}) assert resp.status_code == 403 assert resp.json()["code"] == 9003 async def test_same_message_id_is_idempotent(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信重推") channel = await _channel(client, user, ws) payload = {"msg_id": "dup-1", "text": "只该有一条"} first = unwrap(await _post(client, channel["token"], payload)) second = unwrap(await _post(client, channel["token"], payload)) assert second["duplicate"] is True assert second["reply"] == first["reply"] items = unwrap( await client.get(f"/api/v1/workspaces/{ws['id']}/items", headers=user["headers"]) )["items"] assert [i["title"] for i in items] == ["只该有一条"] async def test_get_verify_echoes_challenge(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信校验") channel = await _channel(client, user, ws) resp = await client.get(f"/api/v1/integrations/clawbot/{channel['token']}?echostr=abc123") assert resp.status_code == 200 assert resp.text == "abc123" async def test_token_belongs_to_one_workspace(client: AsyncClient) -> None: user = await login(client) ws_a = await create_workspace(client, user, "A 区") ws_b = await create_workspace(client, user, "B 区") channel = await _channel(client, user, ws_a) await _post(client, channel["token"], {"text": "只该进 A 区"}) items_a = unwrap( await client.get(f"/api/v1/workspaces/{ws_a['id']}/items", headers=user["headers"]) )["items"] items_b = unwrap( await client.get(f"/api/v1/workspaces/{ws_b['id']}/items", headers=user["headers"]) )["items"] assert [i["title"] for i in items_a] == ["只该进 A 区"] assert items_b == [] async def test_rotate_invalidates_old_token(client: AsyncClient) -> None: user = await login(client) ws = await create_workspace(client, user, "微信换令牌") channel = await _channel(client, user, ws) rotated = unwrap( await client.post( f"/api/v1/workspaces/{ws['id']}/integrations/{channel['channel']['id']}/rotate", headers=user["headers"], ) ) assert rotated["token"] != channel["token"] old = await _post(client, channel["token"], {"text": "旧令牌"}) assert old.status_code == 404 new = unwrap(await _post(client, rotated["token"], {"text": "新令牌能用"})) assert new["action"] == "task" async def test_member_cannot_manage_channels(client: AsyncClient) -> None: owner = await login(client) ws = await create_workspace(client, owner, "微信权限") member = await login(client) code = unwrap( await client.post( f"/api/v1/workspaces/{ws['id']}/invitations", json={"role": "member"}, headers=owner["headers"], ) ) await client.post( f"/api/v1/workspaces/{ws['id']}/join", json={"code": code["code"]}, headers=member["headers"], ) resp = await client.post( f"/api/v1/workspaces/{ws['id']}/integrations", json={"name": "偷加的通道"}, headers=member["headers"], ) assert resp.status_code == 403 assert resp.json()["code"] in (3004, 3003) async def test_webhook_url_carries_api_prefix(client: AsyncClient) -> None: """回调地址必须带 /api/v1：生产上应用挂在域名根，漏了前缀就是 405。""" user = await login(client) ws = await create_workspace(client, user, "微信地址") channel = await _channel(client, user, ws) assert channel["webhook_url"].endswith(f"/api/v1/integrations/clawbot/{channel['token']}")