101 lines
3.6 KiB
Python
101 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
import urllib.error
|
|
from unittest.mock import patch
|
|
|
|
from tests.conftest import sample_config, sample_events
|
|
from webhook_adapter import create_webhook_callback, send_webhook
|
|
|
|
|
|
class MockHTTPResponse:
|
|
def __init__(self, status: int) -> None:
|
|
self.status = status
|
|
|
|
def getcode(self) -> int:
|
|
return self.status
|
|
|
|
def __enter__(self) -> "MockHTTPResponse":
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> bool:
|
|
return False
|
|
|
|
|
|
class WebhookAdapterTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.events = sample_events()
|
|
self.completed_event = self.events[2]
|
|
self.started_event = self.events[1]
|
|
|
|
@patch("webhook_adapter.urllib.request.urlopen", return_value=MockHTTPResponse(200))
|
|
def test_send_webhook_success(self, mock_urlopen) -> None:
|
|
config = sample_config()
|
|
|
|
result = send_webhook(self.completed_event, config)
|
|
|
|
self.assertTrue(result)
|
|
self.assertEqual(mock_urlopen.call_count, 1)
|
|
request = mock_urlopen.call_args.args[0]
|
|
self.assertEqual(request.full_url, "https://hooks.example.com/macp")
|
|
self.assertEqual(request.get_method(), "POST")
|
|
self.assertEqual(request.headers["Authorization"], "Bearer token-123")
|
|
|
|
@patch("webhook_adapter.urllib.request.urlopen", return_value=MockHTTPResponse(500))
|
|
def test_send_webhook_failure(self, mock_urlopen) -> None:
|
|
result = send_webhook(self.completed_event, sample_config(retry_count=0))
|
|
|
|
self.assertFalse(result)
|
|
self.assertEqual(mock_urlopen.call_count, 1)
|
|
|
|
@patch("webhook_adapter.urllib.request.urlopen", side_effect=TimeoutError("timed out"))
|
|
def test_send_webhook_timeout(self, mock_urlopen) -> None:
|
|
result = send_webhook(self.completed_event, sample_config(retry_count=0))
|
|
|
|
self.assertFalse(result)
|
|
self.assertEqual(mock_urlopen.call_count, 1)
|
|
|
|
@patch("webhook_adapter.time.sleep", return_value=None)
|
|
@patch(
|
|
"webhook_adapter.urllib.request.urlopen",
|
|
side_effect=[MockHTTPResponse(500), MockHTTPResponse(200)],
|
|
)
|
|
def test_send_webhook_retry(self, mock_urlopen, mock_sleep) -> None:
|
|
result = send_webhook(self.completed_event, sample_config(retry_count=1))
|
|
|
|
self.assertTrue(result)
|
|
self.assertEqual(mock_urlopen.call_count, 2)
|
|
mock_sleep.assert_called_once()
|
|
|
|
@patch("webhook_adapter.send_webhook", return_value=True)
|
|
def test_event_filter(self, mock_send_webhook) -> None:
|
|
callback = create_webhook_callback(sample_config(event_filter=["task.completed"]))
|
|
|
|
callback(self.started_event)
|
|
callback(self.completed_event)
|
|
|
|
mock_send_webhook.assert_called_once_with(self.completed_event, sample_config(event_filter=["task.completed"]))
|
|
|
|
@patch("webhook_adapter.send_webhook")
|
|
def test_webhook_disabled(self, mock_send_webhook) -> None:
|
|
callback = create_webhook_callback(sample_config(enabled=False))
|
|
|
|
callback(self.completed_event)
|
|
|
|
mock_send_webhook.assert_not_called()
|
|
|
|
@patch("webhook_adapter.urllib.request.urlopen")
|
|
def test_ssrf_blocked(self, mock_urlopen) -> None:
|
|
for url in ("http://127.0.0.1/webhook", "http://10.1.2.3/webhook"):
|
|
with self.subTest(url=url):
|
|
result = send_webhook(
|
|
self.completed_event,
|
|
sample_config(url=url, auth_token="", retry_count=0),
|
|
)
|
|
self.assertFalse(result)
|
|
mock_urlopen.assert_not_called()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|