Update test_webhook_and_start.py

test_webhook_and_start.py

@@ -4,20 +4,30 @@ import hmac
 import hashlib
 import requests
 from unittest.mock import patch
-from http.client import HTTPConnection
 import logging
+import os
 
-# Configure logging for tests
-logging.basicConfig(level=logging.INFO)
+# 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://<your-space>.hf.space"  # Replace with your Space URL
-WEBHOOK_SECRET = "your-base64-encoded-secret"  # Replace with your webhook secret
+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 = {
@@ -37,6 +47,7 @@ class TestPipecatWebhookAndStart(unittest.TestCase):
 
     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'),
@@ -47,95 +58,123 @@ class TestPipecatWebhookAndStart(unittest.TestCase):
 
     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)
-            # 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")
+            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.valid_payload,
+                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"})
-            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")
+            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"}
-        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")
+        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."""
-        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")
+        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"}
-        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")
+        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)."""
-        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")
+        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__':
-    unittest.main()
+    logger.info("Starting test suite")
+    unittest.main()
+    logger.info("Test suite completed")