237 lines
8.3 KiB
Python
237 lines
8.3 KiB
Python
"""webhooks.py 单测：签名校验、事件解析、命令派发、用户映射、通知门禁。
|
||
|
||
只用标准库，不联网、不连真实 GitHub/Slack、不跑真实模型回合。
|
||
签名测试用的 secret 均为单测内临时值，不是生产密钥。
|
||
"""
|
||
|
||
import hashlib
|
||
import hmac
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
import unittest
|
||
|
||
# 本文件位于 cloud/tests/，被测模块位于 cloud/；cloud/ 目前不是包
|
||
# （协调员统一创建 __init__.py），故把 cloud/ 加到 sys.path 后按
|
||
# 顶层模块导入 webhooks。
|
||
sys.path.insert(
|
||
0, os.path.normpath(os.path.join(os.path.dirname(__file__), ".."))
|
||
)
|
||
|
||
import webhooks # noqa: E402
|
||
from webhooks import ( # noqa: E402
|
||
ALLOWED_NOTIFICATION_EVENTS,
|
||
NotificationService,
|
||
SlackUserMap,
|
||
parse_github_pr_comment,
|
||
parse_slack_command,
|
||
verify_github_signature,
|
||
verify_slack_signature,
|
||
)
|
||
|
||
|
||
def _github_signature(secret, payload):
|
||
return "sha256=" + hmac.new(
|
||
secret.encode("utf-8"), payload, hashlib.sha256
|
||
).hexdigest()
|
||
|
||
|
||
def _slack_signature(secret, timestamp, body):
|
||
basestring = f"v0:{timestamp}:{body}"
|
||
return "v0=" + hmac.new(
|
||
secret.encode("utf-8"), basestring.encode("utf-8"), hashlib.sha256
|
||
).hexdigest()
|
||
|
||
|
||
class TestGithubSignature(unittest.TestCase):
|
||
def test_correct_signature_passes(self):
|
||
secret = "test-github-secret"
|
||
payload = b'{"action":"created"}'
|
||
header = _github_signature(secret, payload)
|
||
self.assertTrue(verify_github_signature(secret, payload, header))
|
||
|
||
def test_wrong_signature_rejected(self):
|
||
secret = "test-github-secret"
|
||
payload = b'{"action":"created"}'
|
||
header = _github_signature("wrong-secret", payload)
|
||
self.assertFalse(verify_github_signature(secret, payload, header))
|
||
|
||
def test_tampered_payload_rejected(self):
|
||
secret = "test-github-secret"
|
||
header = _github_signature(secret, b'{"action":"created"}')
|
||
self.assertFalse(
|
||
verify_github_signature(secret, b'{"action":"deleted"}', header)
|
||
)
|
||
|
||
def test_missing_header_rejected(self):
|
||
self.assertFalse(
|
||
verify_github_signature("test-github-secret", b"{}", None)
|
||
)
|
||
self.assertFalse(verify_github_signature("test-github-secret", b"{}", ""))
|
||
|
||
def test_missing_secret_rejected(self):
|
||
payload = b"{}"
|
||
self.assertFalse(verify_github_signature("", payload, "sha256=abc"))
|
||
self.assertFalse(verify_github_signature(None, payload, "sha256=abc"))
|
||
|
||
|
||
class TestParseGithubPrComment(unittest.TestCase):
|
||
def _payload(self, body, action="created"):
|
||
return {
|
||
"action": action,
|
||
"repository": {"full_name": "owner/repo"},
|
||
"issue": {"number": 42},
|
||
"comment": {"body": body, "user": {"login": "alice"}},
|
||
}
|
||
|
||
def test_fix_review_comment_parsed(self):
|
||
result = parse_github_pr_comment(
|
||
self._payload("请帮忙 /fix-review 看一下")
|
||
)
|
||
self.assertEqual(
|
||
result,
|
||
{
|
||
"type": "fix_review",
|
||
"repo": "owner/repo",
|
||
"pr_number": 42,
|
||
"comment": "请帮忙 /fix-review 看一下",
|
||
"author": "alice",
|
||
},
|
||
)
|
||
|
||
def test_comment_without_trigger_returns_none(self):
|
||
self.assertIsNone(parse_github_pr_comment(self._payload("好看，LGTM")))
|
||
|
||
def test_non_created_action_returns_none(self):
|
||
self.assertIsNone(
|
||
parse_github_pr_comment(
|
||
self._payload("有 /fix-review 但是已编辑", action="edited")
|
||
)
|
||
)
|
||
|
||
|
||
class TestSlackSignature(unittest.TestCase):
|
||
def test_correct_signature_passes(self):
|
||
secret = "test-slack-secret"
|
||
timestamp = str(int(time.time()))
|
||
body = "command=/codex&text=status"
|
||
header = _slack_signature(secret, timestamp, body)
|
||
self.assertTrue(
|
||
verify_slack_signature(secret, timestamp, body, header)
|
||
)
|
||
|
||
def test_wrong_signature_rejected(self):
|
||
secret = "test-slack-secret"
|
||
timestamp = str(int(time.time()))
|
||
body = "command=/codex&text=status"
|
||
header = _slack_signature("wrong-secret", timestamp, body)
|
||
self.assertFalse(
|
||
verify_slack_signature(secret, timestamp, body, header)
|
||
)
|
||
|
||
def test_expired_timestamp_rejected(self):
|
||
# 超过 300 秒：用正确密钥签一个 301 秒前的时间戳，仍应拒绝。
|
||
secret = "test-slack-secret"
|
||
timestamp = str(int(time.time()) - 301)
|
||
body = "command=/codex&text=status"
|
||
header = _slack_signature(secret, timestamp, body)
|
||
self.assertFalse(
|
||
verify_slack_signature(secret, timestamp, body, header)
|
||
)
|
||
|
||
def test_missing_params_rejected(self):
|
||
timestamp = str(int(time.time()))
|
||
self.assertFalse(
|
||
verify_slack_signature("", timestamp, "body", "v0=abc")
|
||
)
|
||
self.assertFalse(
|
||
verify_slack_signature("test-slack-secret", timestamp, "body", None)
|
||
)
|
||
|
||
|
||
class TestParseSlackCommand(unittest.TestCase):
|
||
def test_dispatch_prompt(self):
|
||
self.assertEqual(
|
||
parse_slack_command("/codex 做一个加法函数"),
|
||
{"command": "codex", "prompt": "做一个加法函数"},
|
||
)
|
||
|
||
def test_dispatch_status(self):
|
||
self.assertEqual(
|
||
parse_slack_command("/codex status my-task"),
|
||
{"command": "status", "task": "my-task"},
|
||
)
|
||
|
||
def test_empty_or_non_codex_returns_none_command(self):
|
||
self.assertEqual(
|
||
parse_slack_command(""), {"command": None, "prompt": ""}
|
||
)
|
||
self.assertEqual(
|
||
parse_slack_command("hello"),
|
||
{"command": None, "prompt": "hello"},
|
||
)
|
||
self.assertEqual(
|
||
parse_slack_command("/other do something"),
|
||
{"command": None, "prompt": "/other do something"},
|
||
)
|
||
|
||
|
||
class TestSlackUserMap(unittest.TestCase):
|
||
def test_map_and_lookup(self):
|
||
with tempfile.TemporaryDirectory() as tmpdir:
|
||
db_path = os.path.join(tmpdir, "slack_users.db")
|
||
user_map = SlackUserMap(db_path)
|
||
try:
|
||
self.assertIsNone(user_map.lookup("U123"))
|
||
user_map.map("U123", "user-1")
|
||
self.assertEqual(user_map.lookup("U123"), "user-1")
|
||
# 覆盖映射
|
||
user_map.map("U123", "user-2")
|
||
self.assertEqual(user_map.lookup("U123"), "user-2")
|
||
# 未映射的 id 仍返回 None
|
||
self.assertIsNone(user_map.lookup("U999"))
|
||
finally:
|
||
user_map.close()
|
||
|
||
|
||
class TestNotificationService(unittest.TestCase):
|
||
def test_allowed_event_sent_and_senders_called(self):
|
||
calls = {"webhook": [], "slack": []}
|
||
service = NotificationService(
|
||
webhook_sender=lambda payload: calls["webhook"].append(payload),
|
||
slack_sender=lambda payload: calls["slack"].append(payload),
|
||
)
|
||
result = service.notify("completed", "task-1", "proj-1", "done")
|
||
self.assertEqual(result, {"sent": True, "channels": ["webhook", "slack"]})
|
||
self.assertEqual(len(calls["webhook"]), 1)
|
||
self.assertEqual(len(calls["slack"]), 1)
|
||
self.assertEqual(calls["webhook"][0]["event_type"], "completed")
|
||
self.assertEqual(calls["webhook"][0]["task_id"], "task-1")
|
||
|
||
def test_disallowed_event_not_sent_and_senders_not_called(self):
|
||
calls = []
|
||
service = NotificationService(
|
||
webhook_sender=lambda payload: calls.append(("webhook", payload)),
|
||
slack_sender=lambda payload: calls.append(("slack", payload)),
|
||
)
|
||
result = service.notify("running", "task-1", "proj-1")
|
||
self.assertEqual(result, {"sent": False, "reason": "event not allowed"})
|
||
self.assertEqual(calls, [])
|
||
|
||
def test_no_senders_does_not_error(self):
|
||
service = NotificationService()
|
||
result = service.notify("completed", "task-1", "proj-1")
|
||
self.assertEqual(result, {"sent": True, "channels": []})
|
||
|
||
def test_allowed_events_set(self):
|
||
self.assertEqual(
|
||
ALLOWED_NOTIFICATION_EVENTS,
|
||
{"completed", "failed", "goal-unmet", "merge-failed",
|
||
"review-requested"},
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|