refactor(alerter): replace requests with urllib.request

- Remove requests dependency from alerter.py (stdlib-first principle)
- Replace requests.post() with urllib.request.urlopen() in:
  - _send_webhook()
  - _send_latency_webhook()
  - test_webhooks()
- Replace requests.RequestException with (urllib.error.URLError, OSError)
- Update tests to mock urllib.request.urlopen instead of requests.post
- Saves ~2-3MB RAM on Pi 1B+ (identified in BAD_PRACTICES_ANALYSIS.md)

All 53 tests passing.

Co-authored-by: Hustle <hustle@hustle.local>

Author: jmlweb

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"

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)