Merge pull request #10388 from jdufresne/mypy-tests

Run mypy on the tests directory

Author: pradyunsg

.pre-commit-config.yaml

@@ -42,12 +42,13 @@ repos:
   rev: v0.910
   hooks:
   - id: mypy
-    exclude: tests
+    exclude: tests/data
     args: ["--pretty", "--show-error-codes"]
     additional_dependencies: [
         'keyring==23.0.1',
         'nox==2021.6.12',
         'types-docutils==0.1.8',
+        'types-setuptools==57.0.2',
         'types-six==0.1.9',
     ]
 

news/f231bb92-a022-4b48-b7fd-c83edefb4353.trivial.rst

@@ -51,6 +51,10 @@ follow_imports = skip
 [mypy-pip._vendor.requests.*]
 follow_imports = skip
 
+[mypy-tests.*]
+# TODO: The following option should be removed at some point in the future.
+allow_untyped_defs = True
+
 [tool:pytest]
 addopts = --ignore src/pip/_vendor --ignore tests/tests_cache -r aR --color=yes
 markers =

setup.cfg

@@ -8,7 +8,7 @@
 import sys
 import time
 from contextlib import ExitStack, contextmanager
-from typing import Dict, Iterable
+from typing import TYPE_CHECKING, Dict, Iterable, List
 from unittest.mock import patch
 
 import pytest
@@ -21,11 +21,14 @@
 from tests.lib.certs import make_tls_cert, serialize_cert, serialize_key
 from tests.lib.path import Path
 from tests.lib.server import MockServer as _MockServer
-from tests.lib.server import Responder, make_mock_server, server_running
+from tests.lib.server import make_mock_server, server_running
 from tests.lib.venv import VirtualEnvironment
 
 from .lib.compat import nullcontext
 
