Merge pull request #177 from pytest-dev/176-add-mypy-to-ci

Add mypy to CI

Author: vitalk

.gitignore

@@ -29,3 +29,4 @@ src/pytest_flask/_version.py
 # Editors
 .vscode
 .code-workspace
+.python-version

docs/changelog.rst

@@ -6,8 +6,9 @@ Changelog
 UNRELEASED
 ----------
 
-* Added support for Python 3.10, 3.11, and 3.12.
-* Dropped support for EOL Python 3.7.
+* Add support for Python 3.10, 3.11, and 3.12.
+* Drop support for EOL Python 3.7.
+* Add type hints.
 
 1.3.0 (2023-10-23)
 ------------------

pyproject.toml

@@ -4,3 +4,12 @@ requires = [
   "setuptools-scm[toml]",
 ]
 build-backend = "setuptools.build_meta"
+
+[tool.mypy]
+warn_unreachable = true
+warn_unused_ignores = true
+warn_redundant_casts = true
+enable_error_code = [
+    "ignore-without-code",
+    "truthy-bool",
+]

requirements/test.txt

@@ -2,3 +2,4 @@ mock
 pylint
 coverage
 pytest-pep8
+mypy

src/pytest_flask/_internal.py

@@ -1,8 +1,15 @@
 import functools
 import warnings
+from typing import Callable
+from typing import Literal
 
+from pytest import Config as _PytestConfig
 
-def deprecated(reason):
+
+_PytestScopeName = Literal["session", "package", "module", "class", "function"]
+
+
+def deprecated(reason: str) -> Callable:
     """Decorator which can be used to mark function or method as deprecated.
     It will result a warning being emitted when the function is called."""
 
@@ -19,15 +26,15 @@ def deprecated_call(*args, **kwargs):
     return decorator
 
 
-def _rewrite_server_name(server_name, new_port):
+def _rewrite_server_name(server_name: str, new_port: str) -> str:
     """Rewrite server port in ``server_name`` with ``new_port`` value."""
     sep = ":"
     if sep in server_name:
         server_name, _ = server_name.split(sep, 1)
     return sep.join((server_name, new_port))
 
 
-def _determine_scope(*, fixture_name, config):
+def _determine_scope(*, fixture_name: str, config: _PytestConfig) -> _PytestScopeName:
     return config.getini("live_server_scope")
 
 

src/pytest_flask/fixtures.py

@@ -1,7 +1,15 @@
 #!/usr/bin/env python
 import socket
+from typing import Any
+from typing import cast
+from typing import Generator
 
 import pytest
+from flask import Flask as _FlaskApp
+from flask.config import Config as _FlaskAppConfig
+from flask.testing import FlaskClient as _FlaskTestClient
+from pytest import Config as _PytestConfig
+from pytest import FixtureRequest as _PytestFixtureRequest
 
 from ._internal import _determine_scope
 from ._internal import _make_accept_header
@@ -10,7 +18,7 @@
 
 
 @pytest.fixture
-def client(app):
+def client(app: _FlaskApp) -> Generator[_FlaskTestClient, Any, Any]:
     """A Flask test client. An instance of :class:`flask.testing.TestClient`
     by default.
     """
@@ -19,7 +27,7 @@ def client(app):
 
 
 @pytest.fixture
-def client_class(request, client):
+def client_class(request: _PytestFixtureRequest, client: _FlaskTestClient) -> None:
     """Uses to set a ``client`` class attribute to current Flask test client::
 
     @pytest.mark.usefixtures('client_class')
@@ -37,8 +45,10 @@ def test_login(self):
         request.cls.client = client
 
 
-@pytest.fixture(scope=_determine_scope)
-def live_server(request, app, pytestconfig):  # pragma: no cover
+@pytest.fixture(scope=_determine_scope)  # type: ignore[arg-type]
+def live_server(
+    request: _PytestFixtureRequest, app: _FlaskApp, pytestconfig: _PytestConfig
+) -> Generator[LiveServer, Any, Any]:  # pragma: no cover
     """Run application in a separate process.
 
     When the ``live_server`` fixture is applied, the ``url_for`` function
