test_webhook_and_start.py

import pytest
import json
import hmac
import hashlib
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 = "/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 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"

@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(
                webhook_url,
                json=valid_dialin_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,
                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(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_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")
    pytest.main(["-v", "--log-file=/app/test.log", "--log-file-level=INFO"])
    logger.info("Test suite completed")