83 lines
2.9 KiB
Python
83 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from services.webhooks.delivery import (
|
|
canonical_json,
|
|
should_retry,
|
|
webhook_signature,
|
|
)
|
|
from services.webhooks.events import analysis_completed_payload
|
|
from services.webhooks.security import UnsafeWebhookUrl, validate_webhook_url
|
|
|
|
|
|
class WebhookSecurityTest(unittest.IsolatedAsyncioTestCase):
|
|
async def test_requires_https(self):
|
|
with self.assertRaises(UnsafeWebhookUrl):
|
|
await validate_webhook_url("http://hooks.example.com/events")
|
|
|
|
async def test_rejects_loopback_ip(self):
|
|
with self.assertRaises(UnsafeWebhookUrl):
|
|
await validate_webhook_url("https://127.0.0.1/events")
|
|
|
|
async def test_rejects_hostname_resolving_to_private_ip(self):
|
|
records = [(2, 1, 6, "", ("10.0.0.8", 443))]
|
|
with patch("services.webhooks.security.socket.getaddrinfo", return_value=records):
|
|
with self.assertRaises(UnsafeWebhookUrl):
|
|
await validate_webhook_url("https://hooks.example.com/events")
|
|
|
|
async def test_accepts_hostname_resolving_to_public_ip(self):
|
|
records = [(2, 1, 6, "", ("8.8.8.8", 443))]
|
|
with patch("services.webhooks.security.socket.getaddrinfo", return_value=records):
|
|
result = await validate_webhook_url("https://hooks.example.com/events")
|
|
self.assertEqual(result, "https://hooks.example.com/events")
|
|
|
|
|
|
class WebhookDeliveryTest(unittest.TestCase):
|
|
def test_signature_uses_timestamp_dot_raw_body(self):
|
|
body = canonical_json({"b": 2, "a": "中文"})
|
|
expected = hmac.new(
|
|
b"secret",
|
|
b"1700000000." + body,
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
self.assertEqual(
|
|
webhook_signature("secret", "1700000000", body),
|
|
f"v1={expected}",
|
|
)
|
|
|
|
def test_only_transient_responses_are_retried(self):
|
|
for status in (None, 408, 429, 500, 503):
|
|
self.assertTrue(should_retry(status))
|
|
for status in (200, 301, 400, 401, 404):
|
|
self.assertFalse(should_retry(status))
|
|
|
|
def test_analysis_event_has_stable_envelope(self):
|
|
conversation = SimpleNamespace(
|
|
id="conv_001",
|
|
assistant_id="asst_001",
|
|
assistant_name="销售助手",
|
|
)
|
|
from datetime import UTC, datetime
|
|
|
|
payload = analysis_completed_payload(
|
|
conversation, # type: ignore[arg-type]
|
|
{"customer_intent": "high"},
|
|
event_id="evt_001",
|
|
timestamp=datetime(2026, 8, 7, tzinfo=UTC),
|
|
)
|
|
self.assertEqual(payload["id"], "evt_001")
|
|
self.assertEqual(payload["event"], "call.analysis.completed")
|
|
self.assertEqual(payload["data"]["conversationId"], "conv_001")
|
|
self.assertEqual(
|
|
payload["data"]["analysis"], {"customer_intent": "high"}
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|