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://.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()