Skip to content

Tomodachi Testcontainers

tomodachi_testcontainers

DockerContainer 

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

Bases: DockerContainer, ABC

Abstract class for generic Docker containers.

PARAMETER DESCRIPTION
*args

TYPE: Any DEFAULT: ()

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/common/container.py 
28
29
30
31
32
33
34
35
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 
54
55
56
@abc.abstractmethod
def log_message_on_container_start(self) -> str:
    """Returns a message that will be logged when the container starts."""

get_container_host_ip 

get_container_host_ip()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
58
59
60
61
62
63
64
65
def get_container_host_ip(self) -> str:
    if not (host := self.get_docker_client().host()):
        return "localhost"
    if inside_container() and not os.getenv("DOCKER_HOST"):
        if (gateway_ip := self.get_container_gateway_ip()) == host:
            return self.get_container_internal_ip()
        return gateway_ip
    return host

get_container_internal_ip 

get_container_internal_ip()
Source code in src/tomodachi_testcontainers/containers/common/container.py 
67
68
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 
70
71
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 
73
74
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 
76
77
78
79
80
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 
82
83
84
85
86
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 
88
89
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.

PARAMETER DESCRIPTION
dockerfile

TYPE: Optional[Path] DEFAULT: None

context

TYPE: Optional[Path] DEFAULT: None

target

TYPE: Optional[str] DEFAULT: None

docker_client_kwargs

TYPE: Optional[Dict] DEFAULT: None

remove_image_on_exit

TYPE: bool DEFAULT: True

