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