Replace captured_output() and get_url_scheme() with stdlib alternativ…

…es (#12639)

Co-authored-by: Tzu-ping Chung <[email protected]>

news/c678d9e3-4844-4298-a46c-80768b38f652.trivial.rst

@@ -0,0 +1 @@
+Replace ``captured_output()`` and ``get_url_scheme()`` with stdlib alternatives.

src/pip/_internal/operations/install/wheel.py

@@ -51,7 +51,7 @@
 from pip._internal.models.direct_url import DIRECT_URL_METADATA_NAME, DirectUrl
 from pip._internal.models.scheme import SCHEME_KEYS, Scheme
 from pip._internal.utils.filesystem import adjacent_tmp_file, replace
-from pip._internal.utils.misc import captured_stdout, ensure_dir, hash_file, partition
+from pip._internal.utils.misc import StreamWrapper, ensure_dir, hash_file, partition
 from pip._internal.utils.unpacking import (
     current_umask,
     is_within_directory,
@@ -603,7 +603,9 @@ def pyc_output_path(path: str) -> str:
 
     # Compile all of the pyc files for the installed files
     if pycompile:
-        with captured_stdout() as stdout:
+        with contextlib.redirect_stdout(
+            StreamWrapper.from_stream(sys.stdout)
+        ) as stdout:
             with warnings.catch_warnings():
                 warnings.filterwarnings("ignore")
                 for path in pyc_source_file_paths():

src/pip/_internal/req/req_file.py

@@ -26,7 +26,6 @@
 from pip._internal.exceptions import InstallationError, RequirementsFileParseError
 from pip._internal.models.search_scope import SearchScope
 from pip._internal.utils.encoding import auto_decode
-from pip._internal.utils.urls import get_url_scheme
 
 if TYPE_CHECKING:
     from pip._internal.index.package_finder import PackageFinder
@@ -533,8 +532,7 @@ def get_file_content(url: str, session: "PipSession") -> Tuple[str, str]:
     :param url:         File path or url.
     :param session:     PipSession instance.
     """
-    scheme = get_url_scheme(url)
-
+    scheme = urllib.parse.urlsplit(url).scheme
     # Pip has special support for file:// URLs (LocalFSAdapter).
     if scheme in ["http", "https", "file"]:
         # Delay importing heavy network modules until absolutely necessary.

src/pip/_internal/utils/misc.py

@@ -1,4 +1,3 @@
-import contextlib
 import errno
 import getpass
 import hashlib
@@ -21,7 +20,6 @@
     Any,
     BinaryIO,
     Callable,
-    ContextManager,
     Dict,
     Generator,
     Iterable,
@@ -57,7 +55,6 @@
     "normalize_path",
     "renames",
     "get_prog",
-    "captured_stdout",
     "ensure_dir",
     "remove_auth_from_url",
     "check_externally_managed",
@@ -400,40 +397,6 @@ def encoding(self) -> str:  # type: ignore
         return self.orig_stream.encoding
 
 
-@contextlib.contextmanager
-def captured_output(stream_name: str) -> Generator[StreamWrapper, None, None]:
-    """Return a context manager used by captured_stdout/stdin/stderr
-    that temporarily replaces the sys stream *stream_name* with a StringIO.
-
-    Taken from Lib/support/__init__.py in the CPython repo.
-    """
-    orig_stdout = getattr(sys, stream_name)
-    setattr(sys, stream_name, StreamWrapper.from_stream(orig_stdout))
-    try:
-        yield getattr(sys, stream_name)
-    finally:
-        setattr(sys, stream_name, orig_stdout)
-
-
-def captured_stdout() -> ContextManager[StreamWrapper]:
-    """Capture the output of sys.stdout:
-
-       with captured_stdout() as stdout:
-           print('hello')
-       self.assertEqual(stdout.getvalue(), 'hello\n')
-
-    Taken from Lib/support/__init__.py in the CPython repo.
-    """
-    return captured_output("stdout")
-
-
-def captured_stderr() -> ContextManager[StreamWrapper]:
-    """
-    See captured_stdout().
-    """
-    return captured_output("stderr")
-
-
 # Simulates an enum
 def enum(*sequential: Any, **named: Any) -> Type[Any]:
     enums = dict(zip(sequential, range(len(sequential))), **named)

src/pip/_internal/utils/urls.py

@@ -2,17 +2,10 @@
 import string
 import urllib.parse
 import urllib.request
-from typing import Optional
 
 from .compat import WINDOWS
 
 
-def get_url_scheme(url: str) -> Optional[str]:
-    if ":" not in url:
-        return None
-    return url.split(":", 1)[0].lower()
-
-
 def path_to_url(path: str) -> str:
     """
     Convert a path to a file: URL.  The path will be made absolute and have

src/pip/_internal/vcs/versioncontrol.py

@@ -38,7 +38,6 @@
     format_command_args,
     make_command,
 )
-from pip._internal.utils.urls import get_url_scheme
 
 __all__ = ["vcs"]
 
@@ -52,8 +51,8 @@ def is_url(name: str) -> bool:
     """
     Return true if the name looks like a URL.
     """
-    scheme = get_url_scheme(name)
-    if scheme is None:
+    scheme = urllib.parse.urlsplit(name).scheme
+    if not scheme:
         return False
     return scheme in ["http", "https", "file", "ftp"] + vcs.all_schemes
 

tests/unit/test_logging.py

@@ -1,5 +1,7 @@
 import logging
 import time
+from contextlib import redirect_stderr, redirect_stdout
+from io import StringIO
 from threading import Thread
 from unittest.mock import patch
 
@@ -11,7 +13,6 @@
     RichPipStreamHandler,
     indent_log,
 )
