如何使用aiobotocore模擬AWS S3

我有一個項目使用aiohttp和aiobotocore來處理AWS中的資源。我正在測試適用於AWS S3的類，並且我正在使用moto來模擬AWS。懲戒工作得與使用同步代碼（例如，從摩托文檔）的實例細如何使用aiobotocore模擬AWS S3

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。如果我的源代碼如第二個示例中所示，我如何有效地模擬AWS資源？

來源

2017-09-26 Belerafon

來自moto的嘲笑不起作用，因爲它們使用的是同步API。但您可以啓動moto服務器並配置aiobotocore以連接到此測試服務器。看看aiobotocore測試的靈感。

來源

2017-09-26 11:37:25

下面是aiobotocore mock_server.py沒有pytest：

# 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)

來源

2017-10-27 19:15:09 Javier

如何使用aiobotocore模擬AWS S3

回答

相關問題