Add unit tests for urllib3 warning rewriting

Frankly, this is essentially guaranteed to break the next time we
upgrade urllib3 (esp. as we're due for a major bump to the 2.x series),
but these unit tests serve to ensure the rewriting filters works *now*
and demonstrates in what situations it does work.

Author: ichard26

src/pip/_internal/network/utils.py

@@ -153,7 +153,7 @@ def raise_connection_error(error: requests.ConnectionError, *, timeout: float) -
     raise ConnectionFailedError(url, host, reason)
 
 
-class Urllib3RetryFilter:
+class Urllib3RetryFilter(logging.Filter):
     """A logging filter which attempts to rewrite urllib3's retrying
     warnings to be more readable and less technical.
 

tests/lib/server.py

@@ -3,8 +3,9 @@
 import threading
 from base64 import b64encode
 from contextlib import ExitStack, contextmanager
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from textwrap import dedent
-from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List
+from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator, List, Union
 from unittest.mock import Mock
 
 from werkzeug.serving import BaseWSGIServer, WSGIRequestHandler
@@ -104,7 +105,7 @@ def make_mock_server(**kwargs: Any) -> _MockServer:
 
 
 @contextmanager
-def server_running(server: BaseWSGIServer) -> Iterator[None]:
+def server_running(server: Union[BaseWSGIServer, HTTPServer]) -> Iterator[None]:
     """Context manager for running the provided server in a separate thread."""
     thread = threading.Thread(target=server.serve_forever)
     thread.daemon = True
@@ -232,3 +233,13 @@ def get_requests(self) -> List[Dict[str, str]]:
         # Legacy: replace call[0][0] with call.args[0]
         # when pip drops support for python3.7
         return [call[0][0] for call in self._server.mock.call_args_list]
+
+
+class InstantCloseHTTPHandler(BaseHTTPRequestHandler):
+    """A HTTP request handler that closes the underlying request TCP
+    socket immediately. Used for testing "connection reset by peer" and
+    similar errors.
+    """
+
+    def handle(self) -> None:
+        self.request.close()

tests/unit/test_network_session.py

@@ -1,20 +1,24 @@
 import logging
 import os
+from http.server import HTTPServer
 from pathlib import Path
-from typing import Any, List, Optional
+from typing import Any, Iterator, List, Optional
 from urllib.parse import urlparse
 from urllib.request import getproxies
 
 import pytest
 from pip._vendor import requests
 
 from pip import __version__
+from pip._internal.exceptions import DiagnosticPipError
 from pip._internal.models.link import Link
 from pip._internal.network.session import (
     CI_ENVIRONMENT_VARIABLES,
     PipSession,
     user_agent,
 )
+from pip._internal.utils.logging import VERBOSE
+from tests.lib.server import InstantCloseHTTPHandler, server_running
 
 
 def get_user_agent() -> str:
@@ -281,3 +285,63 @@ def test_proxy(self, proxy: Optional[str]) -> None:
             f"Invalid proxy {proxy} or session.proxies: "
             f"{session.proxies} is not correctly passed to session.request."
         )
+
+
+@pytest.mark.network
+class TestRetryWarningRewriting:
+    @pytest.fixture(autouse=True)
+    def setup_caplog_level(self, caplog: pytest.LogCaptureFixture) -> Iterator[None]:
+        with caplog.at_level(logging.WARNING):
+            yield
+
+    @pytest.mark.parametrize(
+        "url, expected_message",
+        [
+            (
+                "https://404.example.com",
+                "failed to connect to 404.example.com via HTTPS",
+            ),
+            ("http://404.example.com", "failed to connect to 404.example.com via HTTP"),
+            ("https://expired.badssl.com", "SSL verification failed"),
+        ],
+    )
+    def test_simple_urls(
+        self, caplog: pytest.LogCaptureFixture, url: str, expected_message: str
+    ) -> None:
+        with PipSession(retries=1) as session:
+            with pytest.raises(DiagnosticPipError):
+                session.get(url)
+        assert caplog.messages == [f"{expected_message}, retrying 1 last time"]
+
+    def test_timeout(self, caplog: pytest.LogCaptureFixture) -> None:
+        with PipSession(retries=1) as session:
+            with pytest.raises(DiagnosticPipError):
+                session.get("https://httpstat.us/200?sleep=400", timeout=0.2)
+        assert caplog.messages == [
+            "server didn't respond within 0.2 seconds, retrying 1 last time"
+        ]
+
+    def test_connection_aborted(self, caplog: pytest.LogCaptureFixture) -> None:
+        with HTTPServer(("localhost", 0), InstantCloseHTTPHandler) as server:
+            with server_running(server), PipSession(retries=1) as session:
+                with pytest.raises(DiagnosticPipError):
+                    session.get(f"http://{server.server_name}:{server.server_port}/")
+            assert caplog.messages == [
+                "the connection was closed unexpectedly, retrying 1 last time"
+            ]
+
+    def test_proxy(self, caplog: pytest.LogCaptureFixture) -> None:
+        with PipSession(retries=1) as session:
+            session.proxies = {"https": "https://404.example.com"}
+            with pytest.raises(DiagnosticPipError):
+                session.get("https://pypi.org")
+        assert caplog.messages == ["failed to connect to proxy, retrying 1 last time"]
+
+    def test_verbose(self, caplog: pytest.LogCaptureFixture) -> None:
+        caplog.set_level(VERBOSE)
+        with PipSession(retries=1) as session:
+            with pytest.raises(DiagnosticPipError):
+                session.get("https://404.example.org")
+        warnings = [r.message for r in caplog.records if r.levelno == logging.WARNING]
+        assert len(warnings) == 1
+        assert not warnings[0].endswith("retrying 1 last time")