Skip to content

Fixtures

tomodachi_testcontainers.fixtures

Pytest fixtures.

testcontainer_image 

testcontainer_image()

Builds a Docker image from a Dockerfile located in the current working directory and returns an image ID.

The Docker image is removed on test session end.

Configuration environment variables (set on host machine):

  • TESTCONTAINER_IMAGE_ID - use given Image ID for creating a container.
  • TESTCONTAINER_DOCKERFILE_PATH - override path to the Dockerfile for building Docker image.
  • TESTCONTAINER_DOCKER_BUILD_CONTEXT - override Docker build context.
  • TESTCONTAINER_DOCKER_BUILD_TARGET - override Docker build target.
YIELDS DESCRIPTION
str
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/containers.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@pytest.fixture(scope="session")
def testcontainer_image() -> Generator[str, None, None]:
    """Builds a Docker image from a Dockerfile located in the current working directory and returns an image ID.

    The Docker image is removed on test session end.

    Configuration environment variables (set on host machine):

    - `TESTCONTAINER_IMAGE_ID` - use given Image ID for creating a container.
    - `TESTCONTAINER_DOCKERFILE_PATH` - override path to the Dockerfile for building Docker image.
    - `TESTCONTAINER_DOCKER_BUILD_CONTEXT` - override Docker build context.
    - `TESTCONTAINER_DOCKER_BUILD_TARGET` - override Docker build target.
    """
    if image_id := os.getenv("TESTCONTAINER_IMAGE_ID"):
        yield image_id
    else:
        with EphemeralDockerImage(
            dockerfile=(Path(v) if (v := os.getenv("TESTCONTAINER_DOCKERFILE_PATH")) else None),
            context=(Path(v) if (v := os.getenv("TESTCONTAINER_DOCKER_BUILD_CONTEXT")) else None),
            target=os.getenv("TESTCONTAINER_DOCKER_BUILD_TARGET"),
            # Don't remove the image on teardown if it's used by other pytest-xdist workers in parallel.
            # The image will be eventually removed by the 'master' worker that exits last.
            remove_image_on_exit=not os.getenv("PYTEST_XDIST_WORKER"),
        ) as image:
            yield str(image.id)

localstack_container 

localstack_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
19
20
21
22
23
24
25
@pytest.fixture(scope="session")
def localstack_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("LOCALSTACK_TESTCONTAINER_IMAGE_ID", "localstack/localstack:3")
    disable_logging = bool(os.getenv("LOCALSTACK_TESTCONTAINER_DISABLE_LOGGING")) or False

    with LocalStackContainer(image, disable_logging=disable_logging) as container:
        yield container

localstack_dynamodb_client async 