-from pip._internal.utils.misc import captured_stderr, captured_stdout
 
 logger = logging.getLogger(__name__)
 
@@ -140,7 +141,7 @@ def test_broken_pipe_in_stderr_flush(self) -> None:
         """
         record = self._make_log_record()
 
-        with captured_stderr() as stderr:
+        with redirect_stderr(StringIO()) as stderr:
             handler = RichPipStreamHandler(stream=stderr, no_color=True)
             with patch("sys.stderr.flush") as mock_flush:
                 mock_flush.side_effect = BrokenPipeError()
@@ -163,7 +164,7 @@ def test_broken_pipe_in_stdout_write(self) -> None:
         """
         record = self._make_log_record()
 
-        with captured_stdout() as stdout:
+        with redirect_stdout(StringIO()) as stdout:
             handler = RichPipStreamHandler(stream=stdout, no_color=True)
             with patch("sys.stdout.write") as mock_write:
                 mock_write.side_effect = BrokenPipeError()
@@ -178,7 +179,7 @@ def test_broken_pipe_in_stdout_flush(self) -> None:
         """
         record = self._make_log_record()
 
-        with captured_stdout() as stdout:
+        with redirect_stdout(StringIO()) as stdout:
             handler = RichPipStreamHandler(stream=stdout, no_color=True)
             with patch("sys.stdout.flush") as mock_flush:
                 mock_flush.side_effect = BrokenPipeError()

tests/unit/test_urls.py

@@ -1,24 +1,10 @@
 import os
 import sys
 import urllib.request
-from typing import Optional
 
 import pytest
 
-from pip._internal.utils.urls import get_url_scheme, path_to_url, url_to_path
-
-
-@pytest.mark.parametrize(
-    "url,expected",
-    [
-        ("http://localhost:8080/", "http"),
-        ("file:c:/path/to/file", "file"),
-        ("file:/dev/null", "file"),
-        ("", None),
-    ],
-)
-def test_get_url_scheme(url: str, expected: Optional[str]) -> None:
-    assert get_url_scheme(url) == expected
+from pip._internal.utils.urls import path_to_url, url_to_path
 
 
 @pytest.mark.skipif("sys.platform == 'win32'")