How to mock AWS S3 with aiobotocore

Question

I have a project that uses aiohttp and aiobotocore to work with resources in AWS. I am trying to test class that works with AWS S3 and I am using moto to mock AWS. Mocking works just fine with examples that use synchronous code (example from moto docs)

import boto3
from moto import mock_s3


class MyModel(object):
    def __init__(self, name, value):
        self.name = name
        self.value = value

    def save(self):
        s3 = boto3.client('s3', region_name='us-east-1')
        s3.put_object(Bucket='mybucket', Key=self.name, Body=self.value)


def test_my_model_save():
    with mock_s3():
        conn = boto3.resource('s3', region_name='us-east-1')
        conn.create_bucket(Bucket='mybucket')

        model_instance = MyModel('steve', 'is awesome')
        model_instance.save()
        body = conn.Object('mybucket', 'steve').get()['Body'].read().decode("utf-8")

        assert body == 'is awesome'

However, after rewriting this to use aiobotocore mocking does not work - it connects to real AWS S3 in my example.

import aiobotocore
import asyncio

import boto3
from moto import mock_s3


class MyModel(object):
    def __init__(self, name, value):
        self.name = name
        self.value = value

    async def save(self, loop):
        session = aiobotocore.get_session(loop=loop)
        s3 = session.create_client('s3', region_name='us-east-1')
        await s3.put_object(Bucket='mybucket', Key=self.name, Body=self.value)


def test_my_model_save():
    with mock_s3():
        conn = boto3.resource('s3', region_name='us-east-1')
        conn.create_bucket(Bucket='mybucket')
        loop = asyncio.get_event_loop()

        model_instance = MyModel('steve', 'is awesome')
        loop.run_until_complete(model_instance.save(loop=loop))
        body = conn.Object('mybucket', 'steve').get()['Body'].read().decode("utf-8")

        assert body == 'is awesome'

So my assumption here is that moto does not work properly with aiobotocore. How can I effectively mock AWS resources if my source code looks like in the second example?

Answer 1

Mocks from moto don't work because they use a synchronous API. However, you can start moto server and configure aiobotocore to connect to this test server. Take a look on aiobotocore tests for inspiration.