diff --git a/tests/test_alerter.py b/tests/test_alerter.py index b64c772..b0646bd 100644 --- a/tests/test_alerter.py +++ b/tests/test_alerter.py @@ -1,5 +1,6 @@ """Tests for the webhook alerter module.""" +import json from datetime import UTC, datetime from unittest.mock import MagicMock, Mock, patch @@ -217,29 +218,30 @@ def test_build_payload_up_event(self, alerter: Alerter, check_result_up: CheckRe assert payload["status"]["success"] is True assert payload["previous_status"] == "down" - @patch("webstatuspi.alerter.requests.post") - def test_send_webhook_success(self, mock_post: Mock, alerter: Alerter, check_result_down: CheckResult) -> None: + @patch("webstatuspi.alerter.urllib.request.urlopen") + def test_send_webhook_success(self, mock_urlopen: Mock, alerter: Alerter, check_result_down: CheckResult) -> None: """Test successful webhook delivery.""" - mock_response = MagicMock() - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response + mock_cm = MagicMock() + mock_urlopen.return_value.__enter__ = Mock(return_value=mock_cm) + mock_urlopen.return_value.__exit__ = Mock(return_value=False) webhook = alerter._config.webhooks[0] alerter._send_webhook(webhook, check_result_down) - mock_post.assert_called_once() - args, kwargs = mock_post.call_args - assert args[0] == "https://example.com/webhook" - assert kwargs["timeout"] == 10 - assert isinstance(kwargs["json"], dict) - assert "event" in kwargs["json"] + mock_urlopen.assert_called_once() + call_args = mock_urlopen.call_args + req = call_args[0][0] + assert req.full_url == "https://example.com/webhook" + assert call_args[1]["timeout"] == 10 + payload = json.loads(req.data) + assert "event" in payload - @patch("webstatuspi.alerter.requests.post") - def test_send_webhook_retry_on_failure(self, mock_post: Mock, check_result_down: CheckResult) -> None: + @patch("webstatuspi.alerter.urllib.request.urlopen") + def test_send_webhook_retry_on_failure(self, mock_urlopen: Mock, check_result_down: CheckResult) -> None: """Test that webhook retries on failure.""" - import requests + import urllib.error - mock_post.side_effect = requests.RequestException("Connection error") + mock_urlopen.side_effect = urllib.error.URLError("Connection error") webhook = WebhookConfig( url="https://example.com/webhook", @@ -253,20 +255,21 @@ def test_send_webhook_retry_on_failure(self, mock_post: Mock, check_result_down: alerter._send_webhook(webhook, check_result_down) # Should attempt 3 times (initial + 2 retries) - assert mock_post.call_count == 3 + assert mock_urlopen.call_count == 3 - @patch("webstatuspi.alerter.requests.post") - def test_send_webhook_success_after_retry(self, mock_post: Mock, check_result_down: CheckResult) -> None: + @patch("webstatuspi.alerter.urllib.request.urlopen") + def test_send_webhook_success_after_retry(self, mock_urlopen: Mock, check_result_down: CheckResult) -> None: """Test successful delivery after retry.""" - mock_response = MagicMock() - mock_response.raise_for_status.return_value = None + import urllib.error - import requests + mock_cm = MagicMock() + mock_cm.__enter__ = Mock(return_value=mock_cm) + mock_cm.__exit__ = Mock(return_value=False) # Fail first, succeed second - mock_post.side_effect = [ - requests.RequestException("Connection error"), - mock_response, + mock_urlopen.side_effect = [ + urllib.error.URLError("Connection error"), + mock_cm, ] webhook = WebhookConfig( @@ -281,26 +284,26 @@ def test_send_webhook_success_after_retry(self, mock_post: Mock, check_result_do alerter._send_webhook(webhook, check_result_down) # Should succeed after retry - assert mock_post.call_count == 2 + assert mock_urlopen.call_count == 2 - @patch("webstatuspi.alerter.requests.post") - def test_test_webhooks_all_success(self, mock_post: Mock, alerter: Alerter) -> None: + @patch("webstatuspi.alerter.urllib.request.urlopen") + def test_test_webhooks_all_success(self, mock_urlopen: Mock, alerter: Alerter) -> None: """Test successful webhook testing.""" - mock_response = MagicMock() - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response + mock_cm = MagicMock() + mock_urlopen.return_value.__enter__ = Mock(return_value=mock_cm) + mock_urlopen.return_value.__exit__ = Mock(return_value=False) results = alerter.test_webhooks() assert results["https://example.com/webhook"] is True - mock_post.assert_called_once() + mock_urlopen.assert_called_once() - @patch("webstatuspi.alerter.requests.post") - def test_test_webhooks_failure(self, mock_post: Mock, alerter: Alerter) -> None: + @patch("webstatuspi.alerter.urllib.request.urlopen") + def test_test_webhooks_failure(self, mock_urlopen: Mock, alerter: Alerter) -> None: """Test failed webhook testing.""" - import requests + import urllib.error - mock_post.side_effect = requests.RequestException("Connection error") + mock_urlopen.side_effect = urllib.error.URLError("Connection error") results = alerter.test_webhooks() @@ -314,10 +317,10 @@ def test_test_webhooks_disabled_webhook(self) -> None: ) alerter = Alerter(AlertsConfig(webhooks=[webhook])) - with patch("webstatuspi.alerter.requests.post") as mock_post: + with patch("webstatuspi.alerter.urllib.request.urlopen") as mock_urlopen: results = alerter.test_webhooks() assert results["https://example.com/webhook"] is False - mock_post.assert_not_called() + mock_urlopen.assert_not_called() def test_multiple_webhooks(self) -> None: """Test alerter with multiple webhooks.""" @@ -476,25 +479,26 @@ def test_latency_counter_reset_on_normal(self, alerter: Alerter, url_config_with assert alerter._state_tracker.consecutive_slow.get("test_url", 0) == 0 mock_send.assert_not_called() # No alert since we never reached threshold - @patch("webstatuspi.alerter.requests.post") + @patch("webstatuspi.alerter.urllib.request.urlopen") def test_send_latency_webhook_success( - self, mock_post: Mock, alerter: Alerter, url_config_with_threshold: UrlConfig + self, mock_urlopen: Mock, alerter: Alerter, url_config_with_threshold: UrlConfig ) -> None: """Test successful latency webhook delivery.""" - mock_response = MagicMock() - mock_response.raise_for_status.return_value = None - mock_post.return_value = mock_response + mock_cm = MagicMock() + mock_urlopen.return_value.__enter__ = Mock(return_value=mock_cm) + mock_urlopen.return_value.__exit__ = Mock(return_value=False) # Trigger alert for _ in range(3): alerter.check_latency_alert(url_config_with_threshold, 1500) - mock_post.assert_called_once() - args, kwargs = mock_post.call_args - assert args[0] == "https://example.com/webhook" - assert kwargs["timeout"] == 10 + mock_urlopen.assert_called_once() + call_args = mock_urlopen.call_args + req = call_args[0][0] + assert req.full_url == "https://example.com/webhook" + assert call_args[1]["timeout"] == 10 - payload = kwargs["json"] + payload = json.loads(req.data) assert payload["event"] == "latency_high" assert payload["url"]["name"] == "test_url" assert payload["url"]["url"] == "https://example.com" diff --git a/webstatuspi/alerter.py b/webstatuspi/alerter.py index 4724c0c..5a648d0 100644 --- a/webstatuspi/alerter.py +++ b/webstatuspi/alerter.py @@ -1,16 +1,17 @@ """Webhook and email alert system with state tracking and cooldown management.""" +import json import logging import smtplib import threading import time +import urllib.error +import urllib.request from dataclasses import dataclass, field from datetime import UTC, datetime from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText -import requests - from webstatuspi.config import AlertsConfig, SmtpConfig, UrlConfig, WebhookConfig from webstatuspi.models import CheckResult from webstatuspi.security import SSRFError, validate_url_for_ssrf @@ -191,15 +192,18 @@ def _send_webhook(self, webhook: WebhookConfig, result: CheckResult) -> None: payload = self._build_payload(result) retry_count = 0 + data = json.dumps(payload).encode() while retry_count <= self._max_retries: try: - response = requests.post( + req = urllib.request.Request( webhook.url, - json=payload, - timeout=10, + data=data, + headers={"Content-Type": "application/json"}, + method="POST", ) - response.raise_for_status() + with urllib.request.urlopen(req, timeout=10): + pass logger.info( "Webhook sent successfully for %s to %s", @@ -209,7 +213,7 @@ def _send_webhook(self, webhook: WebhookConfig, result: CheckResult) -> None: self._state_tracker.last_alert_time[result.url_name] = time.time() return - except requests.RequestException as e: + except (urllib.error.URLError, OSError) as e: retry_count += 1 if retry_count <= self._max_retries: delay = self._retry_delay * (2 ** (retry_count - 1)) @@ -311,14 +315,17 @@ def _send_latency_webhook( } retry_count = 0 + latency_data = json.dumps(payload).encode() while retry_count <= self._max_retries: try: - response = requests.post( + req = urllib.request.Request( webhook.url, - json=payload, - timeout=10, + data=latency_data, + headers={"Content-Type": "application/json"}, + method="POST", ) - response.raise_for_status() + with urllib.request.urlopen(req, timeout=10): + pass logger.info( "Latency webhook sent successfully for %s (%s) to %s", @@ -329,7 +336,7 @@ def _send_latency_webhook( self._state_tracker.last_alert_time[url_config.name] = time.time() return - except requests.RequestException as e: + except (urllib.error.URLError, OSError) as e: retry_count += 1 if retry_count <= self._max_retries: delay = self._retry_delay * (2 ** (retry_count - 1)) @@ -392,16 +399,19 @@ def test_webhooks(self) -> dict[str, bool]: } try: - response = requests.post( + test_data = json.dumps(test_payload).encode() + req = urllib.request.Request( webhook.url, - json=test_payload, - timeout=10, + data=test_data, + headers={"Content-Type": "application/json"}, + method="POST", ) - response.raise_for_status() + with urllib.request.urlopen(req, timeout=10): + pass results[webhook.url] = True logger.info("Test webhook sent successfully to %s", webhook.url) - except requests.RequestException as e: + except (urllib.error.URLError, OSError) as e: results[webhook.url] = False logger.error("Test webhook failed for %s: %s", webhook.url, e)