test_webhook_and_start.py

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()