AWS S3をaiobotocoreでモックする方法

Question

AWSのリソースで作業するためにaiohttpとaiobotocoreを使用するプロジェクトがあります。私はAWS S3で動作するクラスをテストしようとしています。私はAWSを模擬するためにモトを使用しています。モッキングは同期コード（モトドキュメントからの例）

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' 

を使用する例でうまく動作しますが、aiobotocoreモックを使用するために、これを書き換えた後は動作しません - それは私の例では、実際のAWS S3に接続します。

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' 

ここで私の前提は、モトがaiobotocoreで正しく動作しないということです。 2番目の例のようにソースコードが見える場合、AWSリソースを効果的に模擬するにはどうすればよいですか？

Answer 1

モックはmotoからは同期APIを使用しているため動作しません。 しかし、モトサーバーを起動し、このテストサーバーに接続するようにaiobotocoreを構成することができます。 インスピレーションのためのaiobotocoreテストを見てください。

Answer 2

ここpytestなしaiobotocoreからmock_server.pyです：

# Initially from https://raw.githubusercontent.com/aio-libs/aiobotocore/master/tests/mock_server.py 

import shutil 
import signal 
import subprocess as sp 
import sys 
import time 
import requests 


_proxy_bypass = { 
    "http": None, 
    "https": None, 
} 


def start_service(service_name, host, port): 
    moto_svr_path = shutil.which("moto_server") 
    args = [sys.executable, moto_svr_path, service_name, "-H", host, 
      "-p", str(port)] 
    process = sp.Popen(args, stdin=sp.PIPE, stdout=sp.PIPE, stderr=sp.DEVNULL) 
    url = "http://{host}:{port}".format(host=host, port=port) 

    for _ in range(30): 
     if process.poll() is not None: 
      break 

     try: 
      # we need to bypass the proxies due to monkeypatches 
      requests.get(url, timeout=0.1, proxies=_proxy_bypass) 
      break 
     except requests.exceptions.RequestException: 
      time.sleep(0.1) 
    else: 
     stop_process(process) 
     raise AssertionError("Can not start service: {}".format(service_name)) 

    return process 


def stop_process(process, timeout=20): 
    try: 
     process.send_signal(signal.SIGTERM) 
     process.communicate(timeout=timeout/2) 
    except sp.TimeoutExpired: 
     process.kill() 
     outs, errors = process.communicate(timeout=timeout/2) 
     exit_code = process.returncode 
     msg = "Child process finished {} not in clean way: {} {}" \ 
      .format(exit_code, outs, errors) 
     raise RuntimeError(msg)