feat: add types for mypy (#101)

Author: lexi-k

README.md

@@ -180,7 +180,7 @@ run in CI.
 
 Use [black](https://pypi.org/project/black/) with default settings for
 formatting. You can also use `pylint` with `setup.cfg` as the configuration
-file.
+file as well as `mypy` for type checking.
 
 # Contributing
 

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = pytest-docker
-version =  3.0.0
+version =  3.1.0
 description = Simple pytest fixtures for Docker and Docker Compose based tests
 long_description = file: README.md
 long_description_content_type = text/markdown
@@ -41,29 +41,27 @@ docker-compose-v1 =
     docker-compose >=1.27.3, <2.0
 tests =
     requests >=2.22.0, <3.0
+    mypy >=0.500, <2.000
     pytest-pylint >=0.14.1, <1.0
     pytest-pycodestyle >=2.0.0, <3.0
+    pytest-mypy >=0.10, <1.0
+    types-requests >=2.31, <3.0
+    types-setuptools >=69.0, <70.0
 
 [options.entry_points]
 pytest11 =
     docker = pytest_docker
 
 [tool:pytest]
-addopts = --verbose --pylint-rcfile=setup.cfg
-# --pylint --pycodestyle
-
-[pycodestyle]
-max-line-length=120
-ignore=E4,E7,W3
+addopts = --verbose --mypy --pycodestyle --pylint-rcfile=setup.cfg --pylint
 
 # Configuration for pylint
-[MASTER]
-ignore=CVS
-good-names=logger,e,i,j,n,m,f,_
+[pylint.MASTER]
+good-names = "logger,e,i,j,n,m,f,_"
 
-[MESSAGES CONTROL]
-disable=all
-enable=unused-import,
+[pylint]
+disable = all
+enable = unused-import,
        fixme,
        useless-object-inheritance,
        unused-variable,
@@ -73,4 +71,15 @@ enable=unused-import,
        unreachable,
        invalid-name,
        logging-not-lazy,
-       unnecesary-pass
+       unnecesary-pass,
+       broad-except
+
+[pycodestyle]
+max-line-length=120
+ignore=E4,E7,W3
+
+[mypy]
+strict = true
+mypy_path = "src/pytest_docker,tests"
+namespace_packages = true
+warn_unused_ignores = true

src/pytest_docker/__init__.py

@@ -1,12 +1,13 @@
 import pytest
+
 from .plugin import (
+    docker_cleanup,
     docker_compose_command,
     docker_compose_file,
     docker_compose_project_name,
     docker_ip,
-    docker_setup,
-    docker_cleanup,
     docker_services,
+    docker_setup,
 )
 
 __all__ = [

src/pytest_docker/plugin.py

@@ -4,20 +4,24 @@
 import subprocess
 import time
 import timeit
+from typing import Any, Dict, Iterable, Iterator, List, Tuple, Union
 
 import attr
-
 import pytest
+from _pytest.config import Config
+from _pytest.fixtures import FixtureRequest
 
 
 @pytest.fixture
-def container_scope_fixture(request):
+def container_scope_fixture(request: FixtureRequest) -> Any:
     return request.config.getoption("--container-scope")
 
-def containers_scope(fixture_name, config):
+
+def containers_scope(fixture_name: str, config: Config) -> Any:  # pylint: disable=unused-argument
     return config.getoption("--container-scope", "session")
 
-def execute(command, success_codes=(0,)):
+
+def execute(command: str, success_codes: Iterable[int] = (0,)) -> Union[bytes, Any]:
     """Run a shell command."""
     try:
         output = subprocess.check_output(command, stderr=subprocess.STDOUT, shell=True)
@@ -29,14 +33,12 @@ def execute(command, success_codes=(0,)):
 
     if status not in success_codes:
         raise Exception(
-            'Command {} returned {}: """{}""".'.format(
-                command, status, output.decode("utf-8")
-            )
+            'Command {} returned {}: """{}""".'.format(command, status, output.decode("utf-8"))
         )
     return output
 
 
-def get_docker_ip():
+def get_docker_ip() -> Union[str, Any]:
     # When talking to the Docker daemon via a UNIX socket, route all TCP
     # traffic to docker containers via the TCP loopback interface.
     docker_host = os.environ.get("DOCKER_HOST", "").strip()
@@ -50,19 +52,18 @@ def get_docker_ip():
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_ip():
+def docker_ip() -> Union[str, Any]:
     """Determine the IP address for TCP connections to Docker containers."""
 
     return get_docker_ip()
 
 
 @attr.s(frozen=True)
 class Services:
+    _docker_compose: Any = attr.ib()
+    _services: Dict[Any, Dict[Any, Any]] = attr.ib(init=False, default=attr.Factory(dict))
 
-    _docker_compose = attr.ib()
-    _services = attr.ib(init=False, default=attr.Factory(dict))
-
-    def port_for(self, service, container_port):
+    def port_for(self, service: str, container_port: int) -> int:
         """Return the "host" port for `service` and `container_port`.
 
         E.g. If the service is defined like this:
@@ -78,16 +79,14 @@ def port_for(self, service, container_port):
         """
 
         # Lookup in the cache.
-        cache = self._services.get(service, {}).get(container_port, None)
+        cache: int = self._services.get(service, {}).get(container_port, None)
         if cache is not None:
             return cache
 
         output = self._docker_compose.execute("port %s %d" % (service, container_port))
         endpoint = output.strip().decode("utf-8")
         if not endpoint:
-            raise ValueError(
-                'Could not detect port for "%s:%d".' % (service, container_port)
-            )
+            raise ValueError('Could not detect port for "%s:%d".' % (service, container_port))
 
         # This handles messy output that might contain warnings or other text
         if len(endpoint.split("\n")) > 1:
@@ -101,7 +100,13 @@ def port_for(self, service, container_port):
 
         return match
 
-    def wait_until_responsive(self, check, timeout, pause, clock=timeit.default_timer):
+    def wait_until_responsive(
+        self,
+        check: Any,
+        timeout: float,
+        pause: float,
+        clock: Any = timeit.default_timer,
+    ) -> None:
         """Wait until a service is responsive."""
 
         ref = clock()
@@ -115,20 +120,19 @@ def wait_until_responsive(self, check, timeout, pause, clock=timeit.default_time
         raise Exception("Timeout reached while waiting on service!")
 
 
-def str_to_list(arg):
+def str_to_list(arg: Union[str, List[Any], Tuple[Any]]) -> Union[List[Any], Tuple[Any]]:
     if isinstance(arg, (list, tuple)):
         return arg
     return [arg]
 
 
 @attr.s(frozen=True)
 class DockerComposeExecutor:
+    _compose_command: str = attr.ib()
+    _compose_files: Any = attr.ib(converter=str_to_list)
+    _compose_project_name: str = attr.ib()
 
-    _compose_command = attr.ib()
-    _compose_files = attr.ib(converter=str_to_list)
-    _compose_project_name = attr.ib()
-
-    def execute(self, subcommand):
+    def execute(self, subcommand: str) -> Union[bytes, Any]:
         command = self._compose_command
         for compose_file in self._compose_files:
             command += ' -f "{}"'.format(compose_file)
@@ -137,7 +141,7 @@ def execute(self, subcommand):
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_compose_command():
+def docker_compose_command() -> str:
     """Docker Compose command to use, it could be either `docker compose`
     for Docker Compose V2 or `docker-compose` for Docker Compose
     V1."""
@@ -146,40 +150,40 @@ def docker_compose_command():
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_compose_file(pytestconfig):
+def docker_compose_file(pytestconfig: Any) -> str:
     """Get an absolute path to the  `docker-compose.yml` file. Override this
     fixture in your tests if you need a custom location."""
 
     return os.path.join(str(pytestconfig.rootdir), "tests", "docker-compose.yml")
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_compose_project_name():
+def docker_compose_project_name() -> str:
     """Generate a project name using the current process PID. Override this
     fixture in your tests if you need a particular project name."""
 
     return "pytest{}".format(os.getpid())
 
 
-def get_cleanup_command():
+def get_cleanup_command() -> Union[List[str], str]:
     return ["down -v"]
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_cleanup():
+def docker_cleanup() -> Union[List[str], str]:
     """Get the docker_compose command to be executed for test clean-up actions.
     Override this fixture in your tests if you need to change clean-up actions.
     Returning anything that would evaluate to False will skip this command."""
 
     return get_cleanup_command()
 
 
-def get_setup_command():
+def get_setup_command() -> Union[List[str], str]:
     return ["up --build -d"]
 
 
 @pytest.fixture(scope=containers_scope)
-def docker_setup():
+def docker_setup() -> Union[List[str], str]:
     """Get the docker_compose command to be executed for test setup actions.
     Override this fixture in your tests if you need to change setup actions.
     Returning anything that would evaluate to False will skip this command."""
@@ -189,12 +193,12 @@ def docker_setup():
 
 @contextlib.contextmanager
 def get_docker_services(
-    docker_compose_command,
-    docker_compose_file,
-    docker_compose_project_name,
-    docker_setup,
-    docker_cleanup,
-):
+    docker_compose_command: str,
+    docker_compose_file: str,
+    docker_compose_project_name: str,
+    docker_setup: Union[List[str], str],
+    docker_cleanup: Union[List[str], str],
+) -> Iterator[Services]:
     docker_compose = DockerComposeExecutor(
         docker_compose_command, docker_compose_file, docker_compose_project_name
     )
@@ -222,12 +226,12 @@ def get_docker_services(
 
 @pytest.fixture(scope=containers_scope)
 def docker_services(
-    docker_compose_command,
-    docker_compose_file,
-    docker_compose_project_name,
-    docker_setup,
-    docker_cleanup,
-):
+    docker_compose_command: str,
+    docker_compose_file: str,
+    docker_compose_project_name: str,
+    docker_setup: str,
+    docker_cleanup: str,
+) -> Iterator[Services]:
     """Start all services from a docker compose file (`docker-compose up`).
     After test are finished, shutdown all services (`docker-compose down`)."""
 

src/pytest_docker/py.typed

@@ -1 +1 @@
-pytest_plugins = ["pytester"]  # pylint: disable=invalid-name
+pytest_plugins = ["pytester"]

tests/conftest.py

@@ -2,7 +2,7 @@
 from wsgiref.simple_server import make_server
 
 
-def test_app(_, start_response):
+def test_app(_, start_response):  # type: ignore
     # This path is set up as a volume in the test's docker-compose.yml,
     # so we make sure that we really work with Docker Compose.
     if path.exists("/test_volume"):

tests/containers/hello/server.py

@@ -1,33 +1,32 @@
+from typing import Dict
 from unittest import mock
 
 import pytest
 from pytest_docker.plugin import get_docker_ip
 
 
-def test_docker_ip_native():
-    environ = {}
+def test_docker_ip_native() -> None:
+    environ: Dict[str, str] = {}
     with mock.patch("os.environ", environ):
         assert get_docker_ip() == "127.0.0.1"
 
 
-def test_docker_ip_remote():
+def test_docker_ip_remote() -> None:
     environ = {"DOCKER_HOST": "tcp://1.2.3.4:2376"}
     with mock.patch("os.environ", environ):
         assert get_docker_ip() == "1.2.3.4"
 
 
-def test_docker_ip_unix():
+def test_docker_ip_unix() -> None:
     environ = {"DOCKER_HOST": "unix:///run/user/1000/podman/podman.sock"}
     with mock.patch("os.environ", environ):
         assert get_docker_ip() == "127.0.0.1"
 
 
 @pytest.mark.parametrize("docker_host", ["http://1.2.3.4:2376"])
-def test_docker_ip_remote_invalid(docker_host):
+def test_docker_ip_remote_invalid(docker_host: str) -> None:
     environ = {"DOCKER_HOST": docker_host}
     with mock.patch("os.environ", environ):
         with pytest.raises(ValueError) as exc:
             print(get_docker_ip())
-        assert str(exc.value) == (
-            'Invalid value for DOCKER_HOST: "%s".' % (docker_host,)
-        )
+        assert str(exc.value) == ('Invalid value for DOCKER_HOST: "%s".' % (docker_host,))