diff --git a/pr_split/cli.py b/pr_split/cli.py index 848dfb4..909db10 100644 --- a/pr_split/cli.py +++ b/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: diff --git a/tests/test_cli_coverage.py b/tests/test_cli_coverage.py index 808d387..1e2263a 100644 --- a/tests/test_cli_coverage.py +++ b/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()