Skip to content

AioMoto Fixtures

asyncio fixtures

AWS test fixtures

This test suite uses a large suite of moto mocks for the AWS batch infrastructure. These infrastructure mocks are derived from the moto test suite for testing the batch client. The test infrastructure should be used according to the moto license. That license overrides any global license applied to my aio_aws project.

.. seealso::

- https://github.com/spulec/moto/pull/1197/files
- https://github.com/spulec/moto/blob/master/tests/test_batch/test_batch.py

aio_aws_batch_client(aio_aws_session, aio_aws_batch_server) async

AWS Async Client for AioMotoService("batch")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
213
214
215
216
217
218
219
220
221
222
@pytest_asyncio.fixture
async def aio_aws_batch_client(aio_aws_session, aio_aws_batch_server):
    """
    AWS Async Client for AioMotoService("batch")
    """
    async with aio_aws_session.create_client(
        "batch", endpoint_url=aio_aws_batch_server.endpoint_url
    ) as client:
        yield client
    moto_service_reset("batch")

aio_aws_batch_clients(aio_aws_batch_client, aio_aws_ec2_client, aio_aws_ecs_client, aio_aws_iam_client, aio_aws_logs_client, aws_region) async

Async Clients for AWS Batch Infrastructure

Source code in pytest_aiomoto/aiomoto_fixtures.py 
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
@pytest_asyncio.fixture
async def aio_aws_batch_clients(
    aio_aws_batch_client,
    aio_aws_ec2_client,
    aio_aws_ecs_client,
    aio_aws_iam_client,
    aio_aws_logs_client,
    aws_region,
) -> AioAwsBatchClients:
    """
    Async Clients for AWS Batch Infrastructure
    """
    yield AioAwsBatchClients(
        batch=aio_aws_batch_client,
        ec2=aio_aws_ec2_client,
        ecs=aio_aws_ecs_client,
        iam=aio_aws_iam_client,
        logs=aio_aws_logs_client,
        region=aws_region,
    )

aio_aws_batch_infrastructure(aio_aws_batch_clients, compute_env_name, job_queue_name, job_definition_name, iam_role_name) async

AWS Batch Infrastructure with Async Clients

Source code in pytest_aiomoto/aiomoto_fixtures.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
@pytest_asyncio.fixture
async def aio_aws_batch_infrastructure(
    aio_aws_batch_clients: AioAwsBatchClients,
    compute_env_name: str,
    job_queue_name: str,
    job_definition_name: str,
    iam_role_name: str,
) -> AioAwsBatchInfrastructure:
    """
    AWS Batch Infrastructure with Async Clients
    """
    aws_region = aio_aws_batch_clients.region
    async with aio_batch_infrastructure(
        aio_aws_batch_clients=aio_aws_batch_clients,
        aws_region=aws_region,
        compute_env_name=compute_env_name,
        job_queue_name=job_queue_name,
        job_definition_name=job_definition_name,
        iam_role_name=iam_role_name
    ) as aio_batch_resources:
        yield aio_batch_resources

aio_aws_batch_server() async

AioMotoService("batch")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
44
45
46
47
48
49
50
51
52
@pytest_asyncio.fixture
async def aio_aws_batch_server() -> AioMotoService:
    """
    AioMotoService("batch")
    """
    async with AioMotoService("batch") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_cloudformation_server() async

AioMotoService("cloudformation")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
55
56
57
58
59
60
61
62
63
@pytest_asyncio.fixture
async def aio_aws_cloudformation_server() -> AioMotoService:
    """
    AioMotoService("cloudformation")
    """
    async with AioMotoService("cloudformation") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_dynamodb2_server() async

AioMotoService("dynamodb2")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
 99
100
101
102
103
104
105
106
107
@pytest_asyncio.fixture
async def aio_aws_dynamodb2_server() -> AioMotoService:
    """
    AioMotoService("dynamodb2")
    """
    async with AioMotoService("dynamodb2") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_ec2_client(aio_aws_session, aio_aws_ec2_server) async