+if TYPE_CHECKING:
+    from wsgi import WSGIApplication
+
 
 def pytest_addoption(parser):
     parser.addoption(
@@ -521,7 +524,7 @@ def port(self):
     def host(self):
         return self._server.host
 
-    def set_responses(self, responses: Iterable[Responder]) -> None:
+    def set_responses(self, responses: Iterable["WSGIApplication"]) -> None:
         assert not self._running, "responses cannot be set on running server"
         self._server.mock.side_effect = responses
 
@@ -542,7 +545,7 @@ def stop(self) -> None:
         assert self._running, "idle server cannot be stopped"
         self.context.close()
 
-    def get_requests(self) -> Dict[str, str]:
+    def get_requests(self) -> List[Dict[str, str]]:
         """Get environ for each received request."""
         assert not self._running, "cannot get mock from running server"
         # Legacy: replace call[0][0] with call.args[0]

tests/conftest.py

@@ -1,8 +1,5 @@
 import pytest
-
-# The vendored `tomli` package is not used here because it doesn't
-# have write capability
-import toml
+import tomli_w
 
 from pip._internal.build_env import BuildEnvironment
 from pip._internal.req import InstallRequirement
@@ -18,7 +15,7 @@ def make_project(tmpdir, requires=None, backend=None, backend_path=None):
         buildsys["build-backend"] = backend
     if backend_path:
         buildsys["backend-path"] = backend_path
-    data = toml.dumps({"build-system": buildsys})
+    data = tomli_w.dumps({"build-system": buildsys})
     project_dir.joinpath("pyproject.toml").write_text(data)
     return project_dir
 
@@ -189,7 +186,7 @@ def make_pyproject_with_setup(tmpdir, build_system=True, set_backend=True):
         if set_backend:
             buildsys["build-backend"] = "setuptools.build_meta"
             expect_script_dir_on_path = False
-        project_data = toml.dumps({"build-system": buildsys})
+        project_data = tomli_w.dumps({"build-system": buildsys})
     else:
         project_data = ""
 

tests/functional/test_pep517.py

@@ -54,6 +54,7 @@ def local_checkout(
         repo_url_path = os.path.join(repo_url_path, "trunk")
     else:
         vcs_backend = vcs.get_backend(vcs_name)
+        assert vcs_backend is not None
         vcs_backend.obtain(repo_url_path, url=hide_url(remote_repo))
 
     return "{}+{}".format(vcs_name, path_to_url(repo_url_path))

tests/lib/local_repos.py

@@ -5,23 +5,18 @@
 from base64 import b64encode
 from contextlib import contextmanager
 from textwrap import dedent
-from types import TracebackType
-from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple, Type
+from typing import TYPE_CHECKING, Any, Callable, Dict, Iterable, Iterator
 from unittest.mock import Mock
 
 from werkzeug.serving import BaseWSGIServer, WSGIRequestHandler
 from werkzeug.serving import make_server as _make_server
 
 from .compat import nullcontext
 
-Environ = Dict[str, str]
-Status = str
-Headers = Iterable[Tuple[str, str]]
-ExcInfo = Tuple[Type[BaseException], BaseException, TracebackType]
-Write = Callable[[bytes], None]
-StartResponse = Callable[[Status, Headers, Optional[ExcInfo]], Write]
-Body = List[bytes]
-Responder = Callable[[Environ, StartResponse], Body]
+if TYPE_CHECKING:
+    from wsgi import StartResponse, WSGIApplication, WSGIEnvironment
+
+Body = Iterable[bytes]
 
 
 class MockServer(BaseWSGIServer):
@@ -78,13 +73,13 @@ def make_environ(self):
 
 
 def _mock_wsgi_adapter(
-    mock: Callable[[Environ, StartResponse], Responder]
-) -> Responder:
+    mock: Callable[["WSGIEnvironment", "StartResponse"], "WSGIApplication"]
+) -> "WSGIApplication":
     """Uses a mock to record function arguments and provide
     the actual function that should respond.
     """
 
-    def adapter(environ: Environ, start_response: StartResponse) -> Body:
+    def adapter(environ: "WSGIEnvironment", start_response: "StartResponse") -> Body:
         try:
             responder = mock(environ, start_response)
         except StopIteration:
@@ -134,7 +129,7 @@ def make_mock_server(**kwargs: Any) -> MockServer:
 
 
 @contextmanager
-def server_running(server: BaseWSGIServer) -> None:
+def server_running(server: BaseWSGIServer) -> Iterator[None]:
     """Context manager for running the provided server in a separate thread."""
     thread = threading.Thread(target=server.serve_forever)
     thread.daemon = True
@@ -150,8 +145,8 @@ def server_running(server: BaseWSGIServer) -> None:
 # Helper functions for making responses in a declarative way.
 
 
-def text_html_response(text: str) -> Responder:
-    def responder(environ: Environ, start_response: StartResponse) -> Body:
+def text_html_response(text: str) -> "WSGIApplication":
+    def responder(environ: "WSGIEnvironment", start_response: "StartResponse") -> Body:
         start_response(
             "200 OK",
             [
@@ -180,24 +175,24 @@ def html5_page(text: str) -> str:
     )
 
 
-def index_page(spec: Dict[str, str]) -> Responder:
+def index_page(spec: Dict[str, str]) -> "WSGIApplication":
     def link(name, value):
         return '<a href="{}">{}</a>'.format(value, name)
 
     links = "".join(link(*kv) for kv in spec.items())
     return text_html_response(html5_page(links))
 
 
-def package_page(spec: Dict[str, str]) -> Responder:
+def package_page(spec: Dict[str, str]) -> "WSGIApplication":
     def link(name, value):
         return '<a href="{}">{}</a>'.format(value, name)
 
     links = "".join(link(*kv) for kv in spec.items())
     return text_html_response(html5_page(links))
 
 
-def file_response(path: str) -> Responder:
-    def responder(environ: Environ, start_response: StartResponse) -> Body:
+def file_response(path: str) -> "WSGIApplication":
+    def responder(environ: "WSGIEnvironment", start_response: "StartResponse") -> Body:
         size = os.stat(path).st_size
         start_response(
             "200 OK",
@@ -213,10 +208,10 @@ def responder(environ: Environ, start_response: StartResponse) -> Body:
     return responder
 
 
-def authorization_response(path: str) -> Responder:
+def authorization_response(path: str) -> "WSGIApplication":
     correct_auth = "Basic " + b64encode(b"USERNAME:PASSWORD").decode("ascii")
 
-    def responder(environ: Environ, start_response: StartResponse) -> Body:
+    def responder(environ: "WSGIEnvironment", start_response: "StartResponse") -> Body:
 
         if environ.get("HTTP_AUTHORIZATION") == correct_auth:
             size = os.stat(path).st_size

tests/lib/server.py

@@ -12,7 +12,6 @@
 from io import BytesIO, StringIO
 from typing import (
     AnyStr,
-    Callable,
     Dict,
     Iterable,
     List,
@@ -28,9 +27,6 @@
 
 from tests.lib.path import Path
 
-# path, digest, size
-RecordLike = Tuple[str, str, str]
-RecordCallback = Callable[[List["Record"]], Union[str, bytes, List[RecordLike]]]
 # As would be used in metadata
 HeaderValue = Union[str, List[str]]
 
@@ -82,7 +78,7 @@ def make_metadata_file(
     value: Defaulted[Optional[AnyStr]],
     updates: Defaulted[Dict[str, HeaderValue]],
     body: Defaulted[AnyStr],
-) -> File:
+) -> Optional[File]:
     if value is None:
         return None
 
@@ -199,9 +195,8 @@ def digest(contents: bytes) -> str:
 def record_file_maker_wrapper(
     name: str,
     version: str,
-    files: List[File],
+    files: Iterable[File],
     record: Defaulted[Optional[AnyStr]],
-    record_callback: Defaulted[RecordCallback],
 ) -> Iterable[File]:
     records: List[Record] = []
     for file in files:
@@ -221,19 +216,22 @@ def record_file_maker_wrapper(
 
     records.append(Record(record_path, "", ""))
 
-    if record_callback is not _default:
-        records = record_callback(records)
-
     with StringIO(newline="") as buf:
         writer = csv.writer(buf)
-        for record in records:
-            writer.writerow(record)
+        for r in records:
+            writer.writerow(r)
         contents = buf.getvalue().encode("utf-8")
 
     yield File(record_path, contents)
 
 
-def wheel_name(name: str, version: str, pythons: str, abis: str, platforms: str) -> str:
+def wheel_name(
+    name: str,
+    version: str,
+    pythons: Iterable[str],
+    abis: Iterable[str],
+    platforms: Iterable[str],
+) -> str:
     stem = "-".join(
         [
             name,
@@ -249,7 +247,7 @@ def wheel_name(name: str, version: str, pythons: str, abis: str, platforms: str)
 class WheelBuilder:
     """A wheel that can be saved or converted to several formats."""
 
-    def __init__(self, name: str, files: List[File]) -> None:
+    def __init__(self, name: str, files: Iterable[File]) -> None:
         self._name = name
         self._files = files
 
@@ -259,9 +257,9 @@ def save_to_dir(self, path: Union[Path, str]) -> str:
 
         :returns the wheel file path
         """
-        path = Path(path) / self._name
-        path.write_bytes(self.as_bytes())
-        return str(path)
+        p = Path(path) / self._name
+        p.write_bytes(self.as_bytes())
+        return str(p)
 
     def save_to(self, path: Union[Path, str]) -> str:
         """Generate wheel file, saving to the provided path. Any parent
@@ -298,7 +296,6 @@ def make_wheel(
     console_scripts: Defaulted[List[str]] = _default,
     entry_points: Defaulted[Dict[str, List[str]]] = _default,
     record: Defaulted[Optional[AnyStr]] = _default,
-    record_callback: Defaulted[RecordCallback] = _default,
 ) -> WheelBuilder:
     """
     Helper function for generating test wheels which are compliant by default.
@@ -359,9 +356,6 @@ def make_wheel(
     :param entry_points:
     :param record: if provided and None, then no RECORD file is generated;
         else if a string then sets the content of the RECORD file
-    :param record_callback: callback function that receives and can edit the
-        records before they are written to RECORD, ignored if record is
-        provided
     """
     pythons = ["py2", "py3"]
     abis = ["none"]
@@ -388,7 +382,7 @@ def make_wheel(
     actual_files = filter(None, possible_files)
 
     files_and_record_file = record_file_maker_wrapper(
-        name, version, actual_files, record, record_callback
+        name, version, actual_files, record
     )
     wheel_file_name = wheel_name(name, version, pythons, abis, platforms)
 

tests/lib/wheel.py

@@ -10,4 +10,4 @@ setuptools
 virtualenv < 20.0
 werkzeug
 wheel
-toml
+tomli-w