Update test_webhook_and_start.py

test_webhook_and_start.py

@@ -1,4 +1,4 @@
-import pytest
+import unittest
 import json
 import hmac
 import hashlib
@@ -6,10 +6,9 @@ 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,
@@ -21,193 +20,161 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
-# Base URL of your Hugging Face Space or local server
+# 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
 
-@pytest.fixture
-def webhook_url():
-    """Fixture for webhook URL."""
-    return f"{BASE_URL}/webhook"
-
-@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"}'
-        
+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(
-                webhook_url,
-                json=valid_dialin_payload,
+                self.webhook_url,
+                json=self.stopped_payload,
                 headers=headers,
                 timeout=5
             )
-            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,
+            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
             )
-            logger.info("Webhook dialin.connected test passed")
+            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 dialin.connected test failed: {str(e)}")
+            logger.error(f"Webhook invalid signature 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_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_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_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_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_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
+    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")
-    pytest.main(["-v", "--log-file=/app/test.log", "--log-file-level=INFO"])
+    unittest.main()
     logger.info("Test suite completed")