Utils module

tomodachi_testcontainers.utils

AWSClientConfig

Bases: TypedDict

region_name instance-attribute

region_name

aws_access_key_id instance-attribute

aws_access_key_id

aws_secret_access_key instance-attribute

aws_secret_access_key

endpoint_url instance-attribute

endpoint_url

get_docker_image

get_docker_image(image_id, docker_client_kwargs=None)

Returns a Docker image, pulling it if not exists on host.

PARAMETER	DESCRIPTION
image_id	TYPE: str
docker_client_kwargs	TYPE: Optional[Dict] DEFAULT: None

Source code in src/tomodachi_testcontainers/utils.py

def get_docker_image(image_id: str, docker_client_kwargs: Optional[Dict] = None) -> Image:
    """Returns a Docker image, pulling it if not exists on host."""
    client = DockerClient(**(docker_client_kwargs or {}))
    try:
        return cast(Image, client.client.images.get(image_id))
    except ImageNotFound:
        return cast(Image, client.client.images.pull(image_id))

copy_files_to_container

copy_files_to_container(container, host_path, container_path)

Copies a folder or a file from the host to the container.

PARAMETER	DESCRIPTION
container	TYPE: Container
host_path	TYPE: Path
container_path	TYPE: Path

Source code in src/tomodachi_testcontainers/utils.py

def copy_files_to_container(container: Container, host_path: Path, container_path: Path) -> None:
    """Copies a folder or a file from the host to the container."""
    tar_stream = io.BytesIO()
    with tarfile.open(fileobj=tar_stream, mode="w") as tar:
        if host_path.is_dir():
            for root, _, files in os.walk(host_path):
                for file in files:
                    file_path = Path(root) / file
                    arcname = os.path.relpath(file_path, host_path)
                    tar.add(file_path, arcname=arcname)
        else:
            tar.add(host_path, arcname=host_path.name)
    tar_stream.seek(0)
    container.put_archive(path=container_path, data=tar_stream)

copy_files_from_container

copy_files_from_container(container, container_path, host_path)

Copies a folder or a file from the container to the host.

PARAMETER	DESCRIPTION
container	TYPE: Container
container_path	TYPE: Path
host_path	TYPE: Path

Source code in src/tomodachi_testcontainers/utils.py

def copy_files_from_container(container: Container, container_path: Path, host_path: Path) -> None:
    """Copies a folder or a file from the container to the host."""
    tar_stream, _ = container.get_archive(container_path)
    with tarfile.open(fileobj=io.BytesIO(b"".join(tar_stream))) as tar:
        tar.extractall(path=host_path)  # nosec: B202

get_available_port

get_available_port()

Returns a random available port on the host.

Source code in src/tomodachi_testcontainers/utils.py

def get_available_port() -> int:
    """Returns a random available port on the host."""
    with socket.socket() as sock:
        sock.bind(("", 0))
        return int(sock.getsockname()[1])

get_current_ip_address

get_current_ip_address()

Returns the current IP address of the host.

Source code in src/tomodachi_testcontainers/utils.py

def get_current_ip_address() -> str:
    """Returns the current IP address of the host."""
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.connect(("8.8.8.8", 80))
        return str(sock.getsockname()[0])

setup_logger

setup_logger(name)

Outputs logs to stderr for better visibility in pytest output.

Inspired by testcontainers.core.utils.setup_logger

PARAMETER	DESCRIPTION
name	TYPE: str

Source code in src/tomodachi_testcontainers/utils.py

def setup_logger(name: str) -> logging.Logger:
    """Outputs logs to stderr for better visibility in pytest output.

    Inspired by testcontainers.core.utils.setup_logger
    """
    logger = logging.getLogger(name)
    logger.setLevel(logging.INFO)
    handler = logging.StreamHandler()
    handler.setLevel(logging.INFO)
    handler.setFormatter(logging.Formatter(fmt="%(name)s: %(message)s"))
    logger.addHandler(handler)
    return logger