@@ -64,34 +74,36 @@ def test_server_is_up_and_running(live_server):
         port = s.getsockname()[1]
         s.close()
 
-    host = pytestconfig.getvalue("live_server_host")
+    host = cast(str, pytestconfig.getvalue("live_server_host"))
 
     # Explicitly set application ``SERVER_NAME`` for test suite
     original_server_name = app.config["SERVER_NAME"] or "localhost.localdomain"
     final_server_name = _rewrite_server_name(original_server_name, str(port))
     app.config["SERVER_NAME"] = final_server_name
 
-    wait = request.config.getvalue("live_server_wait")
-    clean_stop = request.config.getvalue("live_server_clean_stop")
+    wait = cast(int, request.config.getvalue("live_server_wait"))
+    clean_stop = cast(bool, request.config.getvalue("live_server_clean_stop"))
+
     server = LiveServer(app, host, port, wait, clean_stop)
     if request.config.getvalue("start_live_server"):
         server.start()
 
     request.addfinalizer(server.stop)
+
     yield server
 
     if original_server_name is not None:
         app.config["SERVER_NAME"] = original_server_name
 
 
 @pytest.fixture
-def config(app):
+def config(app: _FlaskApp) -> _FlaskAppConfig:
     """An application config."""
     return app.config
 
 
 @pytest.fixture(params=["application/json", "text/html"])
-def mimetype(request):
+def mimetype(request) -> str:
     return request.param
 
 

src/pytest_flask/live_server.py

@@ -5,13 +5,30 @@
 import signal
 import socket
 import time
+from multiprocessing import Process
+from typing import Any
+from typing import cast
+from typing import Protocol
+from typing import Union
 
 import pytest
 
 
+class _SupportsFlaskAppRun(Protocol):
+    def run(
+        self,
+        host: Union[str, None] = None,
+        port: Union[int, None] = None,
+        debug: Union[bool, None] = None,
+        load_dotenv: bool = True,
+        **options: Any,
+    ) -> None:
+        ...
+
+
 # force 'fork' on macOS
 if platform.system() == "Darwin":
-    multiprocessing = multiprocessing.get_context("fork")
+    multiprocessing = multiprocessing.get_context("fork")  # type: ignore[assignment]
 
 
 class LiveServer:  # pragma: no cover
@@ -25,18 +42,25 @@ class LiveServer:  # pragma: no cover
                  application is not started.
     """
 
-    def __init__(self, app, host, port, wait, clean_stop=False):
+    def __init__(
+        self,
+        app: _SupportsFlaskAppRun,
+        host: str,
+        port: int,
+        wait: int,
+        clean_stop: bool = False,
+    ):
         self.app = app
         self.port = port
         self.host = host
         self.wait = wait
         self.clean_stop = clean_stop
-        self._process = None
+        self._process: Union[Process, None] = None
 
-    def start(self):
+    def start(self) -> None:
         """Start application in a separate process."""
 
-        def worker(app, host, port):
+        def worker(app: _SupportsFlaskAppRun, host: str, port: int) -> None:
             app.run(host=host, port=port, use_reloader=False, threaded=True)
 
         self._process = multiprocessing.Process(
@@ -45,7 +69,7 @@ def worker(app, host, port):
         self._process.daemon = True
         self._process.start()
 
-        keep_trying = True
+        keep_trying: bool = True
         start_time = time.time()
         while keep_trying:
             elapsed_time = time.time() - start_time
@@ -57,7 +81,7 @@ def worker(app, host, port):
             if self._is_ready():
                 keep_trying = False
 
-    def _is_ready(self):
+    def _is_ready(self) -> bool:
         sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
         try:
             sock.connect((self.host, self.port))
@@ -69,13 +93,13 @@ def _is_ready(self):
             sock.close()
         return ret
 
-    def url(self, url=""):
+    def url(self, url: str = "") -> str:
         """Returns the complete url based on server options."""
         return "http://{host!s}:{port!s}{url!s}".format(
             host=self.host, port=self.port, url=url
         )
 
-    def stop(self):
+    def stop(self) -> None:
         """Stop application process."""
         if self._process:
             if self.clean_stop and self._stop_cleanly():
@@ -84,14 +108,17 @@ def stop(self):
                 # If it's still alive, kill it
                 self._process.terminate()
 
-    def _stop_cleanly(self, timeout=5):
+    def _stop_cleanly(self, timeout: int = 5) -> bool:
         """Attempts to stop the server cleanly by sending a SIGINT
         signal and waiting for ``timeout`` seconds.
 
         :return: True if the server was cleanly stopped, False otherwise.
         """
+        if not self._process:
+            return True
+
         try:
-            os.kill(self._process.pid, signal.SIGINT)
+            os.kill(cast(int, self._process.pid), signal.SIGINT)
             self._process.join(timeout)
             return True
         except Exception as ex:

src/pytest_flask/plugin.py

@@ -5,7 +5,15 @@
     :copyright: (c) by Vital Kudzelka
     :license: MIT
 """
