321 lines
12 KiB
Python
321 lines
12 KiB
Python
"""Slack IM 适配器单元测试 (U12)。
|
||
|
||
覆盖场景：
|
||
- 签名校验（有效/无效/过期）
|
||
- URL 验证（challenge / token 不匹配）
|
||
- Events API 消息解析
|
||
- Slash Command 解析
|
||
- <@U12345> 提及剥离
|
||
- 不支持事件类型
|
||
- send_message 成功/失败
|
||
- bot_token 构造器存储
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import time
|
||
from typing import Any
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
from urllib.parse import urlencode
|
||
|
||
import pytest
|
||
|
||
from agentkit.channels.base import ChannelType, IncomingMessage, OutgoingMessage
|
||
from agentkit.channels.slack import (
|
||
SlackMessageAdapter,
|
||
SignatureVerificationError,
|
||
URLVerificationChallenge,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 辅助函数
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _sign(signing_secret: str, timestamp: int, body: bytes) -> str:
|
||
"""构造 Slack X-Slack-Signature 头值：v0= + hmac-sha256(signing_secret, "v0:{ts}:{body}")。"""
|
||
base = f"v0:{timestamp}:{body.decode('utf-8')}"
|
||
digest = hmac.new(
|
||
signing_secret.encode("utf-8"),
|
||
base.encode("utf-8"),
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
return f"v0={digest}"
|
||
|
||
|
||
def _make_event_body(
|
||
*,
|
||
text: str = "hello",
|
||
event_type: str = "message",
|
||
event_id: str = "evt_001",
|
||
user: str = "U123",
|
||
channel: str = "C456",
|
||
ts: str = "1700000000.000123",
|
||
) -> dict[str, Any]:
|
||
"""构造 Slack Events API 事件 JSON 体。"""
|
||
return {
|
||
"event_id": event_id,
|
||
"type": "event_callback",
|
||
"event": {
|
||
"type": event_type,
|
||
"user": user,
|
||
"channel": channel,
|
||
"text": text,
|
||
"ts": ts,
|
||
},
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 签名校验
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSignatureVerification:
|
||
"""Slack 签名校验。"""
|
||
|
||
async def test_valid_signature(self):
|
||
"""正确 v0 签名返回 True。"""
|
||
signing_secret = "sh_secret"
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret)
|
||
body = json.dumps(_make_event_body()).encode("utf-8")
|
||
ts = int(time.time())
|
||
headers = {
|
||
"X-Slack-Signature": _sign(signing_secret, ts, body),
|
||
"X-Slack-Request-Timestamp": str(ts),
|
||
}
|
||
assert await adapter.verify_signature(headers, body) is True
|
||
|
||
async def test_invalid_signature(self):
|
||
"""篡改签名返回 False。"""
|
||
signing_secret = "sh_secret"
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret)
|
||
body = json.dumps(_make_event_body()).encode("utf-8")
|
||
ts = int(time.time())
|
||
headers = {"X-Slack-Signature": "v0=tampered", "X-Slack-Request-Timestamp": str(ts)}
|
||
assert await adapter.verify_signature(headers, body) is False
|
||
|
||
async def test_expired_timestamp(self):
|
||
"""时间戳超过 5 分钟返回 False。"""
|
||
signing_secret = "sh_secret"
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret=signing_secret)
|
||
body = json.dumps(_make_event_body()).encode("utf-8")
|
||
old_ts = int(time.time()) - 600
|
||
headers = {
|
||
"X-Slack-Signature": _sign(signing_secret, old_ts, body),
|
||
"X-Slack-Request-Timestamp": str(old_ts),
|
||
}
|
||
assert await adapter.verify_signature(headers, body) is False
|
||
|
||
async def test_missing_signature_headers(self):
|
||
"""缺签名头返回 False。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
assert await adapter.verify_signature({}, b"{}") is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# URL 验证
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestURLVerification:
|
||
"""Slack URL 验证流程。"""
|
||
|
||
async def test_url_verification_raises_challenge(self):
|
||
"""url_verification 事件抛 URLVerificationChallenge。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
body = json.dumps(
|
||
{"type": "url_verification", "challenge": "verify_abc", "token": "t"}
|
||
).encode("utf-8")
|
||
with pytest.raises(URLVerificationChallenge) as exc_info:
|
||
await adapter.receive_message({}, body)
|
||
assert exc_info.value.challenge == "verify_abc"
|
||
|
||
async def test_url_verification_token_mismatch_raises(self):
|
||
"""verification_token 不匹配抛 SignatureVerificationError。"""
|
||
adapter = SlackMessageAdapter(
|
||
bot_token="xoxb-x",
|
||
signing_secret="s",
|
||
verification_token="right_token",
|
||
)
|
||
body = json.dumps(
|
||
{"type": "url_verification", "challenge": "abc", "token": "wrong_token"}
|
||
).encode("utf-8")
|
||
with pytest.raises(SignatureVerificationError):
|
||
await adapter.receive_message({}, body)
|
||
|
||
async def test_url_verification_token_match_passes(self):
|
||
"""verification_token 匹配时正常抛出 challenge。"""
|
||
adapter = SlackMessageAdapter(
|
||
bot_token="xoxb-x",
|
||
signing_secret="s",
|
||
verification_token="right_token",
|
||
)
|
||
body = json.dumps(
|
||
{"type": "url_verification", "challenge": "ok123", "token": "right_token"}
|
||
).encode("utf-8")
|
||
with pytest.raises(URLVerificationChallenge) as exc_info:
|
||
await adapter.receive_message({}, body)
|
||
assert exc_info.value.challenge == "ok123"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 消息解析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestEventsAPIParsing:
|
||
"""Events API 消息解析。"""
|
||
|
||
async def test_event_message_parsing(self):
|
||
"""Events API 消息解析为 IncomingMessage。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
body = json.dumps(_make_event_body(text="hello slack")).encode("utf-8")
|
||
msg = await adapter.receive_message({}, body)
|
||
assert isinstance(msg, IncomingMessage)
|
||
assert msg.channel == ChannelType.SLACK
|
||
assert msg.content == "hello slack"
|
||
assert msg.platform_message_id == "evt_001"
|
||
assert msg.user_id == "U123"
|
||
assert msg.chat_id == "C456"
|
||
assert msg.timestamp == "1700000000.000123"
|
||
|
||
async def test_mention_stripping(self):
|
||
"""<@U12345> 提及标记被剥离。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
body = json.dumps(_make_event_body(text="<@U12345> hello there")).encode("utf-8")
|
||
msg = await adapter.receive_message({}, body)
|
||
assert msg.content == "hello there"
|
||
|
||
async def test_mention_with_name_stripped(self):
|
||
"""<@U12345|name> 形式的提及标记也被剥离。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
body = json.dumps(_make_event_body(text="<@U999|bob> hi")).encode("utf-8")
|
||
msg = await adapter.receive_message({}, body)
|
||
assert msg.content == "hi"
|
||
|
||
async def test_unsupported_event_type(self):
|
||
"""非 message/app_mention 事件返回 unsupported 占位内容。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
body = json.dumps(_make_event_body(event_type="reaction_added")).encode("utf-8")
|
||
msg = await adapter.receive_message({}, body)
|
||
assert msg.content.startswith("[unsupported event type: reaction_added]")
|
||
|
||
|
||
class TestSlashCommandParsing:
|
||
"""Slash Command 解析。"""
|
||
|
||
async def test_slash_command_parsing(self):
|
||
"""form-encoded slash command 解析为 IncomingMessage。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
form = urlencode(
|
||
{
|
||
"text": "echo me",
|
||
"user_id": "U_slash",
|
||
"channel_id": "C_slash",
|
||
"command": "/echo",
|
||
"response_url": "https://hooks.slack.com/x",
|
||
}
|
||
)
|
||
body = form.encode("utf-8")
|
||
msg = await adapter.receive_message(
|
||
{"Content-Type": "application/x-www-form-urlencoded"}, body
|
||
)
|
||
assert msg.channel == ChannelType.SLACK
|
||
assert msg.content == "echo me"
|
||
assert msg.user_id == "U_slash"
|
||
assert msg.chat_id == "C_slash"
|
||
assert msg.platform_message_id.startswith("slash-")
|
||
assert msg.raw_event["command"] == "/echo"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# send_message
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestSendMessage:
|
||
"""send_message 行为。"""
|
||
|
||
async def test_send_message_success(self):
|
||
"""HTTP 200 + ok=true 返回 True。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-123", signing_secret="s")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
mock_resp.json.return_value = {"ok": True, "channel": "C1", "ts": "1.0"}
|
||
|
||
mock_client = AsyncMock()
|
||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||
adapter._client = mock_client
|
||
|
||
out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi")
|
||
assert await adapter.send_message(out) is True
|
||
|
||
call = mock_client.post.call_args
|
||
assert "chat.postMessage" in call.args[0]
|
||
assert call.kwargs["headers"]["Authorization"] == "Bearer xoxb-123"
|
||
assert call.kwargs["json"]["channel"] == "C1"
|
||
|
||
async def test_send_message_failure(self):
|
||
"""HTTP 200 但 ok=false 返回 False。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 200
|
||
mock_resp.json.return_value = {"ok": False, "error": "channel_not_found"}
|
||
|
||
mock_client = AsyncMock()
|
||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||
adapter._client = mock_client
|
||
|
||
out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi")
|
||
assert await adapter.send_message(out) is False
|
||
|
||
async def test_send_message_http_error(self):
|
||
"""非 200 HTTP 状态返回 False。"""
|
||
adapter = SlackMessageAdapter(bot_token="xoxb-x", signing_secret="s")
|
||
mock_resp = MagicMock()
|
||
mock_resp.status_code = 500
|
||
mock_resp.text = "server error"
|
||
|
||
mock_client = AsyncMock()
|
||
mock_client.post = AsyncMock(return_value=mock_resp)
|
||
adapter._client = mock_client
|
||
|
||
out = OutgoingMessage(channel=ChannelType.SLACK, chat_id="C1", content="hi")
|
||
assert await adapter.send_message(out) is False
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 构造器 / 资源释放
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestConstructor:
|
||
"""构造器与资源释放。"""
|
||
|
||
def test_bot_token_stored(self):
|
||
"""bot_token 由构造器存储。"""
|
||
adapter = SlackMessageAdapter(
|
||
bot_token="xoxb-secret-token", signing_secret="s", verification_token="vt"
|
||
)
|
||
assert adapter.bot_token == "xoxb-secret-token"
|
||
assert adapter.signing_secret == "s"
|
||
assert adapter.verification_token == "vt"
|
||
|
||
async def test_close_no_client_is_noop(self):
|
||
"""未创建 httpx 客户端时 close 不抛异常。"""
|
||
adapter = SlackMessageAdapter(bot_token="x", signing_secret="s")
|
||
await adapter.close()
|
||
|
||
async def test_close_resets_client(self):
|
||
"""close 后客户端引用清空。"""
|
||
adapter = SlackMessageAdapter(bot_token="x", signing_secret="s")
|
||
adapter._get_client()
|
||
assert adapter._client is not None
|
||
await adapter.close()
|
||
assert adapter._client is None
|