fix: report a redirected webhook URL instead of silently posting nothing

pr_split/cli.py

@@ -4,6 +4,7 @@
 import shutil
 import tempfile
 import time
+import urllib.error
 import urllib.parse
 import urllib.request
 from collections.abc import Callable, Generator
@@ -2086,11 +2087,29 @@ def _validate_webhook_url(value: str | None) -> str | None:
     return value
 
 
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    """Refuse to follow redirects.
+
+    urllib re-issues a redirected POST as a body-less GET, so a moved
+    webhook URL would receive an empty request while the tool reports the
+    notification as sent. Surface the redirect instead so the user can
+    update the URL.
+    """
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        raise urllib.error.HTTPError(
+            req.full_url, code, f"redirected to {newurl}; update the webhook URL", headers, fp
+        )
+
+
+_webhook_opener = urllib.request.build_opener(_NoRedirect)
+
+
 def _send_webhook(url: str, payload: dict[str, object]) -> None:
     try:
         data = json_mod.dumps(payload).encode("utf-8")
         req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
-        with urllib.request.urlopen(req, timeout=10) as resp:
+        with _webhook_opener.open(req, timeout=10) as resp:
             resp.read()
         logger.info(f"Webhook notification sent to {url}")
     except Exception as exc:

tests/test_cli_coverage.py

@@ -345,7 +345,7 @@ def test_template_os_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 # _send_webhook
 # ---------------------------------------------------------------------------
 class TestSendWebhook:
-    @patch("pr_split.cli.urllib.request.urlopen")
+    @patch("pr_split.cli._webhook_opener.open")
     def test_successful_webhook(self, mock_urlopen: MagicMock) -> None:
         mock_resp = MagicMock()
         mock_resp.read.return_value = b""
@@ -356,7 +356,7 @@ def test_successful_webhook(self, mock_urlopen: MagicMock) -> None:
         _send_webhook("https://example.com/hook", {"event": "test"})
         mock_urlopen.assert_called_once()
 
-    @patch("pr_split.cli.urllib.request.urlopen", side_effect=Exception("timeout"))
+    @patch("pr_split.cli._webhook_opener.open", side_effect=Exception("timeout"))
     def test_failed_webhook_logs_warning(self, mock_urlopen: MagicMock) -> None:
         # Should not raise
         _send_webhook("https://example.com/hook", {"event": "test"})
@@ -1980,3 +1980,84 @@ def test_http_urls_pass(self, mock_pe: MagicMock) -> None:
         result = runner.invoke(app, ["merge", "--notify", "https://hooks.example/abc"])
         assert result.exit_code == 0
         mock_pe.assert_called_once()
+
+
+class TestWebhookRedirects:
+    def _serve(self, handler_cls: type) -> tuple[object, int]:
+        import threading
+        from http.server import HTTPServer
+
+        server = HTTPServer(("127.0.0.1", 0), handler_cls)
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        return server, server.server_address[1]
+
+    def test_redirect_is_reported_instead_of_a_body_less_get(self) -> None:
+        from http.server import BaseHTTPRequestHandler
+
+        from pr_split.cli import _send_webhook
+
+        received: list[tuple[str, int]] = []
+
+        class Target(BaseHTTPRequestHandler):
+            def do_GET(self) -> None:
+                received.append(("GET", int(self.headers.get("Content-Length") or 0)))
+                self.send_response(200)
+                self.end_headers()
+
+            def do_POST(self) -> None:
+                received.append(("POST", int(self.headers.get("Content-Length") or 0)))
+                self.send_response(200)
+                self.end_headers()
+
+            def log_message(self, *args: object) -> None:
+                pass
+
+        target, target_port = self._serve(Target)
+
+        class Mover(BaseHTTPRequestHandler):
+            def do_POST(self) -> None:
+                self.send_response(301)
+                self.send_header("Location", f"http://127.0.0.1:{target_port}/hook")
+                self.end_headers()
+
+            def log_message(self, *args: object) -> None:
+                pass
+
+        mover, mover_port = self._serve(Mover)
+        try:
+            with patch("pr_split.cli.logger") as mock_logger:
+                _send_webhook(f"http://127.0.0.1:{mover_port}/old", {"event": "merge_complete"})
+        finally:
+            mover.shutdown()  # type: ignore[attr-defined]
+            target.shutdown()  # type: ignore[attr-defined]
+
+        assert received == []
+        warning = str(mock_logger.warning.call_args)
+        assert "redirected to" in warning and f"127.0.0.1:{target_port}/hook" in warning
+        mock_logger.info.assert_not_called()
+
+    def test_direct_post_still_delivers_the_payload(self) -> None:
+        from http.server import BaseHTTPRequestHandler
+
+        from pr_split.cli import _send_webhook
+
+        bodies: list[bytes] = []
+
+        class Sink(BaseHTTPRequestHandler):
+            def do_POST(self) -> None:
+                bodies.append(self.rfile.read(int(self.headers["Content-Length"])))
+                self.send_response(200)
+                self.end_headers()
+
+            def log_message(self, *args: object) -> None:
+                pass
+
+        sink, port = self._serve(Sink)
+        try:
+            with patch("pr_split.cli.logger") as mock_logger:
+                _send_webhook(f"http://127.0.0.1:{port}/hook", {"event": "merge_complete"})
+        finally:
+            sink.shutdown()  # type: ignore[attr-defined]
+
+        assert bodies == [b'{"event": "merge_complete"}']
+        mock_logger.warning.assert_not_called()