Update test_webhook_and_start.py

test_webhook_and_start.py

@@ -1,4 +1,4 @@
-import unittest
+import pytest
 import json
 import hmac
 import hashlib
@@ -6,9 +6,10 @@ import requests
 from unittest.mock import patch
 import logging
 import os
+from http.client import HTTPConnection
 
 # Configure logging to write to both file and console
-LOG_FILE = "pipecat/test.log"
+LOG_FILE = "/app/test.log"
 os.makedirs(os.path.dirname(LOG_FILE), exist_ok=True)
 logging.basicConfig(
     level=logging.INFO,
@@ -20,161 +21,193 @@ logging.basicConfig(
 )
 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
+# Base URL of your Hugging Face Space or local server
+BASE_URL = "https://<your-space>.hf.space"  # Replace with your Space URL
+WEBHOOK_SECRET = "your-base64-encoded-secret"  # Replace with your webhook secret
+
+@pytest.fixture
+def webhook_url():
+    """Fixture for webhook URL."""
+    return f"{BASE_URL}/webhook"
 
-    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"}
+@pytest.fixture
+def start_url():
+    """Fixture for start URL."""
+    return f"{BASE_URL}/start"
+
+@pytest.fixture
+def valid_dialin_payload():
+    """Fixture for valid dialin.connected payload."""
+    return {
+        "event": "dialin.connected",
+        "callId": "test-call-123",
+        "callDomain": "test.daily.co",
+        "From": "+12345678901",
+        "To": "+12345678902",
+        "timestamp": "2025-05-11T01:33:07.000Z"
+    }
+
+@pytest.fixture
+def stopped_payload():
+    """Fixture for dialin.stopped payload."""
+    return {
+        "event": "dialin.stopped",
+        "callId": "test-call-123",
+        "callDomain": "test.daily.co",
+        "timestamp": "2025-05-11T01:33:07.000Z"
+    }
+
+@pytest.fixture
+def invalid_payload():
+    """Fixture for invalid payload."""
+    return {"invalid": "data"}
+
+def generate_signature(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():
+    """Test that the server is running and accessible."""
+    logger.info("Starting test_server_availability")
+    try:
+        response = requests.get(BASE_URL, timeout=5)
+        assert response.status_code in [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)}")
+        pytest.fail(f"Server not reachable: {str(e)}")
+
+def test_webhook_dialin_connected(webhook_url, start_url, valid_dialin_payload):
+    """Test webhook handling of dialin.connected event."""
+    logger.info("Starting test_webhook_dialin_connected")
+    headers = {"X-Daily-Signature": generate_signature(valid_dialin_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,
+                webhook_url,
+                json=valid_dialin_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,
+            assert response.status_code == 200, f"Webhook failed: {response.text}"
+            assert response.json() == {"status": "Event received"}
+            mocked_post.assert_called_once_with(
+                start_url,
+                json=valid_dialin_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("Webhook dialin.connected test passed")
         except Exception as e:
-            logger.error(f"Start endpoint valid payload test failed: {str(e)}")
+            logger.error(f"Webhook dialin.connected 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_webhook_dialin_stopped(webhook_url, stopped_payload):
+    """Test webhook handling of dialin.stopped event."""
+    logger.info("Starting test_webhook_dialin_stopped")
+    headers = {"X-Daily-Signature": generate_signature(stopped_payload)}
+    try:
+        response = requests.post(
+            webhook_url,
+            json=stopped_payload,
+            headers=headers,
+            timeout=5
+        )
+        assert response.status_code == 200, f"Webhook failed: {response.text}"
+        assert 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_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
+def test_webhook_invalid_signature(webhook_url, valid_dialin_payload):
+    """Test webhook rejects requests with invalid signature."""
+    logger.info("Starting test_webhook_invalid_signature")
+    headers = {"X-Daily-Signature": "invalid-signature"}
+    try:
+        response = requests.post(
+            webhook_url,
+            json=valid_dialin_payload,
+            headers=headers,
+            timeout=5
+        )
+        assert 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(start_url, valid_dialin_payload):
+    """Test /start endpoint with valid dial-in payload."""
+    logger.info("Starting test_start_endpoint_valid_payload")
+    try:
+        response = requests.post(
+            start_url,
+            json=valid_dialin_payload,
+            timeout=5
+        )
+        assert response.status_code == 200, f"Start endpoint failed: {response.text}"
+        response_json = response.json()
+        assert response_json.get("status") == "Bot started", f"Unexpected status: {response_json}"
+        assert response_json.get("bot_type") == "call_transfer", f"Unexpected bot type: {response_json}"
+        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(start_url, invalid_payload):
+    """Test /start endpoint with invalid payload."""
+    logger.info("Starting test_start_endpoint_invalid_payload")
+    try:
+        response = requests.post(
+            start_url,
+            json=invalid_payload,
+            timeout=5
+        )
+        assert 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():
+    """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)
+        assert 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
+
+def test_start_endpoint_test_mode(start_url):
+    """Test /start endpoint with test mode payload."""
+    logger.info("Starting test_start_endpoint_test_mode")
+    test_payload = {"test": True}
+    try:
+        response = requests.post(
+            start_url,
+            json=test_payload,
+            timeout=5
+        )
+        assert response.status_code == 200, f"Test mode failed: {response.text}"
+        assert response.json() == {"test": True}, f"Unexpected test mode response: {response.json()}"
+        logger.info("Start endpoint test mode test passed")
+    except Exception as e:
+        logger.error(f"Start endpoint test mode test failed: {str(e)}")
+        raise
 
 if __name__ == '__main__':
     logger.info("Starting test suite")
-    unittest.main()
+    pytest.main(["-v", "--log-file=/app/test.log", "--log-file-level=INFO"])
     logger.info("Test suite completed")