AWS Async Client for AioMotoService("ec2")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
225
226
227
228
229
230
231
232
233
234
@pytest_asyncio.fixture
async def aio_aws_ec2_client(aio_aws_session, aio_aws_ec2_server):
    """
    AWS Async Client for AioMotoService("ec2")
    """
    async with aio_aws_session.create_client(
        "ec2", endpoint_url=aio_aws_ec2_server.endpoint_url
    ) as client:
        yield client
    moto_service_reset("ec2")

aio_aws_ec2_server() async

AioMotoService("ec2")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
66
67
68
69
70
71
72
73
74
@pytest_asyncio.fixture
async def aio_aws_ec2_server() -> AioMotoService:
    """
    AioMotoService("ec2")
    """
    async with AioMotoService("ec2") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_ecs_client(aio_aws_session, aio_aws_ecs_server) async

AWS Async Client for AioMotoService("ecs")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
237
238
239
240
241
242
243
244
245
246
@pytest_asyncio.fixture
async def aio_aws_ecs_client(aio_aws_session, aio_aws_ecs_server):
    """
    AWS Async Client for AioMotoService("ecs")
    """
    async with aio_aws_session.create_client(
        "ecs", endpoint_url=aio_aws_ecs_server.endpoint_url
    ) as client:
        yield client
    moto_service_reset("ecs")

aio_aws_ecs_server() async

AioMotoService("ecs")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
77
78
79
80
81
82
83
84
85
@pytest_asyncio.fixture
async def aio_aws_ecs_server() -> AioMotoService:
    """
    AioMotoService("ecs")
    """
    async with AioMotoService("ecs") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_iam_client(aio_aws_session, aio_aws_iam_server) async

AWS Async Client for AioMotoService("iam")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
249
250
251
252
253
254
255
256
257
258
259
@pytest_asyncio.fixture
async def aio_aws_iam_client(aio_aws_session, aio_aws_iam_server):
    """
    AWS Async Client for AioMotoService("iam")
    """
    async with aio_aws_session.create_client(
        "iam", endpoint_url=aio_aws_iam_server.endpoint_url
    ) as client:
        client.meta.config.region_name = "aws-global"  # not AWS_REGION
        yield client
    moto_service_reset("iam")

aio_aws_iam_server() async

AioMotoService("iam")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
88
89
90
91
92
93
94
95
96
@pytest_asyncio.fixture
async def aio_aws_iam_server() -> AioMotoService:
    """
    AioMotoService("iam")
    """
    async with AioMotoService("iam") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_lambda_client(aio_aws_session, aio_aws_lambda_server) async

AWS Async Client for AioMotoService("lambda")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
262
263
264
265
266
267
268
269
270
271
@pytest_asyncio.fixture
async def aio_aws_lambda_client(aio_aws_session, aio_aws_lambda_server):
    """
    AWS Async Client for AioMotoService("lambda")
    """
    async with aio_aws_session.create_client(
        "lambda", endpoint_url=aio_aws_lambda_server.endpoint_url
    ) as client:
        yield client
    moto_service_reset("lambda")

aio_aws_lambda_server() async

AioMotoService("lambda")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
110
111
112
113
114
115
116
117
118
@pytest_asyncio.fixture
async def aio_aws_lambda_server() -> AioMotoService:
    """
    AioMotoService("lambda")
    """
    async with AioMotoService("lambda") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_logs_client(aio_aws_session, aio_aws_logs_server) async

AWS Async Client for AioMotoService("logs")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
274
275
276
277
278
279
280
281
282
283
@pytest_asyncio.fixture
async def aio_aws_logs_client(aio_aws_session, aio_aws_logs_server):
    """
    AWS Async Client for AioMotoService("logs")
    """
    async with aio_aws_session.create_client(
        "logs", endpoint_url=aio_aws_logs_server.endpoint_url
    ) as client:
        yield client
    moto_service_reset("logs")

aio_aws_logs_server() async

AioMotoService("logs")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
121
122
123
124
125
126
127
128
129
130
@pytest_asyncio.fixture
async def aio_aws_logs_server() -> AioMotoService:
    """
    AioMotoService("logs")
    """
    # cloud watch logs
    async with AioMotoService("logs") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_s3_client(aio_aws_session, aio_aws_s3_server, mocker) async