Source code in src/tomodachi_testcontainers/containers/common/image.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
def __init__(
    self,
    dockerfile: Optional[Path] = None,
    context: Optional[Path] = None,
    target: Optional[str] = None,
    docker_client_kwargs: Optional[Dict] = 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.

PARAMETER DESCRIPTION
image

TYPE: str

internal_port

TYPE: int

edge_port

TYPE: Optional[int] DEFAULT: None

http_healthcheck_path

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/common/web.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    image: str,
    internal_port: int,
    edge_port: Optional[int] = None,
    http_healthcheck_path: Optional[str] = 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 
37
38
39
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 
41
42
43
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 
45
46
47
48
49
50
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
PARAMETER DESCRIPTION
dynamo_endpoint

TYPE: str

image

TYPE: str DEFAULT: 'aaronshaf/dynamodb-admin:latest'

internal_port

TYPE: int DEFAULT: 8001

edge_port

TYPE: Optional[int] DEFAULT: None

region_name

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

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: Optional[int] = None,
    region_name: Optional[str] = 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:3', 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
PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'localstack/localstack:3'

internal_port

TYPE: int DEFAULT: 4566

edge_port

TYPE: Optional[int] DEFAULT: None

region_name

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/localstack.py 
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
def __init__(
    self,
    image: str = "localstack/localstack:3",
    internal_port: int = 4566,
    edge_port: Optional[int] = None,
    region_name: Optional[str] = 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 
51
52
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 
54
55
56
57
58
59
60
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 
62
63
64
65
def start(self) -> "LocalStackContainer":
    super().start()
    wait_for_logs(self, r"Ready\.\n", timeout=10.0)
    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
PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'minio/minio:latest'

s3_api_internal_port

TYPE: int DEFAULT: 9000

s3_api_edge_port

TYPE: Optional[int] DEFAULT: None

console_internal_port

TYPE: int DEFAULT: 9001

console_edge_port

TYPE: Optional[int] DEFAULT: None

region_name

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/minio.py 
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
44
45
46
47
48
49
50
51
52
53
def __init__(
    self,
    image: str = "minio/minio:latest",
    s3_api_internal_port: int = 9000,
    s3_api_edge_port: Optional[int] = None,
    console_internal_port: int = 9001,
    console_edge_port: Optional[int] = None,
    region_name: Optional[str] = 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 
55
56
57
58
59
60
def log_message_on_container_start(self) -> str:
    return (
        "Minio started: "
        f"S3-API: http://localhost:{self.s3_api_edge_port}/; "  # noqa: E702
        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 
62
63
64
65
66
67
68
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 
70
71
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
PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'motoserver/moto:latest'

internal_port

TYPE: int DEFAULT: 5000

edge_port

TYPE: Optional[int] DEFAULT: None

region_name

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/moto.py 
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
def __init__(
    self,
    image: str = "motoserver/moto:latest",
    internal_port: int = 5000,
    edge_port: Optional[int] = None,
    region_name: Optional[str] = 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 
53
54
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 
56
57
58
59
60
61
62
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 
64
65
66
67
def start(self) -> "MotoContainer":
    super().start()
    wait_for_logs(self, "Running on all addresses", timeout=10.0)
    return self

reset_moto 

reset_moto()
Source code in src/tomodachi_testcontainers/containers/moto.py 
69
70
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
PARAMETER DESCRIPTION
image

TYPE: str

internal_port

TYPE: int DEFAULT: 9700

edge_port

TYPE: Optional[int] DEFAULT: None

http_healthcheck_path

TYPE: Optional[str] DEFAULT: None

export_coverage

TYPE: bool DEFAULT: False

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
def __init__(
    self,
    image: str,
    internal_port: int = 9700,
    edge_port: Optional[int] = None,
    http_healthcheck_path: Optional[str] = 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 
44
45
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 
47
48
49
50
51
52
53
54
55
56
57
58
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
    wait_for_logs(self, "(?i)started service", timeout=10.0)
    return self

stop 

stop()
Source code in src/tomodachi_testcontainers/containers/tomodachi.py 
60
61
62
63
64
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

PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'wiremock/wiremock:latest'

internal_port

TYPE: int DEFAULT: 8080

edge_port

TYPE: Optional[int] DEFAULT: None

mapping_stubs

TYPE: Optional[Path] DEFAULT: None

mapping_files

TYPE: Optional[Path] DEFAULT: None

verbose

TYPE: bool DEFAULT: False

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/wiremock.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
def __init__(
    self,
    image: str = "wiremock/wiremock:latest",
    internal_port: int = 8080,
    edge_port: Optional[int] = None,
    mapping_stubs: Optional[Path] = None,
    mapping_files: Optional[Path] = 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 
50
51
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 
53
54
55
56
57
def start(self) -> "WireMockContainer":
    super().start()
    wait_for_logs(self, "port:", timeout=10.0)
    self.load_mappings_from_files()
    return self

load_mappings_from_files 

load_mappings_from_files()
Source code in src/tomodachi_testcontainers/containers/wiremock.py 
59
60
61
62
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 
64
65
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 
67
68
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.

PARAMETER DESCRIPTION
image

TYPE: str

internal_port

TYPE: int

edge_port

TYPE: Optional[int] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/common/database.py 
41
42
43
44
45
46
47
48
49
50
51
52
def __init__(
    self,
    image: str,
    internal_port: int,
    edge_port: Optional[int] = 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 
54
55
56
57
58
59
60
61
62
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 
64
65
66
67
68
69
70
71
72
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)
PARAMETER DESCRIPTION
timeout

TYPE: float DEFAULT: 20.0

Source code in src/tomodachi_testcontainers/containers/common/database.py 
74
75
76
77
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:8', 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
PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'mysql:8'

internal_port

TYPE: int DEFAULT: 3306

edge_port

TYPE: Optional[int] DEFAULT: None

drivername

TYPE: Optional[str] DEFAULT: None

username

TYPE: Optional[str] DEFAULT: None

root_password

TYPE: Optional[str] DEFAULT: None

password

TYPE: Optional[str] DEFAULT: None

database

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

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:8",
    internal_port: int = 3306,
    edge_port: Optional[int] = None,
    drivername: Optional[str] = None,
    username: Optional[str] = None,
    root_password: Optional[str] = None,
    password: Optional[str] = None,
    database: Optional[str] = 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:16', 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
PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'postgres:16'

internal_port

TYPE: int DEFAULT: 5432

edge_port

TYPE: Optional[int] DEFAULT: None

drivername

TYPE: Optional[str] DEFAULT: None

username

TYPE: Optional[str] DEFAULT: None

password

TYPE: Optional[str] DEFAULT: None

database

TYPE: Optional[str] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

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:16",
    internal_port: int = 5432,
    edge_port: Optional[int] = None,
    drivername: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
    database: Optional[str] = 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

PARAMETER DESCRIPTION
image

TYPE: str DEFAULT: 'atmoz/sftp:latest'

internal_port

TYPE: int DEFAULT: 22

edge_port

TYPE: Optional[int] DEFAULT: None

disable_logging

TYPE: bool DEFAULT: False

**kwargs

TYPE: Any DEFAULT: {}

Source code in src/tomodachi_testcontainers/containers/sftp.py 
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
def __init__(
    self,
    image: str = "atmoz/sftp:latest",
    internal_port: int = 22,
    edge_port: Optional[int] = 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 
31
32
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 
34
35
36
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 
38
39
40
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 
42
43
44
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 
46
47
48
49
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 
51
52
53
54
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 
56
57
58
59
60
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)
PARAMETER DESCRIPTION
username

TYPE: str

uid

TYPE: str

gid

TYPE: str

public_key

TYPE: str

Source code in src/tomodachi_testcontainers/containers/sftp.py 
62
63
64
65
66
67
68
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)
PARAMETER DESCRIPTION
timeout

TYPE: float DEFAULT: 10.0

Source code in src/tomodachi_testcontainers/containers/sftp.py 
70
71
72
73
74
75
76
77
78
79
80
81
def start(self, timeout: float = 10.0) -> "SFTPContainer":
    super().start()
    wait_for_logs(self, r"Server listening on", timeout=timeout)

    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