localstack_dynamodb_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
37
38
39
40
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_dynamodb_client(localstack_container: LocalStackContainer) -> AsyncGenerator[DynamoDBClient, None]:
    async with get_session().create_client("dynamodb", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_iam_client async 

localstack_iam_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
43
44
45
46
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_iam_client(localstack_container: LocalStackContainer) -> AsyncGenerator[IAMClient, None]:
    async with get_session().create_client("iam", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_lambda_client async 

localstack_lambda_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
49
50
51
52
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_lambda_client(localstack_container: LocalStackContainer) -> AsyncGenerator[LambdaClient, None]:
    async with get_session().create_client("lambda", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_s3_client async 

localstack_s3_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
55
56
57
58
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_s3_client(localstack_container: LocalStackContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_sns_client async 

localstack_sns_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
61
62
63
64
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_sns_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SNSClient, None]:
    async with get_session().create_client("sns", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_snssqs_tc 

localstack_snssqs_tc(localstack_sns_client, localstack_sqs_client)
PARAMETER DESCRIPTION
localstack_sns_client

TYPE: SNSClient

localstack_sqs_client

TYPE: SQSClient

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
79
80
81
@pytest.fixture(scope="session")
def localstack_snssqs_tc(localstack_sns_client: SNSClient, localstack_sqs_client: SQSClient) -> SNSSQSTestClient:
    return SNSSQSTestClient(localstack_sns_client, localstack_sqs_client)

localstack_sqs_client async 

localstack_sqs_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
67
68
69
70
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_sqs_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SQSClient, None]:
    async with get_session().create_client("sqs", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_ssm_client async 

localstack_ssm_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
73
74
75
76
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_ssm_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SSMClient, None]:
    async with get_session().create_client("ssm", **localstack_container.get_aws_client_config()) as c:
        yield c

restart_localstack_container_on_teardown 

restart_localstack_container_on_teardown(localstack_container)

Restarts the LocalStack container after each test.

PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

YIELDS DESCRIPTION
None
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
28
29
30
31
32
33
34
@pytest.fixture
def restart_localstack_container_on_teardown(  # noqa: PT004
    localstack_container: LocalStackContainer,
) -> Generator[None, None, None]:
    """Restarts the LocalStack container after each test."""
    yield
    localstack_container.restart()

minio_container 

minio_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/minio.py 
12
13
14
15
16
17
18
@pytest.fixture(scope="session")
def minio_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MINIO_TESTCONTAINER_IMAGE_ID", "minio/minio:latest")
    disable_logging = bool(os.getenv("MINIO_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MinioContainer(image, disable_logging=disable_logging) as container:
        yield container

minio_s3_client async 

minio_s3_client(minio_container)
PARAMETER DESCRIPTION
minio_container

TYPE: MinioContainer

Source code in src/tomodachi_testcontainers/fixtures/minio.py 
21
22
23
24
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def minio_s3_client(minio_container: MinioContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **minio_container.get_aws_client_config()) as c:
        yield c

moto_container 

moto_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/moto.py 
19
20
21
22
23
24
25
@pytest.fixture(scope="session")
def moto_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MOTO_TESTCONTAINER_IMAGE_ID", "motoserver/moto:latest")
    disable_logging = bool(os.getenv("MOTO_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MotoContainer(image, disable_logging=disable_logging) as container:
        yield container

moto_dynamodb_client async 

moto_dynamodb_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
35
36
37
38
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_dynamodb_client(moto_container: MotoContainer) -> AsyncGenerator[DynamoDBClient, None]:
    async with get_session().create_client("dynamodb", **moto_container.get_aws_client_config()) as c:
        yield c

moto_iam_client async 

moto_iam_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
41
42
43
44
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_iam_client(moto_container: MotoContainer) -> AsyncGenerator[IAMClient, None]:
    async with get_session().create_client("iam", **moto_container.get_aws_client_config()) as c:
        yield c

moto_lambda_client async 

moto_lambda_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
47
48
49
50
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_lambda_client(moto_container: MotoContainer) -> AsyncGenerator[LambdaClient, None]:
    async with get_session().create_client("lambda", **moto_container.get_aws_client_config()) as c:
        yield c

moto_s3_client async 

moto_s3_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
53
54
55
56
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_s3_client(moto_container: MotoContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **moto_container.get_aws_client_config()) as c:
        yield c

moto_sns_client async 

moto_sns_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
59
60
61
62
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_sns_client(moto_container: MotoContainer) -> AsyncGenerator[SNSClient, None]:
    async with get_session().create_client("sns", **moto_container.get_aws_client_config()) as c:
        yield c

moto_snssqs_tc 

moto_snssqs_tc(moto_sns_client, moto_sqs_client)
PARAMETER DESCRIPTION
moto_sns_client

TYPE: SNSClient

moto_sqs_client

TYPE: SQSClient

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
77
78
79
@pytest.fixture(scope="session")
def moto_snssqs_tc(moto_sns_client: SNSClient, moto_sqs_client: SQSClient) -> SNSSQSTestClient:
    return SNSSQSTestClient(moto_sns_client, moto_sqs_client)

moto_sqs_client async 

moto_sqs_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
65
66
67
68
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_sqs_client(moto_container: MotoContainer) -> AsyncGenerator[SQSClient, None]:
    async with get_session().create_client("sqs", **moto_container.get_aws_client_config()) as c:
        yield c

moto_ssm_client async 

moto_ssm_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
71
72
73
74
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_ssm_client(moto_container: MotoContainer) -> AsyncGenerator[SSMClient, None]:
    async with get_session().create_client("ssm", **moto_container.get_aws_client_config()) as c:
        yield c

reset_moto_container_on_teardown 

reset_moto_container_on_teardown(moto_container)

Removes all mocked resources from Moto after each test without restarting the container.

PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

YIELDS DESCRIPTION
None
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/moto.py 
28
29
30
31
32
@pytest.fixture
def reset_moto_container_on_teardown(moto_container: MotoContainer) -> Generator[None, None, None]:  # noqa: PT004
    """Removes all mocked resources from Moto after each test without restarting the container."""
    yield
    moto_container.reset_moto()

reset_wiremock_container_on_teardown 

reset_wiremock_container_on_teardown(wiremock_container)

Deletes all stub mappings from WireMock after each test.

PARAMETER DESCRIPTION
wiremock_container

TYPE: WireMockContainer

YIELDS DESCRIPTION
None
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/wiremock.py 
26
27
28
29
30
31
32
@pytest.fixture
def reset_wiremock_container_on_teardown(  # noqa: PT004
    wiremock_container: WireMockContainer,
) -> Generator[None, None, None]:
    """Deletes all stub mappings from WireMock after each test."""
    yield
    wiremock_container.delete_mappings()

wiremock_container 

wiremock_container()
YIELDS DESCRIPTION
WireMockContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/wiremock.py 
14
15
16
17
18
19
20
21
22
23
@pytest.fixture(scope="session")
def wiremock_container() -> Generator[WireMockContainer, None, None]:
    image = os.getenv("WIREMOCK_TESTCONTAINER_IMAGE_ID", "wiremock/wiremock:latest")
    disable_logging = bool(os.getenv("WIREMOCK_TESTCONTAINER_DISABLE_LOGGING")) or False

    with WireMockContainer(image, disable_logging=disable_logging) as container:
        container = cast(WireMockContainer, container)
        if WireMockConfig is not None:
            WireMockConfig.base_url = f"{container.get_external_url()}/__admin/"
        yield container

mysql_container 

mysql_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/mysql.py 
 9
10
11
12
13
14
15
@pytest.fixture(scope="session")
def mysql_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MYSQL_TESTCONTAINER_IMAGE_ID", "mysql:8")
    disable_logging = bool(os.getenv("MYSQL_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MySQLContainer(image, disable_logging=disable_logging) as container:
        yield container

postgres_container 

postgres_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/postgres.py 
11
12
13
14
15
16
17
@pytest.fixture(scope="session")
def postgres_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("POSTGRES_TESTCONTAINER_IMAGE_ID", "postgres:16")
    disable_logging = bool(os.getenv("POSTGRES_TESTCONTAINER_DISABLE_LOGGING")) or False

    with PostgreSQLContainer(image, disable_logging=disable_logging) as container:
        yield container

sftp_container 

sftp_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
13
14
15
16
17
18
19
@pytest.fixture(scope="session")
def sftp_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("SFTP_TESTCONTAINER_IMAGE_ID", "atmoz/sftp:latest")
    disable_logging = bool(os.getenv("SFTP_TESTCONTAINER_DISABLE_LOGGING")) or False

    with SFTPContainer(image, disable_logging=disable_logging) as container:
        yield container

userpass_sftp_client async 

userpass_sftp_client(sftp_container)
PARAMETER DESCRIPTION
sftp_container

TYPE: SFTPContainer

Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
22
23
24
25
26
27
28
29
30
31
32
33
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def userpass_sftp_client(sftp_container: SFTPContainer) -> AsyncGenerator[asyncssh.SFTPClient, None]:
    conn_details = sftp_container.get_external_conn_details()
    async with asyncssh.connect(
        host=conn_details.host,
        port=conn_details.port,
        username="userpass",
        password="pass",  # nosec: B106
        known_hosts=sftp_container.get_known_hosts(),
    ) as ssh_conn:
        async with ssh_conn.start_sftp_client() as sftp_client:
            yield sftp_client

userssh_sftp_client async 

userssh_sftp_client(sftp_container)
PARAMETER DESCRIPTION
sftp_container

TYPE: SFTPContainer

Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
36
37
38
39
40
41
42
43
44
45
46
47
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def userssh_sftp_client(sftp_container: SFTPContainer) -> AsyncGenerator[asyncssh.SFTPClient, None]:
    conn_details = sftp_container.get_external_conn_details()
    async with asyncssh.connect(
        host=conn_details.host,
        port=conn_details.port,
        username="userssh",
        client_keys=sftp_container.authorized_private_key,
        known_hosts=sftp_container.get_known_hosts(),
    ) as ssh_conn:
        async with ssh_conn.start_sftp_client() as sftp_client:
            yield sftp_client

tomodachi_testcontainers.fixtures.containers

testcontainer_image 

testcontainer_image()

Builds a Docker image from a Dockerfile located in the current working directory and returns an image ID.

The Docker image is removed on test session end.

Configuration environment variables (set on host machine):

  • TESTCONTAINER_IMAGE_ID - use given Image ID for creating a container.
  • TESTCONTAINER_DOCKERFILE_PATH - override path to the Dockerfile for building Docker image.
  • TESTCONTAINER_DOCKER_BUILD_CONTEXT - override Docker build context.
  • TESTCONTAINER_DOCKER_BUILD_TARGET - override Docker build target.
YIELDS DESCRIPTION
str
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/containers.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
@pytest.fixture(scope="session")
def testcontainer_image() -> Generator[str, None, None]:
    """Builds a Docker image from a Dockerfile located in the current working directory and returns an image ID.

    The Docker image is removed on test session end.

    Configuration environment variables (set on host machine):

    - `TESTCONTAINER_IMAGE_ID` - use given Image ID for creating a container.
    - `TESTCONTAINER_DOCKERFILE_PATH` - override path to the Dockerfile for building Docker image.
    - `TESTCONTAINER_DOCKER_BUILD_CONTEXT` - override Docker build context.
    - `TESTCONTAINER_DOCKER_BUILD_TARGET` - override Docker build target.
    """
    if image_id := os.getenv("TESTCONTAINER_IMAGE_ID"):
        yield image_id
    else:
        with EphemeralDockerImage(
            dockerfile=(Path(v) if (v := os.getenv("TESTCONTAINER_DOCKERFILE_PATH")) else None),
            context=(Path(v) if (v := os.getenv("TESTCONTAINER_DOCKER_BUILD_CONTEXT")) else None),
            target=os.getenv("TESTCONTAINER_DOCKER_BUILD_TARGET"),
            # Don't remove the image on teardown if it's used by other pytest-xdist workers in parallel.
            # The image will be eventually removed by the 'master' worker that exits last.
            remove_image_on_exit=not os.getenv("PYTEST_XDIST_WORKER"),
        ) as image:
            yield str(image.id)

tomodachi_testcontainers.fixtures.localstack

localstack_container 

localstack_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
19
20
21
22
23
24
25
@pytest.fixture(scope="session")
def localstack_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("LOCALSTACK_TESTCONTAINER_IMAGE_ID", "localstack/localstack:3")
    disable_logging = bool(os.getenv("LOCALSTACK_TESTCONTAINER_DISABLE_LOGGING")) or False

    with LocalStackContainer(image, disable_logging=disable_logging) as container:
        yield container

restart_localstack_container_on_teardown 

restart_localstack_container_on_teardown(localstack_container)

Restarts the LocalStack container after each test.

PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

YIELDS DESCRIPTION
None
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
28
29
30
31
32
33
34
@pytest.fixture
def restart_localstack_container_on_teardown(  # noqa: PT004
    localstack_container: LocalStackContainer,
) -> Generator[None, None, None]:
    """Restarts the LocalStack container after each test."""
    yield
    localstack_container.restart()

localstack_dynamodb_client async 

localstack_dynamodb_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
37
38
39
40
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_dynamodb_client(localstack_container: LocalStackContainer) -> AsyncGenerator[DynamoDBClient, None]:
    async with get_session().create_client("dynamodb", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_iam_client async 

localstack_iam_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
43
44
45
46
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_iam_client(localstack_container: LocalStackContainer) -> AsyncGenerator[IAMClient, None]:
    async with get_session().create_client("iam", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_lambda_client async 

localstack_lambda_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
49
50
51
52
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_lambda_client(localstack_container: LocalStackContainer) -> AsyncGenerator[LambdaClient, None]:
    async with get_session().create_client("lambda", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_s3_client async 

localstack_s3_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
55
56
57
58
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_s3_client(localstack_container: LocalStackContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_sns_client async 

localstack_sns_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
61
62
63
64
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_sns_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SNSClient, None]:
    async with get_session().create_client("sns", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_sqs_client async 

localstack_sqs_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
67
68
69
70
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_sqs_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SQSClient, None]:
    async with get_session().create_client("sqs", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_ssm_client async 

localstack_ssm_client(localstack_container)
PARAMETER DESCRIPTION
localstack_container

TYPE: LocalStackContainer

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
73
74
75
76
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def localstack_ssm_client(localstack_container: LocalStackContainer) -> AsyncGenerator[SSMClient, None]:
    async with get_session().create_client("ssm", **localstack_container.get_aws_client_config()) as c:
        yield c

localstack_snssqs_tc 

localstack_snssqs_tc(localstack_sns_client, localstack_sqs_client)
PARAMETER DESCRIPTION
localstack_sns_client

TYPE: SNSClient

localstack_sqs_client

TYPE: SQSClient

Source code in src/tomodachi_testcontainers/fixtures/localstack.py 
79
80
81
@pytest.fixture(scope="session")
def localstack_snssqs_tc(localstack_sns_client: SNSClient, localstack_sqs_client: SQSClient) -> SNSSQSTestClient:
    return SNSSQSTestClient(localstack_sns_client, localstack_sqs_client)

tomodachi_testcontainers.fixtures.minio

minio_container 

minio_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/minio.py 
12
13
14
15
16
17
18
@pytest.fixture(scope="session")
def minio_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MINIO_TESTCONTAINER_IMAGE_ID", "minio/minio:latest")
    disable_logging = bool(os.getenv("MINIO_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MinioContainer(image, disable_logging=disable_logging) as container:
        yield container

minio_s3_client async 

minio_s3_client(minio_container)
PARAMETER DESCRIPTION
minio_container

TYPE: MinioContainer

Source code in src/tomodachi_testcontainers/fixtures/minio.py 
21
22
23
24
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def minio_s3_client(minio_container: MinioContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **minio_container.get_aws_client_config()) as c:
        yield c

tomodachi_testcontainers.fixtures.moto

moto_container 

moto_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/moto.py 
19
20
21
22
23
24
25
@pytest.fixture(scope="session")
def moto_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MOTO_TESTCONTAINER_IMAGE_ID", "motoserver/moto:latest")
    disable_logging = bool(os.getenv("MOTO_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MotoContainer(image, disable_logging=disable_logging) as container:
        yield container

reset_moto_container_on_teardown 

reset_moto_container_on_teardown(moto_container)

Removes all mocked resources from Moto after each test without restarting the container.

PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

YIELDS DESCRIPTION
None
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/moto.py 
28
29
30
31
32
@pytest.fixture
def reset_moto_container_on_teardown(moto_container: MotoContainer) -> Generator[None, None, None]:  # noqa: PT004
    """Removes all mocked resources from Moto after each test without restarting the container."""
    yield
    moto_container.reset_moto()

moto_dynamodb_client async 

moto_dynamodb_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
35
36
37
38
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_dynamodb_client(moto_container: MotoContainer) -> AsyncGenerator[DynamoDBClient, None]:
    async with get_session().create_client("dynamodb", **moto_container.get_aws_client_config()) as c:
        yield c

moto_iam_client async 

moto_iam_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
41
42
43
44
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_iam_client(moto_container: MotoContainer) -> AsyncGenerator[IAMClient, None]:
    async with get_session().create_client("iam", **moto_container.get_aws_client_config()) as c:
        yield c

moto_lambda_client async 

moto_lambda_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
47
48
49
50
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_lambda_client(moto_container: MotoContainer) -> AsyncGenerator[LambdaClient, None]:
    async with get_session().create_client("lambda", **moto_container.get_aws_client_config()) as c:
        yield c

moto_s3_client async 

moto_s3_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
53
54
55
56
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_s3_client(moto_container: MotoContainer) -> AsyncGenerator[S3Client, None]:
    async with get_session().create_client("s3", **moto_container.get_aws_client_config()) as c:
        yield c

moto_sns_client async 

moto_sns_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
59
60
61
62
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_sns_client(moto_container: MotoContainer) -> AsyncGenerator[SNSClient, None]:
    async with get_session().create_client("sns", **moto_container.get_aws_client_config()) as c:
        yield c

moto_sqs_client async 

moto_sqs_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
65
66
67
68
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_sqs_client(moto_container: MotoContainer) -> AsyncGenerator[SQSClient, None]:
    async with get_session().create_client("sqs", **moto_container.get_aws_client_config()) as c:
        yield c

moto_ssm_client async 

moto_ssm_client(moto_container)
PARAMETER DESCRIPTION
moto_container

TYPE: MotoContainer

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
71
72
73
74
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def moto_ssm_client(moto_container: MotoContainer) -> AsyncGenerator[SSMClient, None]:
    async with get_session().create_client("ssm", **moto_container.get_aws_client_config()) as c:
        yield c

moto_snssqs_tc 

moto_snssqs_tc(moto_sns_client, moto_sqs_client)
PARAMETER DESCRIPTION
moto_sns_client

TYPE: SNSClient

moto_sqs_client

TYPE: SQSClient

Source code in src/tomodachi_testcontainers/fixtures/moto.py 
77
78
79
@pytest.fixture(scope="session")
def moto_snssqs_tc(moto_sns_client: SNSClient, moto_sqs_client: SQSClient) -> SNSSQSTestClient:
    return SNSSQSTestClient(moto_sns_client, moto_sqs_client)

tomodachi_testcontainers.fixtures.mysql

mysql_container 

mysql_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/mysql.py 
 9
10
11
12
13
14
15
@pytest.fixture(scope="session")
def mysql_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("MYSQL_TESTCONTAINER_IMAGE_ID", "mysql:8")
    disable_logging = bool(os.getenv("MYSQL_TESTCONTAINER_DISABLE_LOGGING")) or False

    with MySQLContainer(image, disable_logging=disable_logging) as container:
        yield container

tomodachi_testcontainers.fixtures.postgres

postgres_container 

postgres_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/postgres.py 
11
12
13
14
15
16
17
@pytest.fixture(scope="session")
def postgres_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("POSTGRES_TESTCONTAINER_IMAGE_ID", "postgres:16")
    disable_logging = bool(os.getenv("POSTGRES_TESTCONTAINER_DISABLE_LOGGING")) or False

    with PostgreSQLContainer(image, disable_logging=disable_logging) as container:
        yield container

tomodachi_testcontainers.fixtures.sftp

sftp_container 

sftp_container()
YIELDS DESCRIPTION
DockerContainer
RECEIVES DESCRIPTION
None
RETURNS DESCRIPTION
None
Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
13
14
15
16
17
18
19
@pytest.fixture(scope="session")
def sftp_container() -> Generator[DockerContainer, None, None]:
    image = os.getenv("SFTP_TESTCONTAINER_IMAGE_ID", "atmoz/sftp:latest")
    disable_logging = bool(os.getenv("SFTP_TESTCONTAINER_DISABLE_LOGGING")) or False

    with SFTPContainer(image, disable_logging=disable_logging) as container:
        yield container

userpass_sftp_client async 

userpass_sftp_client(sftp_container)
PARAMETER DESCRIPTION
sftp_container

TYPE: SFTPContainer

Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
22
23
24
25
26
27
28
29
30
31
32
33
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def userpass_sftp_client(sftp_container: SFTPContainer) -> AsyncGenerator[asyncssh.SFTPClient, None]:
    conn_details = sftp_container.get_external_conn_details()
    async with asyncssh.connect(
        host=conn_details.host,
        port=conn_details.port,
        username="userpass",
        password="pass",  # nosec: B106
        known_hosts=sftp_container.get_known_hosts(),
    ) as ssh_conn:
        async with ssh_conn.start_sftp_client() as sftp_client:
            yield sftp_client

userssh_sftp_client async 

userssh_sftp_client(sftp_container)
PARAMETER DESCRIPTION
sftp_container

TYPE: SFTPContainer

Source code in src/tomodachi_testcontainers/fixtures/sftp.py 
36
37
38
39
40
41
42
43
44
45
46
47
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def userssh_sftp_client(sftp_container: SFTPContainer) -> AsyncGenerator[asyncssh.SFTPClient, None]:
    conn_details = sftp_container.get_external_conn_details()
    async with asyncssh.connect(
        host=conn_details.host,
        port=conn_details.port,
        username="userssh",
        client_keys=sftp_container.authorized_private_key,
        known_hosts=sftp_container.get_known_hosts(),
    ) as ssh_conn:
        async with ssh_conn.start_sftp_client() as sftp_client:
            yield sftp_client