+from typing import Any
+from typing import List
+from typing import Protocol
+from typing import Type
+from typing import TypeVar
+from typing import Union
+
 import pytest
+from _pytest.config import Config as _PytestConfig
 
 from .fixtures import accept_any
 from .fixtures import accept_json
@@ -18,31 +26,48 @@
 from .pytest_compat import getfixturevalue
 
 
+_Response = TypeVar("_Response")
+
+
+class _SupportsPytestFlaskEqual(Protocol):
+    status_code: int
+
+    def __eq__(self, other: Any) -> bool:
+        ...
+
+    def __ne__(self, other: Any) -> bool:
+        ...
+
+
 class JSONResponse:
     """Mixin with testing helper methods for JSON responses."""
 
-    def __eq__(self, other):
+    status_code: int
+
+    def __eq__(self, other) -> bool:
         if isinstance(other, int):
             return self.status_code == other
         return super().__eq__(other)
 
-    def __ne__(self, other):
+    def __ne__(self, other) -> bool:
         return not self == other
 
 
-def pytest_assertrepr_compare(op, left, right):
+def pytest_assertrepr_compare(
+    op: str, left: _SupportsPytestFlaskEqual, right: int
+) -> Union[List[str], None]:
     if isinstance(left, JSONResponse) and op == "==" and isinstance(right, int):
         return [
             "Mismatch in status code for response: {} != {}".format(
                 left.status_code,
                 right,
             ),
-            f"Response status: {left.status}",
+            f"Response status: {left.status_code}",
         ]
     return None
 
 
-def _make_test_response_class(response_class):
+def _make_test_response_class(response_class: Type[_Response]) -> Type[_Response]:
     """Extends the response class with special attribute to test JSON
     responses. Don't override user-defined `json` attribute if any.
 
@@ -186,7 +211,7 @@ def pytest_addoption(parser):
     )
 
 
-def pytest_configure(config):
+def pytest_configure(config: _PytestConfig) -> None:
     config.addinivalue_line(
         "markers", "app(options): pass options to your application factory"
     )

src/pytest_flask/py.typed

@@ -1,6 +1,6 @@
 [tox]
 envlist =
-    py{37,38,39,linting}
+    py{38,39,310,311,312,linting},mypy
 
 [pytest]
 norecursedirs = .git .tox env coverage docs
@@ -16,6 +16,7 @@ deps=
     -rrequirements/test.txt
     -rrequirements/docs.txt
     pre-commit>=1.11.0
+    mypy>=1.6.1
     tox
 basepython = python3.8
 usedevelop = True
@@ -56,6 +57,11 @@ basepython = python3.8
 deps = pre-commit>=1.11.0
 commands = pre-commit run --all-files --show-diff-on-failure {posargs:}
 
+[testenv:mypy]
+basepython = python3.8
+deps = mypy>=1.6.1
+commands = mypy src
+
 [testenv:docs]
 changedir = docs
 skipsdist = True