test_webhook_and_start.py

import unittest
import json
import hmac
import hashlib
import requests
from unittest.mock import patch
import logging
import os

# 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://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

    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
            )
            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,
                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"}
        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)."""
        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")
    unittest.main()
    logger.info("Test suite completed")