File size: 7,907 Bytes
47cee22 848219e ca10ff4 47cee22 848219e ca10ff4 47cee22 ca10ff4 848219e 47cee22 848219e 47cee22 ca10ff4 47cee22 ca10ff4 47cee22 ca10ff4 47cee22 ca10ff4 47cee22 ca10ff4 848219e 47cee22 848219e 47cee22 848219e ca10ff4 47cee22 ca10ff4 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 |
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") |