"""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()