Upload test_webhook_and_start.py

test_webhook_and_start.py

@@ -0,0 +1,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()