python-3.x 如何使用aiobotocore模拟AWS S3

g6baxovj 于 2023-11-20 发布在 Python

关注(0)|答案(7)|浏览(135)

我有一个使用aiohtp和aiobotocore来处理AWS中资源的项目。我正在尝试测试与AWS S3一起工作的类，我正在使用moto来模拟AWS。Mocking在使用同步代码的示例中工作得很好（来自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'

字符串
然而，在重写此以使用aiobotocore mocking之后，它就不起作用了--在我的示例中，它连接到了真实的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'

型
所以我在这里的假设是moto不能与aiobotocore正常工作。如果我的源代码看起来像第二个例子中的那样，我如何有效地模拟AWS资源？

python-3.x

来源：https://stackoverflow.com/questions/46420709/how-to-mock-aws-s3-with-aiobotocore

7条答案

按热度按时间

k97glaaz1#

来自moto的模拟不起作用，因为它们使用同步API。但是，您可以启动moto服务器并配置aiobotocore连接到此测试服务器。Take a look on aiobotocore tests作为灵感。

赞(0）回复(0）举报 2023-11-20

2g32fytz2#

使用AWS的编译器应该可以做到这一点。下面是我在一个tornado应用程序中为AWS读取操作所做的操作：

import aiobotocore
from botocore.stub import Stubber
from tornado.testing import AsyncTestCase
from aiobotocore.response import StreamingBody

class RawStream(io.BytesIO):

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        pass

    async def read(self, n):
        return super().read(n)

class S3TestCase(AsyncTestCase):

    def setUp(self):
        super().setUp()
        session = aiobotocore.get_session()
        self.client = session.create_client("s3", region_name="AWS_S3_REGION",
                                    aws_secret_access_key="AWS_SECRET_ACCESS_KEY",
                                    aws_access_key_id="AWS_ACCESS_KEY_ID")

    @tornado.testing.gen_test
    def test_read(self):
        stubber = Stubber(self.client)
        stubber.add_response("get_object",
                          {"Body": StreamingBody(raw_stream=RawStream(self.binary_content), content_length=128),
                          "ContentLength": 128},
                         expected_params={"Bucket": "AWS_S3_BUCKET",
                                          "Key": "filename"})
        stubber.activate()

        response = await client.get_object(Bucket="AWS_S3_BUCKET", Key="filename")

字符串
应该是类似的写操作。希望这将引导你在正确的方向。
更多信息：https://botocore.amazonaws.com/v1/documentation/api/latest/reference/stubber.html

赞(0）回复(0）举报 2023-11-20

pes8fvy93#

我认为塞巴斯蒂安Brestins的答案应该是公认的。我将发布这个新的答案，因为自从它发布以来，有些事情发生了变化，例如，python 3.8现在支持python测试用例，aioboto3客户端现在是上下文管理器。
使用Python 3.8的最小示例如下所示：

from unittest import IsolatedAsyncioTestCase

import aioboto3
from botocore.stub import Stubber

class Test(IsolatedAsyncioTestCase):

    async def asyncSetUp(self):
        self._s3_client = await aioboto3.client('s3').__aenter__()
        self._s3_stub = Stubber(self._s3_client)

    async def asyncTearDown(self):
        await self._s3_client.__aexit__(None, None, None)

    async def test_case(self):
        self._s3_stub.add_response(
            "get_object",
            {"Body": "content"},
            expected_params={"Bucket": "AWS_S3_BUCKET", "Key": "filename"}
        )
        self._s3_stub.activate()

        response = await self._s3_client.get_object(Bucket="AWS_S3_BUCKET", Key="filename")

        self.assertEquals(response, "content")

字符串

赞(0）回复(0）举报 2023-11-20

bvuwiixz4#

不幸的是，这不是一个完整的答案，但有一个pull request添加了这个功能，已经开放了6个月：https://github.com/aio-libs/aiobotocore/pull/766
当我处理类似的问题时，我已经为sync对象手工编写了“Pencc” Package 器。

赞(0）回复(0）举报 2023-11-20

f1tvaqid5#

下面是aiobotocore中没有pytest的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)

字符串

赞(0）回复(0）举报 2023-11-20

n8ghc7c16#

我们可以使用moto[server]创建S3服务器，然后使用它创建一个类似于aioboto3的pytest fixture

@pytest.yield_fixture(scope='session')
def s3_server():
    host = 'localhost'
    port = 5002
    url = 'http://{host}:{port}'.format(host=host, port=port)
    process = start_service('s3', host, port)
    yield url
    stop_process(process)

字符串
然后patch('aiobotocore.AioSession.create_client') return_value和aiobotocore.get_session().create_client('s3', region_name='us-east-1', end_point_url=s3_server)如下所示

async with aiobotocore.get_session().create_client('s3', region_name='us-east-1', end_point_url=s3_server) as client:
    with patch('aiobotocore.AioSession.create_client') as mock:
        mock.return_value = client
        # Test your code

型

赞(0）回复(0）举报 2023-11-20

yzckvree7#

下面是一个fixture，它使用moto包为您完成此任务：

pip install moto[server]

个字符
你可以在测试中包含这个fixture，它将使aws为这个测试“工作”。

赞(0）回复(0）举报 2023-11-20

我来回答

python-3.x 如何使用aiobotocore模拟AWS S3

7条答案

相关问题

热门标签

最新问答