File size: 7,418 Bytes
bda4f40 848219e ca10ff4 848219e ca10ff4 bda4f40 ca10ff4 848219e bda4f40 dcd4279 47cee22 bda4f40 ca10ff4 bda4f40 ca10ff4 bda4f40 ca10ff4 bda4f40 ca10ff4 bda4f40 ca10ff4 848219e bda4f40 47cee22 bda4f40 47cee22 bda4f40 848219e ca10ff4 bda4f40 ca10ff4 |
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 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 |
import unittest
import json
import hmac
import hashlib
import requests
from unittest.mock import patch
import logging
import os
# Configure logging to write to both file and console
LOG_FILE = "/app/test.log"
os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler(LOG_FILE),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
# Base URL of your Hugging Face Space
BASE_URL="https://deadmon-pipecat.hf.space/webhook" # Replace with your endpoint
WEBHOOK_SECRET="eW91ci1zZWNyZXQtc3RyaW5n" # Base64-encoded secret
class TestPipecatWebhookAndStart(unittest.TestCase):
def setUp(self):
"""Set up test case with common variables."""
logger.info("Setting up test case")
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."""
logger.debug("Generating HMAC signature")
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."""
logger.info("Starting test_server_availability")
try:
response = requests.get(BASE_URL, timeout=5)
self.assertIn(response.status_code, [200, 404], f"Server not accessible: {response.status_code}")
logger.info(f"Server availability test passed: Status code {response.status_code}")
except requests.RequestException as e:
logger.error(f"Server not reachable: {str(e)}")
self.fail(f"Server not reachable: {str(e)}")
def test_webhook_dialin_connected(self):
"""Test webhook handling of dialin.connected event."""
logger.info("Starting test_webhook_dialin_connected")
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"}'
try:
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")
except Exception as e:
logger.error(f"Webhook dialin.connected test failed: {str(e)}")
raise
def test_webhook_dialin_stopped(self):
"""Test webhook handling of dialin.stopped event."""
logger.info("Starting test_webhook_dialin_stopped")
headers = {"X-Daily-Signature": self.generate_signature(self.stopped_payload)}
try:
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")
except Exception as e:
logger.error(f"Webhook dialin.stopped test failed: {str(e)}")
raise
def test_webhook_invalid_signature(self):
"""Test webhook rejects requests with invalid signature."""
logger.info("Starting test_webhook_invalid_signature")
headers = {"X-Daily-Signature": "invalid-signature"}
try:
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")
except Exception as e:
logger.error(f"Webhook invalid signature test failed: {str(e)}")
raise
def test_start_endpoint_valid_payload(self):
"""Test /start endpoint with valid dial-in payload."""
logger.info("Starting test_start_endpoint_valid_payload")
try:
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")
except Exception as e:
logger.error(f"Start endpoint valid payload test failed: {str(e)}")
raise
def test_start_endpoint_invalid_payload(self):
"""Test /start endpoint with invalid payload."""
logger.info("Starting test_start_endpoint_invalid_payload")
invalid_payload = {"invalid": "data"}
try:
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")
except Exception as e:
logger.error(f"Start endpoint invalid payload test failed: {str(e)}")
raise
def test_404_on_root(self):
"""Test that root endpoint returns 404 (as seen in logs)."""
logger.info("Starting test_404_on_root")
try:
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")
except Exception as e:
logger.error(f"Root endpoint 404 test failed: {str(e)}")
raise
if __name__ == '__main__':
logger.info("Starting test suite")
unittest.main()
logger.info("Test suite completed") |