diff --git a/core/testcontainers/core/config.py b/core/testcontainers/core/config.py index 110a441e..397f7202 100644 --- a/core/testcontainers/core/config.py +++ b/core/testcontainers/core/config.py @@ -4,7 +4,7 @@ from os import environ from os.path import exists from pathlib import Path -from typing import Optional, Union +from typing import Optional, Union, cast class ConnectionMode(Enum): @@ -19,7 +19,7 @@ def use_mapped_port(self) -> bool: This is true for everything but bridge mode. """ - if self == self.bridge_ip: + if cast(str, self) == self.bridge_ip: return False return True @@ -43,7 +43,7 @@ def get_user_overwritten_connection_mode() -> Optional[ConnectionMode]: """ Return the user overwritten connection mode. """ - connection_mode: str | None = environ.get("TESTCONTAINERS_CONNECTION_MODE") + connection_mode: Union[str, None] = environ.get("TESTCONTAINERS_CONNECTION_MODE") if connection_mode: try: return ConnectionMode(connection_mode) diff --git a/core/testcontainers/core/docker_client.py b/core/testcontainers/core/docker_client.py index 83127884..0874f677 100644 --- a/core/testcontainers/core/docker_client.py +++ b/core/testcontainers/core/docker_client.py @@ -19,7 +19,7 @@ import urllib import urllib.parse from collections.abc import Iterable -from typing import Callable, Optional, TypeVar, Union +from typing import Any, Callable, Optional, TypeVar, Union, cast import docker from docker.models.containers import Container, ContainerCollection @@ -59,7 +59,7 @@ class DockerClient: Thin wrapper around :class:`docker.DockerClient` for a more functional interface. """ - def __init__(self, **kwargs) -> None: + def __init__(self, **kwargs: Any) -> None: docker_host = get_docker_host() if docker_host: @@ -82,14 +82,14 @@ def run( self, image: str, command: Optional[Union[str, list[str]]] = None, - environment: Optional[dict] = None, - ports: Optional[dict] = None, + environment: Optional[dict[str, str]] = None, + ports: Optional[dict[int, Optional[int]]] = None, labels: Optional[dict[str, str]] = None, detach: bool = False, stdout: bool = True, stderr: bool = False, remove: bool = False, - **kwargs, + **kwargs: Any, ) -> Container: # If the user has specified a network, we'll assume the user knows best if "network" not in kwargs and not get_docker_host(): @@ -112,7 +112,7 @@ def run( return container @_wrapped_image_collection - def build(self, path: str, tag: str, rm: bool = True, **kwargs) -> tuple[Image, Iterable[dict]]: + def build(self, path: str, tag: str, rm: bool = True, **kwargs: Any) -> tuple[Image, Iterable[dict[str, Any]]]: """ Build a Docker image from a directory containing the Dockerfile. @@ -151,28 +151,28 @@ def find_host_network(self) -> Optional[str]: except ipaddress.AddressValueError: continue if docker_host in subnet: - return network.name + return cast(str, network.name) except (ipaddress.AddressValueError, OSError): pass return None - def port(self, container_id: str, port: int) -> int: + def port(self, container_id: str, port: int) -> str: """ Lookup the public-facing port that is NAT-ed to :code:`port`. """ port_mappings = self.client.api.port(container_id, port) if not port_mappings: raise ConnectionError(f"Port mapping for container {container_id} and port {port} is " "not available") - return port_mappings[0]["HostPort"] + return cast(str, port_mappings[0]["HostPort"]) - def get_container(self, container_id: str) -> Container: + def get_container(self, container_id: str) -> dict[str, Any]: """ Get the container with a given identifier. """ containers = self.client.api.containers(filters={"id": container_id}) if not containers: raise RuntimeError(f"Could not get container with id {container_id}") - return containers[0] + return cast(dict[str, Any], containers[0]) def bridge_ip(self, container_id: str) -> str: """ @@ -180,7 +180,7 @@ def bridge_ip(self, container_id: str) -> str: """ container = self.get_container(container_id) network_name = self.network_name(container_id) - return container["NetworkSettings"]["Networks"][network_name]["IPAddress"] + return cast(str, container["NetworkSettings"]["Networks"][network_name]["IPAddress"]) def network_name(self, container_id: str) -> str: """ @@ -190,7 +190,7 @@ def network_name(self, container_id: str) -> str: name = container["HostConfig"]["NetworkMode"] if name == "default": return "bridge" - return name + return cast(str, name) def gateway_ip(self, container_id: str) -> str: """ @@ -198,7 +198,7 @@ def gateway_ip(self, container_id: str) -> str: """ container = self.get_container(container_id) network_name = self.network_name(container_id) - return container["NetworkSettings"]["Networks"][network_name]["Gateway"] + return cast(str, container["NetworkSettings"]["Networks"][network_name]["Gateway"]) def get_connection_mode(self) -> ConnectionMode: """ @@ -237,7 +237,7 @@ def host(self) -> str: # see https://github.com/testcontainers/testcontainers-python/issues/415 if url.hostname == "localnpipe" and utils.is_windows(): return "localhost" - return url.hostname + return cast(str, url.hostname) if utils.inside_container() and ("unix" in url.scheme or "npipe" in url.scheme): ip_address = utils.default_gateway_ip() if ip_address: @@ -251,9 +251,9 @@ def login(self, auth_config: DockerAuthInfo) -> None: login_info = self.client.login(**auth_config._asdict()) LOGGER.debug(f"logged in using {login_info}") - def client_networks_create(self, name: str, param: dict): + def client_networks_create(self, name: str, param: dict[str, Any]) -> dict[str, Any]: labels = create_labels("", param.get("labels")) - return self.client.networks.create(name, **{**param, "labels": labels}) + return cast(dict[str, Any], self.client.networks.create(name, **{**param, "labels": labels})) def get_docker_host() -> Optional[str]: