Skip to content

Tomodachi Testcontainers

tomodachi_testcontainers

DockerContainer 

DockerContainer(*args, disable_logging=False, **kwargs)

Bases: DockerContainer, ABC

Abstract class for generic Docker containers.

Source code in src/tomodachi_testcontainers/containers/common/container.py 
27
28
29
30
31
32
33
34
def __init__(self, *args: Any, disable_logging: bool = False, **kwargs: Any) -> None:
    self._set_container_network()

    super().__init__(*args, **kwargs, network=self.network)

    self._set_default_container_name()

    self._disable_logging = disable_logging

log_message_on_container_start abstractmethod 

log_message_on_container_start()

Returns a message that will be logged when the container starts.

Source code in src/tomodachi_testcontainers/containers/common/container.py 
53
54
55
@abc.abstractmethod
def log_message_on_container_start(self) -> str:
    """Returns a message that will be logged when the container starts."""

get_container_internal_ip 

get_container_internal_ip()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
57
58
def get_container_internal_ip(self) -> str:
    return self.docker_inspect()["NetworkSettings"]["Networks"][self.network]["IPAddress"]

get_container_gateway_ip 

get_container_gateway_ip()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
60
61
def get_container_gateway_ip(self) -> str:
    return self.docker_inspect()["NetworkSettings"]["Networks"][self.network]["Gateway"]

docker_inspect 

docker_inspect()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
63
64
def docker_inspect(self) -> dict[str, Any]:
    return self.get_docker_client().get_container(self.get_wrapped_container().id)

start 

start()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
66
67
68
69
70
def start(self) -> "DockerContainer":
    self._setup_logger()
    self._start()
    self._log_message_on_container_start()
    return self

stop 

stop()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
72
73
74
75
76
def stop(self) -> None:
    with suppress(Exception):
        container = self._container or cast("Container", self.get_docker_client().client.containers.get(self._name))
        container.remove(force=True, v=True)
    self._container = None

restart 

restart()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
78
79
def restart(self) -> None:
    self.get_wrapped_container().restart()

EphemeralDockerImage 

EphemeralDockerImage(
    dockerfile=None,
    context=None,
    target=None,
    docker_client_kwargs=None,
    *,
    remove_image_on_exit=True,
)

Builds a Docker image from a given Dockerfile and removes it when the context manager exits.

Source code in src/tomodachi_testcontainers/containers/common/image.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
def __init__(
    self,
    dockerfile: Path | None = None,
    context: Path | None = None,
    target: str | None = None,
    docker_client_kwargs: dict | None = None,
    *,
    remove_image_on_exit: bool = True,
) -> None:
    self.dockerfile = str(dockerfile) if dockerfile else None
    self.context = str(context) if context else "."
    self.target = target
    self._docker_client = DockerClient(**(docker_client_kwargs or {}))
    self._remove_image_on_exit = remove_image_on_exit

image instance-attribute 

image

dockerfile instance-attribute 

dockerfile = str(dockerfile) if dockerfile else None

context instance-attribute 

context = str(context) if context else '.'

target instance-attribute 

target = target

WebContainer 

WebContainer(
    image,
    internal_port,
    edge_port=None,
    http_healthcheck_path=None,
    disable_logging=False,
    **kwargs,
)

Bases: DockerContainer, ABC

Abstract class for web application containers.

Source code in src/tomodachi_testcontainers/containers/common/web.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(
    self,
    image: str,
    internal_port: int,
    edge_port: int | None = None,
    http_healthcheck_path: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(image, disable_logging=disable_logging, **kwargs)
    self.internal_port = internal_port
    self.edge_port = edge_port or get_available_port()
    self.http_healthcheck_path = http_healthcheck_path
    self.with_bind_ports(internal_port, self.edge_port)

internal_port instance-attribute 

internal_port = internal_port

edge_port instance-attribute 

edge_port = edge_port or get_available_port()

http_healthcheck_path instance-attribute 

http_healthcheck_path = http_healthcheck_path

get_internal_url 

get_internal_url()
Source code in src/tomodachi_testcontainers/containers/common/web.py 
38
39
40
def get_internal_url(self) -> str:
    ip = self.get_container_internal_ip()
    return f"http://{ip}:{self.internal_port}"

get_external_url 

get_external_url()
Source code in src/tomodachi_testcontainers/containers/common/web.py 
42
43
44
def get_external_url(self) -> str:
    host = self.get_container_host_ip()
    return f"http://{host}:{self.edge_port}"

start 

start()
Source code in src/tomodachi_testcontainers/containers/common/web.py 
46
47
48
49
50
51
def start(self) -> "WebContainer":
    super().start()
    if self.http_healthcheck_path:
        url = urllib.parse.urljoin(self.get_external_url(), self.http_healthcheck_path)
        wait_for_http_healthcheck(url=url)
    return self

DynamoDBAdminContainer 

DynamoDBAdminContainer(
    dynamo_endpoint,
    image="aaronshaf/dynamodb-admin:latest",
    internal_port=8001,
    edge_port=None,
    region_name=None,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

DynamoDB Admin container.

Configuration environment variables (set on host machine):

  • AWS_REGION or AWS_DEFAULT_REGION - defaults to us-east-1
Source code in src/tomodachi_testcontainers/containers/dynamodb_admin.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    dynamo_endpoint: str,
    image: str = "aaronshaf/dynamodb-admin:latest",
    internal_port: int = 8001,
    edge_port: int | None = None,
    region_name: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image=image,
        internal_port=internal_port,
        edge_port=edge_port,
        http_healthcheck_path="/",
        disable_logging=disable_logging,
        **kwargs,
    )

    self.region_name = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-1"
    self.aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID") or "testing"  # nosec: B105
    self.aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY") or "testing"  # nosec: B105

    self.with_env("AWS_REGION", self.region_name)
    self.with_env("AWS_DEFAULT_REGION", self.region_name)
    self.with_env("AWS_ACCESS_KEY_ID", self.aws_access_key_id)
    self.with_env("AWS_SECRET_ACCESS_KEY", self.aws_secret_access_key)

    self.with_env("DYNAMO_ENDPOINT", dynamo_endpoint)

region_name instance-attribute 

region_name = (
    region_name
    or getenv("AWS_REGION")
    or getenv("AWS_DEFAULT_REGION")
    or "us-east-1"
)

aws_access_key_id instance-attribute 

aws_access_key_id = getenv("AWS_ACCESS_KEY_ID") or "testing"

aws_secret_access_key instance-attribute 

aws_secret_access_key = (
    getenv("AWS_SECRET_ACCESS_KEY") or "testing"
)

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/dynamodb_admin.py 
45
46
def log_message_on_container_start(self) -> str:
    return f"DynamoDB Admin: http://localhost:{self.edge_port}"

LocalStackContainer 

LocalStackContainer(
    image="localstack/localstack:4",
    internal_port=4566,
    edge_port=None,
    region_name=None,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

LocalStack container.

Configuration environment variables (set on host machine):

  • AWS_REGION or AWS_DEFAULT_REGION - defaults to us-east-1
  • AWS_ACCESS_KEY_ID - defaults to testing
  • AWS_SECRET_ACCESS_KEY - defaults to testing
Source code in src/tomodachi_testcontainers/containers/localstack.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
def __init__(
    self,
    image: str = "localstack/localstack:4",
    internal_port: int = 4566,
    edge_port: int | None = None,
    region_name: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=internal_port,
        edge_port=edge_port,
        http_healthcheck_path="/_localstack/health",
        disable_logging=disable_logging,
        **kwargs,
    )

    self.region_name = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-1"
    self.aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID") or "testing"  # nosec: B105
    self.aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY") or "testing"  # nosec: B105

    self.with_env("AWS_REGION", self.region_name)
    self.with_env("AWS_DEFAULT_REGION", self.region_name)
    self.with_env("AWS_ACCESS_KEY_ID", self.aws_access_key_id)
    self.with_env("AWS_SECRET_ACCESS_KEY", self.aws_secret_access_key)

    # Docker is needed for running AWS Lambda container
    self.with_env("LAMBDA_DOCKER_NETWORK", self.network)
    self.with_volume_mapping("/var/run/docker.sock", "/var/run/docker.sock")

region_name instance-attribute 

region_name = (
    region_name
    or getenv("AWS_REGION")
    or getenv("AWS_DEFAULT_REGION")
    or "us-east-1"
)

aws_access_key_id instance-attribute 

aws_access_key_id = getenv("AWS_ACCESS_KEY_ID") or "testing"

aws_secret_access_key instance-attribute 

aws_secret_access_key = (
    getenv("AWS_SECRET_ACCESS_KEY") or "testing"
)

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/localstack.py 
52
53
def log_message_on_container_start(self) -> str:
    return f"LocalStack started: http://localhost:{self.edge_port}/"

get_aws_client_config 

get_aws_client_config()
Source code in src/tomodachi_testcontainers/containers/localstack.py 
55
56
57
58
59
60
61
def get_aws_client_config(self) -> AWSClientConfig:
    return AWSClientConfig(
        region_name=self.region_name,
        aws_access_key_id=self.aws_access_key_id,
        aws_secret_access_key=self.aws_secret_access_key,
        endpoint_url=self.get_external_url(),
    )

start 

start()
Source code in src/tomodachi_testcontainers/containers/localstack.py 
63
64
65
66
def start(self) -> "LocalStackContainer":
    super().start()
    LogMessageWaitStrategy(r"Ready\.\n").with_startup_timeout(10).wait_until_ready(self)
    return self

MinioContainer 

MinioContainer(
    image="minio/minio:latest",
    s3_api_internal_port=9000,
    s3_api_edge_port=None,
    console_internal_port=9001,
    console_edge_port=None,
    region_name=None,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

Minio container.

Configuration environment variables (set on host machine):

  • AWS_REGION or AWS_DEFAULT_REGION - defaults to us-east-1
  • MINIO_ROOT_USER - defaults to minioadmin
  • MINIO_ROOT_PASSWORD - defaults to minioadmin
Source code in src/tomodachi_testcontainers/containers/minio.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    image: str = "minio/minio:latest",
    s3_api_internal_port: int = 9000,
    s3_api_edge_port: int | None = None,
    console_internal_port: int = 9001,
    console_edge_port: int | None = None,
    region_name: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=s3_api_internal_port,
        edge_port=s3_api_edge_port,
        http_healthcheck_path="/minio/health/live",
        disable_logging=disable_logging,
        **kwargs,
    )
    self.s3_api_internal_port = s3_api_internal_port
    self.s3_api_edge_port = self.edge_port
    self.console_internal_port = console_internal_port
    self.console_edge_port = console_edge_port or get_available_port()

    self.with_bind_ports(console_internal_port, self.console_edge_port)

    self.region_name = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-1"
    self.minio_root_user = os.getenv("MINIO_ROOT_USER") or "minioadmin"  # nosec: B105
    self.minio_root_password = os.getenv("MINIO_ROOT_PASSWORD") or "minioadmin"  # nosec: B105

    self.with_env("MINIO_ROOT_USER", self.minio_root_user)
    self.with_env("MINIO_ROOT_PASSWORD", self.minio_root_password)

    self.with_command(
        f'server /data --address ":{self.s3_api_internal_port}" --console-address ":{self.console_internal_port}"'
    )

s3_api_internal_port instance-attribute 

s3_api_internal_port = s3_api_internal_port

s3_api_edge_port instance-attribute 

s3_api_edge_port = edge_port

console_internal_port instance-attribute 

console_internal_port = console_internal_port

console_edge_port instance-attribute 

console_edge_port = (
    console_edge_port or get_available_port()
)

region_name instance-attribute 

region_name = (
    region_name
    or getenv("AWS_REGION")
    or getenv("AWS_DEFAULT_REGION")
    or "us-east-1"
)

minio_root_user instance-attribute 

minio_root_user = getenv('MINIO_ROOT_USER') or 'minioadmin'

minio_root_password instance-attribute 

minio_root_password = (
    getenv("MINIO_ROOT_PASSWORD") or "minioadmin"
)

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/minio.py 
56
57
58
59
60
61
def log_message_on_container_start(self) -> str:
    return (
        "Minio started: "
        f"S3-API: http://localhost:{self.s3_api_edge_port}/; "
        f"console: http://localhost:{self.console_edge_port}/"
    )

get_aws_client_config 

get_aws_client_config()
Source code in src/tomodachi_testcontainers/containers/minio.py 
63
64
65
66
67
68
69
def get_aws_client_config(self) -> AWSClientConfig:
    return AWSClientConfig(
        region_name=self.region_name,
        aws_access_key_id=self.minio_root_user,
        aws_secret_access_key=self.minio_root_password,
        endpoint_url=self.get_external_url(),
    )

reset_minio 

reset_minio()
Source code in src/tomodachi_testcontainers/containers/minio.py 
71
72
def reset_minio(self) -> None:
    self.exec(["mc", "rm", "--recursive", "--dangerous", "--force", "data/"])

MotoContainer 

MotoContainer(
    image="motoserver/moto:latest",
    internal_port=5000,
    edge_port=None,
    region_name=None,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

Moto container.

Configuration environment variables (set on host machine):

  • AWS_REGION or AWS_DEFAULT_REGION - defaults to us-east-1
  • AWS_ACCESS_KEY_ID - defaults to testing
  • AWS_SECRET_ACCESS_KEY - defaults to testing
Source code in src/tomodachi_testcontainers/containers/moto.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    image: str = "motoserver/moto:latest",
    internal_port: int = 5000,
    edge_port: int | None = None,
    region_name: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=internal_port,
        edge_port=edge_port,
        http_healthcheck_path="/moto-api/data.json",
        disable_logging=disable_logging,
        **kwargs,
    )

    self.region_name = region_name or os.getenv("AWS_REGION") or os.getenv("AWS_DEFAULT_REGION") or "us-east-1"
    self.aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID") or "testing"  # nosec: B105
    self.aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY") or "testing"  # nosec: B105

    self.with_env("AWS_REGION", self.region_name)
    self.with_env("AWS_DEFAULT_REGION", self.region_name)
    self.with_env("AWS_ACCESS_KEY_ID", self.aws_access_key_id)
    self.with_env("AWS_SECRET_ACCESS_KEY", self.aws_secret_access_key)

    self.with_env("MOTO_PORT", str(self.internal_port))

    # Docker is needed for running AWS Lambda container
    self.with_env("MOTO_DOCKER_NETWORK_NAME", self.network)
    self.with_volume_mapping("/var/run/docker.sock", "/var/run/docker.sock")

region_name instance-attribute 

region_name = (
    region_name
    or getenv("AWS_REGION")
    or getenv("AWS_DEFAULT_REGION")
    or "us-east-1"
)

aws_access_key_id instance-attribute 

aws_access_key_id = getenv("AWS_ACCESS_KEY_ID") or "testing"

aws_secret_access_key instance-attribute 

aws_secret_access_key = (
    getenv("AWS_SECRET_ACCESS_KEY") or "testing"
)

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/moto.py 
54
55
def log_message_on_container_start(self) -> str:
    return f"Moto dashboard: http://localhost:{self.edge_port}/moto-api"

get_aws_client_config 

get_aws_client_config()
Source code in src/tomodachi_testcontainers/containers/moto.py 
57
58
59
60
61
62
63
def get_aws_client_config(self) -> AWSClientConfig:
    return AWSClientConfig(
        region_name=self.region_name,
        aws_access_key_id=self.aws_access_key_id,
        aws_secret_access_key=self.aws_secret_access_key,
        endpoint_url=self.get_external_url(),
    )

start 

start()
Source code in src/tomodachi_testcontainers/containers/moto.py 
65
66
67
68
def start(self) -> "MotoContainer":
    super().start()
    LogMessageWaitStrategy("Running on all addresses").with_startup_timeout(10).wait_until_ready(self)
    return self

reset_moto 

reset_moto()
Source code in src/tomodachi_testcontainers/containers/moto.py 
70
71
def reset_moto(self) -> None:
    self.exec(["curl", "-X", "POST", f"http://localhost:{self.internal_port}/moto-api/reset"])

TomodachiContainer 

TomodachiContainer(
    image,
    internal_port=9700,
    edge_port=None,
    http_healthcheck_path=None,
    *,
    export_coverage=False,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

Tomodachi container.

Configuration environment variables (set on host machine):

  • TOMODACHI_TESTCONTAINER_EXPORT_COVERAGE - defaults to False
Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
def __init__(
    self,
    image: str,
    internal_port: int = 9700,
    edge_port: int | None = None,
    http_healthcheck_path: str | None = None,
    *,
    export_coverage: bool = False,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image=image,
        internal_port=internal_port,
        edge_port=edge_port,
        http_healthcheck_path=http_healthcheck_path,
        disable_logging=disable_logging,
        **kwargs,
    )
    self._export_coverage = export_coverage or bool(os.getenv("TOMODACHI_TESTCONTAINER_EXPORT_COVERAGE"))
    if self._export_coverage:
        self._configure_coverage_export()

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
45
46
def log_message_on_container_start(self) -> str:
    return f"Tomodachi service: http://localhost:{self.edge_port}"

start 

start()
Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
48
49
50
51
52
53
54
55
56
57
58
59
def start(self) -> "TomodachiContainer":
    super().start()
    # Apart from HTTP healthcheck, we need to wait for "started service" log
    # to make sure messaging transport like AWS SNS SQS is also up and running.
    # It's started independently from HTTP transport.

    # Different service start messages depending on tomodachi version
    # tomodachi < 0.26.0 - "Started service"
    # tomodachi >= 0.26.0 - "started service successfully"
    # using (?i) to ignore case to support both versions
    LogMessageWaitStrategy("(?i)started service").with_startup_timeout(10).wait_until_ready(self)
    return self

stop 

stop()
Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
61
62
63
64
65
def stop(self) -> None:
    if self._export_coverage:
        with suppress(Exception):
            self._stop_container_and_copy_coverage_report()
    super().stop()

WireMockContainer 

WireMockContainer(
    image="wiremock/wiremock:latest",
    internal_port=8080,
    edge_port=None,
    mapping_stubs=None,
    mapping_files=None,
    *,
    verbose=False,
    disable_logging=False,
    **kwargs,
)

Bases: WebContainer

Source code in src/tomodachi_testcontainers/containers/wiremock.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def __init__(
    self,
    image: str = "wiremock/wiremock:latest",
    internal_port: int = 8080,
    edge_port: int | None = None,
    mapping_stubs: Path | None = None,
    mapping_files: Path | None = None,
    *,
    verbose: bool = False,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=internal_port,
        edge_port=edge_port,
        disable_logging=disable_logging,
        **kwargs,
    )

    mapping_stubs_env = os.getenv("WIREMOCK_TESTCONTAINER_MAPPING_STUBS")
    mapping_files_env = os.getenv("WIREMOCK_TESTCONTAINER_MAPPING_FILES")

    self.mapping_stubs = mapping_stubs or (Path(mapping_stubs_env) if mapping_stubs_env else None)
    self.mapping_files = mapping_files or (Path(mapping_files_env) if mapping_files_env else None)

    if verbose or os.getenv("WIREMOCK_TESTCONTAINER_VERBOSE"):
        self.with_command("--verbose")

MAPPING_STUBS_DIR class-attribute instance-attribute 

MAPPING_STUBS_DIR = Path('/home/wiremock/mappings/')

MAPPING_FILES_DIR class-attribute instance-attribute 

MAPPING_FILES_DIR = Path('/home/wiremock/__files/')

mapping_stubs instance-attribute 

mapping_stubs = mapping_stubs or (
    Path(mapping_stubs_env) if mapping_stubs_env else None
)

mapping_files instance-attribute 

mapping_files = mapping_files or (
    Path(mapping_files_env) if mapping_files_env else None
)

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
51
52
def log_message_on_container_start(self) -> str:
    return f"Wiremock admin: http://localhost:{self.edge_port}/__admin"

start 

start()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
54
55
56
57
58
def start(self) -> "WireMockContainer":
    super().start()
    LogMessageWaitStrategy("port:").with_startup_timeout(10).wait_until_ready(self)
    self.load_mappings_from_files()
    return self

load_mappings_from_files 

load_mappings_from_files()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
60
61
62
63
def load_mappings_from_files(self) -> None:
    self._copy_mapping_stubs_to_container()
    self._copy_mapping_files_to_container()
    self.reset_mappings()

delete_mappings 

delete_mappings()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
65
66
def delete_mappings(self) -> None:
    self.exec(["curl", "-X", "DELETE", f"http://localhost:{self.internal_port}/__admin/mappings"])

reset_mappings 

reset_mappings()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
68
69
def reset_mappings(self) -> None:
    self.exec(["curl", "-X", "POST", f"http://localhost:{self.internal_port}/__admin/mappings/reset"])

DatabaseContainer 

DatabaseContainer(
    image,
    internal_port,
    edge_port=None,
    disable_logging=False,
    **kwargs,
)

Bases: DockerContainer, ABC

Abstract class for relational database containers.

Source code in src/tomodachi_testcontainers/containers/common/database.py 
42
43
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    image: str,
    internal_port: int,
    edge_port: int | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(image, disable_logging=disable_logging, **kwargs)
    self.internal_port = internal_port
    self.edge_port = edge_port or get_available_port()
    self.with_bind_ports(internal_port, self.edge_port)

internal_port instance-attribute 

internal_port = internal_port

edge_port instance-attribute 

edge_port = edge_port or get_available_port()

drivername instance-attribute 

drivername

username instance-attribute 

username

password instance-attribute 

password

database instance-attribute 

database

get_internal_url 

get_internal_url()
Source code in src/tomodachi_testcontainers/containers/common/database.py 
55
56
57
58
59
60
61
62
63
def get_internal_url(self) -> DatabaseURL:
    return DatabaseURL(
        drivername=self.drivername,
        username=self.username,
        password=self.password,
        host=self.get_container_internal_ip(),
        port=self.internal_port,
        database=self.database,
    )

get_external_url 

get_external_url()
Source code in src/tomodachi_testcontainers/containers/common/database.py 
65
66
67
68
69
70
71
72
73
def get_external_url(self) -> DatabaseURL:
    return DatabaseURL(
        drivername=self.drivername,
        username=self.username,
        password=self.password,
        host=self.get_container_host_ip(),
        port=self.edge_port,
        database=self.database,
    )

start 

start(timeout=20.0)
Source code in src/tomodachi_testcontainers/containers/common/database.py 
75
76
77
78
def start(self, timeout: float = 20.0) -> "DatabaseContainer":
    super().start()
    wait_for_database_healthcheck(url=self.get_external_url(), timeout=timeout)
    return self

MySQLContainer 

MySQLContainer(
    image="mysql:9",
    internal_port=3306,
    edge_port=None,
    drivername=None,
    username=None,
    root_password=None,
    password=None,
    database=None,
    disable_logging=False,
    **kwargs,
)

Bases: DatabaseContainer

MySQL container.

Configuration environment variables (set on host machine):

  • MYSQL_DRIVERNAME - defaults to mysql+pymysql
  • MYSQL_USER - defaults to username
  • MYSQL_ROOT_PASSWORD - defaults to root
  • MYSQL_PASSWORD - defaults to password
  • MYSQL_DATABASE - defaults to db
Source code in src/tomodachi_testcontainers/containers/mysql.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(
    self,
    image: str = "mysql:9",
    internal_port: int = 3306,
    edge_port: int | None = None,
    drivername: str | None = None,
    username: str | None = None,
    root_password: str | None = None,
    password: str | None = None,
    database: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=internal_port,
        edge_port=edge_port,
        disable_logging=disable_logging,
        **kwargs,
    )

    self.drivername = drivername or os.getenv("MYSQL_DRIVERNAME") or "mysql+pymysql"
    self.username = username or os.getenv("MYSQL_USER") or "username"
    self.root_password = root_password or os.getenv("MYSQL_ROOT_PASSWORD") or "root"
    self.password = password or os.getenv("MYSQL_PASSWORD") or "password"
    self.database = database or os.getenv("MYSQL_DATABASE") or "db"

    if self.username == "root":
        self.root_password = self.password

    self.with_env("MYSQL_USER", self.username)
    self.with_env("MYSQL_ROOT_PASSWORD", self.root_password)
    self.with_env("MYSQL_PASSWORD", self.password)
    self.with_env("MYSQL_DATABASE", self.database)

    # Do not flush data on disk to improve test container performance
    # https://pythonspeed.com/articles/faster-db-tests/
    self.with_command("--innodb_flush_method=O_DIRECT_NO_FSYNC")

drivername instance-attribute 

drivername = (
    drivername
    or getenv("MYSQL_DRIVERNAME")
    or "mysql+pymysql"
)

username instance-attribute 

username = username or getenv('MYSQL_USER') or 'username'

root_password instance-attribute 

root_password = (
    root_password or getenv("MYSQL_ROOT_PASSWORD") or "root"
)

password instance-attribute 

password = (
    password or getenv("MYSQL_PASSWORD") or "password"
)

database instance-attribute 

database = database or getenv('MYSQL_DATABASE') or 'db'

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/mysql.py 
63
64
def log_message_on_container_start(self) -> str:
    return f"MySQL started: {self.get_external_url()}"

PostgreSQLContainer 

PostgreSQLContainer(
    image="postgres:18",
    internal_port=5432,
    edge_port=None,
    drivername=None,
    username=None,
    password=None,
    database=None,
    disable_logging=False,
    **kwargs,
)

Bases: DatabaseContainer

PostgreSQL container.

Configuration environment variables (set on host machine):

  • POSTGRES_DRIVERNAME - defaults to postgresql+psycopg
  • POSTGRES_USER - defaults to username
  • POSTGRES_PASSWORD - defaults to password
  • POSTGRES_DB - defaults to db
Source code in src/tomodachi_testcontainers/containers/postgres.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def __init__(
    self,
    image: str = "postgres:18",
    internal_port: int = 5432,
    edge_port: int | None = None,
    drivername: str | None = None,
    username: str | None = None,
    password: str | None = None,
    database: str | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(
        image,
        internal_port=internal_port,
        edge_port=edge_port,
        disable_logging=disable_logging,
        **kwargs,
    )

    self.drivername = drivername or os.getenv("POSTGRES_DRIVERNAME") or "postgresql+psycopg"
    self.username = username or os.getenv("POSTGRES_USER") or "username"
    self.password = password or os.getenv("POSTGRES_PASSWORD") or "password"
    self.database = database or os.getenv("POSTGRES_DB") or "db"

    self.with_env("POSTGRES_USER", self.username)
    self.with_env("POSTGRES_PASSWORD", self.password)
    self.with_env("POSTGRES_DB", self.database)

    # Do not flush data on disk to improve test container performance
    # https://pythonspeed.com/articles/faster-db-tests/
    self.with_command("-c fsync=off")

drivername instance-attribute 

drivername = (
    drivername
    or getenv("POSTGRES_DRIVERNAME")
    or "postgresql+psycopg"
)

username instance-attribute 

username = username or getenv("POSTGRES_USER") or "username"

password instance-attribute 

password = (
    password or getenv("POSTGRES_PASSWORD") or "password"
)

database instance-attribute 

database = database or getenv('POSTGRES_DB') or 'db'

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/postgres.py 
56
57
def log_message_on_container_start(self) -> str:
    return f"PostgreSQL started: {self.get_external_url()}"

SFTPContainer 

SFTPContainer(
    image="atmoz/sftp:latest",
    internal_port=22,
    edge_port=None,
    disable_logging=False,
    **kwargs,
)

Bases: DockerContainer

Source code in src/tomodachi_testcontainers/containers/sftp.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def __init__(
    self,
    image: str = "atmoz/sftp:latest",
    internal_port: int = 22,
    edge_port: int | None = None,
    disable_logging: bool = False,
    **kwargs: Any,
) -> None:
    super().__init__(image, disable_logging=disable_logging, **kwargs)
    self.internal_port = internal_port
    self.edge_port = edge_port or get_available_port()
    self.with_bind_ports(self.internal_port, self.edge_port)

    self.with_command("userpass:pass:1001 userssh::1002")

    self.authorized_private_key = asyncssh.generate_private_key("ssh-ed25519")
    self.authorized_public_key = self.authorized_private_key.export_public_key().decode()

internal_port instance-attribute 

internal_port = internal_port

edge_port instance-attribute 

edge_port = edge_port or get_available_port()

authorized_private_key instance-attribute 

authorized_private_key = generate_private_key("ssh-ed25519")

authorized_public_key instance-attribute 

authorized_public_key = decode()

log_message_on_container_start 

log_message_on_container_start()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
35
36
def log_message_on_container_start(self) -> str:
    return f"SFTP server running on port: {self.edge_port}"

get_internal_conn_details 

get_internal_conn_details()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
38
39
40
def get_internal_conn_details(self) -> ConnectionDetails:
    host = self.get_container_internal_ip()
    return ConnectionDetails(host=host, port=self.internal_port)

get_external_conn_details 

get_external_conn_details()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
42
43
44
def get_external_conn_details(self) -> ConnectionDetails:
    host = self.get_container_host_ip()
    return ConnectionDetails(host=host, port=self.edge_port)

get_host_public_key 

get_host_public_key()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
46
47
48
def get_host_public_key(self) -> str:
    _, output = self.exec("bash -c 'cat /etc/ssh/ssh_host_ed25519_key.pub'")
    return bytes(output).decode().strip()

get_internal_known_host 

get_internal_known_host()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
50
51
52
53
def get_internal_known_host(self) -> str:
    internal = self.get_internal_conn_details()
    public_key = self.get_host_public_key()
    return f"{internal.host} {public_key}"

get_external_known_host 

get_external_known_host()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
55
56
57
58
def get_external_known_host(self) -> str:
    external = self.get_external_conn_details()
    public_key = self.get_host_public_key()
    return f"{external.host} {public_key}"

get_known_hosts 

get_known_hosts()
Source code in src/tomodachi_testcontainers/containers/sftp.py 
60
61
62
63
64
def get_known_hosts(self) -> asyncssh.SSHKnownHosts:
    known_hosts = asyncssh.SSHKnownHosts()
    for known_host in (self.get_internal_known_host(), self.get_external_known_host()):
        known_hosts.load(known_host)
    return known_hosts

add_authorized_key 

add_authorized_key(username, uid, gid, public_key)
Source code in src/tomodachi_testcontainers/containers/sftp.py 
66
67
68
69
70
71
72
def add_authorized_key(self, username: str, uid: str, gid: str, public_key: str) -> None:
    self.exec(f"bash -c 'mkdir -p /home/{username}/.ssh'")
    self.exec(f"bash -c 'chmod 700 /home/{username}/.ssh'")
    self.exec(f"bash -c 'touch /home/{username}/.ssh/authorized_keys'")
    self.exec(f"bash -c 'chmod 600 /home/{username}/.ssh/authorized_keys'")
    self.exec(f"bash -c 'echo \"{public_key}\" >> /home/{username}/.ssh/authorized_keys'")
    self.exec(f"bash -c 'chown -R {uid}:{gid} /home/{username}/.ssh'")

start 

start(timeout=10.0)
Source code in src/tomodachi_testcontainers/containers/sftp.py 
74
75
76
77
78
79
80
81
82
83
84
85
def start(self, timeout: float = 10.0) -> "SFTPContainer":
    super().start()
    LogMessageWaitStrategy(r"Server listening on").with_startup_timeout(int(timeout)).wait_until_ready(self)

    self.exec("bash -c 'mkdir /home/userpass/upload && chown -R 1001:1001 /home/userpass/upload'")
    self.exec("bash -c 'mkdir /home/userpass/download && chown -R 1001:1001 /home/userpass/download'")
    self.exec("bash -c 'mkdir /home/userssh/upload && chown -R 1002:1002 /home/userssh/upload'")
    self.exec("bash -c 'mkdir /home/userssh/download && chown -R 1002:1002 /home/userssh/download'")

    self.add_authorized_key(username="userssh", uid="1002", gid="1002", public_key=self.authorized_public_key)

    return self