AWS Async Client for AioMotoService("s3")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
@pytest_asyncio.fixture
async def aio_aws_s3_client(aio_aws_session, aio_aws_s3_server, mocker):
    """
    AWS Async Client for AioMotoService("s3")
    """
    async with aio_aws_session.create_client(
        "s3", endpoint_url=aio_aws_s3_server.endpoint_url
    ) as client:
        # TODO: find a way to apply this method mock only for creating "s3" clients;
        # mocker.patch(
        #     "aiobotocore.session.AioSession.create_client",
        #     side_effect=partial(
        #         aio_aws_session.create_client, "s3", endpoint_url=aio_aws_s3_server.endpoint_url
        #     )
        # )
        yield client
    moto_service_reset("s3")

aio_aws_s3_server() async

AioMotoService("s3")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
133
134
135
136
137
138
139
140
141
@pytest_asyncio.fixture
async def aio_aws_s3_server() -> AioMotoService:
    """
    AioMotoService("s3")
    """
    async with AioMotoService("s3") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_session(aws_credentials, aws_region, event_loop)

An AioSession configured with credentials for moto services

Source code in pytest_aiomoto/aiomoto_fixtures.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
@pytest_asyncio.fixture
def aio_aws_session(aws_credentials, aws_region, event_loop) -> AioSession:
    """
    An AioSession configured with credentials for moto services
    """
    # pytest-asyncio provides and manages the `event_loop`
    # and it should be set as the default loop for this session
    assert event_loop  # but it's not == asyncio.get_event_loop() ?

    session = get_session()
    session.user_agent_name = "aiomoto"

    assert session.get_default_client_config() is None
    aioconfig = AioConfig(max_pool_connections=1, region_name=aws_region)

    # Note: tried to use proxies for the aiobotocore.endpoint, to replace
    #      'https://batch.us-west-2.amazonaws.com/v1/describejobqueues', but
    #      the moto.server does not behave as a proxy server.  Leaving this
    #      here for the record to avoid trying to do it again sometime later.
    # proxies = {
    #     'http': os.getenv("HTTP_PROXY", "http://127.0.0.1:5000/moto-api/"),
    #     'https': os.getenv("HTTPS_PROXY", "http://127.0.0.1:5000/moto-api/"),
    # }
    # assert aioconfig.proxies is None
    # aioconfig.proxies = proxies

    session.set_default_client_config(aioconfig)
    assert session.get_default_client_config() == aioconfig

    session.set_credentials(AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
    session.set_debug_logger(logger_name="aiomoto")

    yield session

aio_aws_sns_server() async

AioMotoService("sns")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
144
145
146
147
148
149
150
151
152
@pytest_asyncio.fixture
async def aio_aws_sns_server() -> AioMotoService:
    """
    AioMotoService("sns")
    """
    async with AioMotoService("sns") as svc:
        svc.reset()
        yield svc
        svc.reset()

aio_aws_sqs_server() async

AioMotoService("sqs")

Source code in pytest_aiomoto/aiomoto_fixtures.py 
155
156
157
158
159
160
161
162
163
@pytest_asyncio.fixture
async def aio_aws_sqs_server() -> AioMotoService:
    """
    AioMotoService("sqs")
    """
    async with AioMotoService("sqs") as svc:
        svc.reset()
        yield svc
        svc.reset()

Test Asyncio AWS Lambda

aws_lambda_func(aws_lambda_zip, lambda_iam_role, aws_lambda_python_runtime, aio_aws_lambda_client) async

Create an AWS Lambda function and return the function name 'lambda_dev'.

Source code in pytest_aiomoto/aiomoto_lambda.py 
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@pytest_asyncio.fixture
async def aws_lambda_func(
    aws_lambda_zip, lambda_iam_role, aws_lambda_python_runtime, aio_aws_lambda_client
) -> str:
    """
    Create an AWS Lambda function and return the function name 'lambda_dev'.
    """

    func = "lambda_dev"

    response = await aio_aws_lambda_client.create_function(
        FunctionName=func,
        Runtime=aws_lambda_python_runtime,
        Handler="lambda_function.lambda_handler",
        Code={"ZipFile": aws_lambda_zip},
        Role=lambda_iam_role,
        Description="lambda_dev function",
        Timeout=10,
        MemorySize=128,
        Publish=True,
    )
    assert response_success(response)

    return func

aws_lambda_python_runtime()

The python3.8 runtime for AWS Lambda.

Source code in pytest_aiomoto/aiomoto_lambda.py 
39
40
41
42
43
44
@pytest.fixture
def aws_lambda_python_runtime() -> str:
    """
    The python3.8 runtime for AWS Lambda.
    """
    return "python3.8"

aws_lambda_zip()

A ZIP archive to create an AWS Lambda function from the :py:func:aws_lambda_src function.

Source code in pytest_aiomoto/aiomoto_lambda.py 
30
31
32
33
34
35
36
@pytest.fixture
def aws_lambda_zip() -> bytes:
    """
    A ZIP archive to create an AWS Lambda function
    from the :py:func:`aws_lambda_src` function.
    """
    return zip_lambda(aws_lambda_src())

lambda_iam_role(aio_aws_iam_client) async

An IAM role to create an AWS Lambda function

Source code in pytest_aiomoto/aiomoto_lambda.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@pytest_asyncio.fixture
async def lambda_iam_role(aio_aws_iam_client):
    """
    An IAM role to create an AWS Lambda function
    """
    try:
        response = await aio_aws_iam_client.get_role(RoleName="my-role")
        return response["Role"]["Arn"]
    except botocore.client.ClientError:
        response = await aio_aws_iam_client.create_role(
            RoleName="my-role",
            AssumeRolePolicyDocument="some policy",
            Path="/my-path/",
        )
        return response["Role"]["Arn"]

aio_s3_bucket_name(aio_s3_uuid)

A valid S3 bucket name :return: str for the bucket component of 's3://{bucket}/{key}'

Source code in pytest_aiomoto/aiomoto_s3.py 
29
30
31
32
33
34
@pytest.fixture
def aio_s3_bucket_name(aio_s3_uuid) -> str:
    """A valid S3 bucket name
    :return: str for the bucket component of 's3://{bucket}/{key}'
    """
    return f"aio-moto-bucket-{aio_s3_uuid}"

aio_s3_key(aio_s3_key_path, aio_s3_key_file)

A valid S3 key composed of a key_path and a key_file The key component of 's3://{bucket}/{key}' that is composed of '{key_path}/{key_file}'; the key does not begin or end with any delimiters (e.g. '/') :return: str for the key component of 's3://{bucket}/{key}'

Source code in pytest_aiomoto/aiomoto_s3.py 
57
58
59
60
61
62
63
64
@pytest.fixture
def aio_s3_key(aio_s3_key_path, aio_s3_key_file) -> str:
    """A valid S3 key composed of a key_path and a key_file
    The key component of 's3://{bucket}/{key}' that is composed of '{key_path}/{key_file}';
    the key does not begin or end with any delimiters (e.g. '/')
    :return: str for the key component of 's3://{bucket}/{key}'
    """
    return f"{aio_s3_key_path}/{aio_s3_key_file}"

aio_s3_key_file()

A valid S3 key name that is also a file name The key component of 's3://{bucket}/{key}' that is composed of '{key_file}'; the key does not begin or end with any delimiters (e.g. '/') :return: str for the key component of 's3://{bucket}/{key}'

Source code in pytest_aiomoto/aiomoto_s3.py 
47
48
49
50
51
52
53
54
@pytest.fixture
def aio_s3_key_file() -> str:
    """A valid S3 key name that is also a file name
    The key component of 's3://{bucket}/{key}' that is composed of '{key_file}';
    the key does not begin or end with any delimiters (e.g. '/')
    :return: str for the key component of 's3://{bucket}/{key}'
    """
    return "aio_s3_file_test.txt"

aio_s3_key_path()

A valid S3 key name that is not a file name, it's like a directory The key component of 's3://{bucket}/{key}' that is composed of '{key_path}'; the key does not begin or end with any delimiters (e.g. '/') :return: str for the key component of 's3://{bucket}/{key}'

Source code in pytest_aiomoto/aiomoto_s3.py 
37
38
39
40
41
42
43
44
@pytest.fixture
def aio_s3_key_path() -> str:
    """A valid S3 key name that is not a file name, it's like a directory
    The key component of 's3://{bucket}/{key}' that is composed of '{key_path}';
    the key does not begin or end with any delimiters (e.g. '/')
    :return: str for the key component of 's3://{bucket}/{key}'
    """
    return "aio_s3_key_path"

aio_s3_object_text()

s3 object data: 's3 test object text '

Source code in pytest_aiomoto/aiomoto_s3.py 
75
76
77
78
@pytest.fixture
def aio_s3_object_text() -> str:
    """s3 object data: 's3 test object text\n'"""
    return "aio-s3 test object text\n"

aio_s3_object_uri(aio_s3_bucket_name, aio_s3_key, aio_s3_uri, aio_s3_object_text, aio_s3_bucket, aio_aws_s3_client) async

s3 object data is PUT to the aio_s3_uri

Source code in pytest_aiomoto/aiomoto_s3.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
@pytest_asyncio.fixture
async def aio_s3_object_uri(
    aio_s3_bucket_name, aio_s3_key, aio_s3_uri, aio_s3_object_text, aio_s3_bucket, aio_aws_s3_client
) -> str:
    """s3 object data is PUT to the aio_s3_uri"""
    resp = await aio_aws_s3_client.put_object(
        Bucket=aio_s3_bucket_name,
        Key=aio_s3_key,
        Body=aio_s3_object_text,
        ACL="public-read-write",
    )
    assert response_success(resp)
    resp = await aio_aws_s3_client.head_object(Bucket=aio_s3_bucket_name, Key=aio_s3_key)
    assert response_success(resp)

    return aio_s3_uri

aio_s3_uri(aio_s3_bucket_name, aio_s3_key)

A valid S3 URI comprised of 's3://{bucket}/{key}' :return: str

Source code in pytest_aiomoto/aiomoto_s3.py 
67
68
69
70
71
72
@pytest.fixture
def aio_s3_uri(aio_s3_bucket_name, aio_s3_key) -> str:
    """A valid S3 URI comprised of 's3://{bucket}/{key}'
    :return: str
    """
    return f"s3://{aio_s3_bucket_name}/{aio_s3_key}"

aio_s3_uuid()

A UUID for S3 artifacts

Source code in pytest_aiomoto/aiomoto_s3.py 
23
24
25
26
@pytest.fixture
def aio_s3_uuid() -> str:
    """A UUID for S3 artifacts"""
    return str(uuid.uuid4())

aio_s3fs(aio_aws_session, aio_aws_s3_server, mocker, monkeypatch)

The aio_s3fs fixture mocks generation of any aiobotocore.client.S3 object so that it calls a localhost server provided by an asyncio version of the moto-server. The aio_s3fs fixture is simply a context to apply the mock, it is not intended to be a replacement for s3fs. Just add the aio_s3fs fixture to a test function and then use s3fs as normal.

Source code in pytest_aiomoto/aiomoto_s3fs.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
52
53
54
55
56
57
@pytest.fixture()
def aio_s3fs(
    aio_aws_session,
    aio_aws_s3_server,
    mocker,
    monkeypatch,
):
    """
    The `aio_s3fs` fixture mocks generation of any `aiobotocore.client.S3` object
    so that it calls a localhost server provided by an asyncio version of the moto-server.
    The `aio_s3fs` fixture is simply a context to apply the mock, it is not intended to be a
    replacement for `s3fs`.  Just add the `aio_s3fs` fixture to a test function and then use
    `s3fs` as normal.
    """
    try:
        import s3fs

        try:
            monkeypatch.setenv("S3_ENDPOINT_URL", aio_aws_s3_server.endpoint_url)
            # TODO: find a way to apply this method mock only for creating "s3" clients;
            aio_client_patch = mocker.patch(
                "aiobotocore.session.AioSession.create_client",
                side_effect=partial(
                    aio_aws_session.create_client, "s3", endpoint_url=aio_aws_s3_server.endpoint_url
                )
            )
            s3fs.S3FileSystem.clear_instance_cache()

            yield

            assert aio_client_patch.call_count > 0

        finally:
            s3fs.S3FileSystem.clear_instance_cache()
            monkeypatch.delenv("S3_ENDPOINT_URL", raising=False)

    except ImportError:
        pytest.skip("The extra pytest_aiomoto[s3fs] dependency is missing")