File size: 5,563 Bytes
848219e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 |
import unittest
import json
import hmac
import hashlib
import requests
from unittest.mock import patch
from http.client import HTTPConnection
import logging
# Configure logging for tests
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# Base URL of your Hugging Face Space
BASE_URL = "https://<your-space>.hf.space" # Replace with your Space URL
WEBHOOK_SECRET = "your-base64-encoded-secret" # Replace with your webhook secret
class TestPipecatWebhookAndStart(unittest.TestCase):
def setUp(self):
"""Set up test case with common variables."""
self.webhook_url = f"{BASE_URL}/webhook"
self.start_url = f"{BASE_URL}/start"
self.valid_payload = {
"event": "dialin.connected",
"callId": "test-call-123",
"callDomain": "test.daily.co",
"From": "+12345678901",
"To": "+12345678902",
"timestamp": "2025-05-11T01:33:07.000Z"
}
self.stopped_payload = {
"event": "dialin.stopped",
"callId": "test-call-123",
"callDomain": "test.daily.co",
"timestamp": "2025-05-11T01:33:07.000Z"
}
def generate_signature(self, payload):
"""Generate HMAC-SHA256 signature for a payload."""
payload_str = json.dumps(payload)
computed = hmac.new(
WEBHOOK_SECRET.encode('utf-8'),
payload_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
return computed
def test_server_availability(self):
"""Test that the server is running and accessible."""
try:
response = requests.get(BASE_URL, timeout=5)
# Allow 404 for root since the app doesn't serve it
self.assertIn(response.status_code, [200, 404], f"Server not accessible: {response.status_code}")
logger.info("Server availability test passed")
except requests.RequestException as e:
self.fail(f"Server not reachable: {str(e)}")
def test_webhook_dialin_connected(self):
"""Test webhook handling of dialin.connected event."""
headers = {"X-Daily-Signature": self.generate_signature(self.valid_payload)}
with patch('requests.post') as mocked_post:
mocked_post.return_value.status_code = 200
mocked_post.return_value.text = '{"status": "Bot started"}'
response = requests.post(
self.webhook_url,
json=self.valid_payload,
headers=headers,
timeout=5
)
self.assertEqual(response.status_code, 200, f"Webhook failed: {response.text}")
self.assertEqual(response.json(), {"status": "Event received"})
mocked_post.assert_called_once_with(
self.start_url,
json=self.valid_payload,
timeout=5
)
logger.info("Webhook dialin.connected test passed")
def test_webhook_dialin_stopped(self):
"""Test webhook handling of dialin.stopped event."""
headers = {"X-Daily-Signature": self.generate_signature(self.stopped_payload)}
response = requests.post(
self.webhook_url,
json=self.stopped_payload,
headers=headers,
timeout=5
)
self.assertEqual(response.status_code, 200, f"Webhook failed: {response.text}")
self.assertEqual(response.json(), {"status": "Event received"})
logger.info("Webhook dialin.stopped test passed")
def test_webhook_invalid_signature(self):
"""Test webhook rejects requests with invalid signature."""
headers = {"X-Daily-Signature": "invalid-signature"}
response = requests.post(
self.webhook_url,
json=self.valid_payload,
headers=headers,
timeout=5
)
self.assertEqual(response.status_code, 401, f"Expected 401 for invalid signature: {response.status_code}")
logger.info("Webhook invalid signature test passed")
def test_start_endpoint_valid_payload(self):
"""Test /start endpoint with valid dial-in payload."""
response = requests.post(
self.start_url,
json=self.valid_payload,
timeout=5
)
self.assertEqual(response.status_code, 200, f"Start endpoint failed: {response.text}")
response_json = response.json()
self.assertEqual(response_json.get("status"), "Bot started")
self.assertEqual(response_json.get("bot_type"), "call_transfer")
logger.info("Start endpoint valid payload test passed")
def test_start_endpoint_invalid_payload(self):
"""Test /start endpoint with invalid payload."""
invalid_payload = {"invalid": "data"}
response = requests.post(
self.start_url,
json=invalid_payload,
timeout=5
)
self.assertEqual(response.status_code, 400, f"Expected 400 for invalid payload: {response.status_code}")
logger.info("Start endpoint invalid payload test passed")
def test_404_on_root(self):
"""Test that root endpoint returns 404 (as seen in logs)."""
response = requests.get(f"{BASE_URL}/?__theme=dark", timeout=5)
self.assertEqual(response.status_code, 404, f"Expected 404 for root endpoint: {response.status_code}")
logger.info("Root endpoint 404 test passed")
if __name__ == '__main__':
unittest.main() |