Skip to content

fix(core): Typing in docker_client #702

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 6 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions core/testcontainers/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ def use_mapped_port(self) -> bool:

This is true for everything but bridge mode.
"""
if self == self.bridge_ip:
if self == self.bridge_ip: # type: ignore[comparison-overlap]
return False
return True

Expand Down Expand Up @@ -63,7 +63,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)
Expand Down
37 changes: 19 additions & 18 deletions core/testcontainers/core/docker_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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():
Expand All @@ -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.

Expand Down Expand Up @@ -151,43 +151,43 @@ 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:
"""
Get the bridge ip address for a container.
"""
container = self.get_container(container_id)
network_name = self.network_name(container_id)
return container["NetworkSettings"]["Networks"][network_name]["IPAddress"]
return str(container["NetworkSettings"]["Networks"][network_name]["IPAddress"])

def network_name(self, container_id: str) -> str:
"""
Get the name of the network this container runs on
"""
container = self.get_container(container_id)
name = container["HostConfig"]["NetworkMode"]
name = str(container["HostConfig"]["NetworkMode"])
if name == "default":
return "bridge"
return name
Expand All @@ -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 str(container["NetworkSettings"]["Networks"][network_name]["Gateway"])

def get_connection_mode(self) -> ConnectionMode:
"""
Expand Down Expand Up @@ -235,9 +235,10 @@ def host(self) -> str:
return "localhost"
if "http" in url.scheme or "tcp" in url.scheme and url.hostname:
# see https://github.com/testcontainers/testcontainers-python/issues/415
if url.hostname == "localnpipe" and utils.is_windows():
hostname = url.hostname
if not hostname or (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:
Expand All @@ -251,9 +252,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]:
Expand Down