Mocking boto3 S3 클라이언트 메소드 Python

Mocking boto3 S3 클라이언트 메소드 Python

나는 예외를 던지기 위해 bot to3 s3 클라이언트 객체의 단일 설법을 조롱하려고 합니다.하지만 이 수업이 정상적으로 진행되려면 다른 모든 방법이 필요합니다.

이는 upload_part_copy를 수행하는 동안 오류가 발생할 때 단일 예외 테스트를 테스트할 수 있기 때문입니다.

첫 번째 시도

import boto3
from mock import patch

with patch('botocore.client.S3.upload_part_copy', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

그러나 이 경우 다음과 같은 오류가 발생합니다.

ImportError: No module named S3

두 번째 시도

botocore.client.py 소스 코드를 보고 나서 나는 그것이 영리한 무언가를 하고 있다는 것과 그 방법을 발견했습니다.upload_part_copy존재하지 않습니다.전화가 온 것 같습니다.BaseClient._make_api_call대신에 저는 그것을 조롱하려고 했습니다.

import boto3
from mock import patch

with patch('botocore.client.BaseClient._make_api_call', side_effect=Exception('Error Uploading')) as mock:
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

이 경우 예외가 발생합니다.바로get_object내가 피하고 싶은 것.

내가 어떻게 예외를 둘 수 있는지에 대한 아이디어는.upload_part_copy방법?

Botocore에는 이러한 용도로만 사용할 수 있는 클라이언트 스텁이 있습니다. 즉, 문서입니다.

다음은 오류를 입력하는 예입니다.

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
stubber.add_client_error('upload_part_copy')
stubber.activate()

# Will raise a ClientError
client.upload_part_copy()

여기에 정상적인 반응을 넣는 예가 있습니다.또한 이제 스텁버를 컨텍스트에서 사용할 수 있습니다.스터버는 제공된 응답이 서비스가 실제로 반환하는 응답과 일치하는지 확인합니다.이것은 완벽하지는 않지만, 완전히 무의미한 응답을 삽입하는 것을 방지할 수 있습니다.

import boto3
from botocore.stub import Stubber

client = boto3.client('s3')
stubber = Stubber(client)
list_buckets_response = {
    "Owner": {
        "DisplayName": "name",
        "ID": "EXAMPLE123"
    },
    "Buckets": [{
        "CreationDate": "2016-05-25T16:55:48.000Z",
        "Name": "foo"
    }]
}
expected_params = {}
stubber.add_response('list_buckets', list_buckets_response, expected_params)

with stubber:
    response = client.list_buckets()

assert response == list_buckets_response

제가 여기에 글을 올리자마자 해결책을 생각해냈습니다.여기 그것이 도움이 되기를 바랍니다 :)

import botocore
from botocore.exceptions import ClientError
from mock import patch
import boto3

orig = botocore.client.BaseClient._make_api_call

def mock_make_api_call(self, operation_name, kwarg):
    if operation_name == 'UploadPartCopy':
        parsed_response = {'Error': {'Code': '500', 'Message': 'Error Uploading'}}
        raise ClientError(parsed_response, operation_name)
    return orig(self, operation_name, kwarg)

with patch('botocore.client.BaseClient._make_api_call', new=mock_make_api_call):
    client = boto3.client('s3')
    # Should return actual result
    o = client.get_object(Bucket='my-bucket', Key='my-key')
    # Should return mocked exception
    e = client.upload_part_copy()

Jordan Philips도 bootocore.stub을 사용하여 훌륭한 솔루션을 게시했습니다.스터버 클래스.더 깨끗한 솔루션을 사용하는 동안 특정 작업을 조롱할 수 없었습니다.

둘 다 사용하지 않으려면moto또는 bootocore stubber(stubber는 AWS API 엔드포인트에 HTTP 요청을 수행하는 것을 방해하지 않음), 보다 자세한 unittest.mock 방법을 사용할 수 있습니다.

foo/bar.py

import boto3

def my_bar_function():
    client = boto3.client('s3')
    buckets = client.list_buckets()
    ...

bar_test.py

import unittest
from unittest import mock


class MyTest(unittest.TestCase):

     @mock.patch('foo.bar.boto3.client')
     def test_that_bar_works(self, mock_s3_client):
         self.assertTrue(mock_s3_client.return_value.list_buckets.call_count == 1)

여기 클라이언트 = boto3.client("client") api 호출을 위장하는 데 사용할 수 있는 간단한 파이썬 유닛 테스트의 예가 있습니다...

import boto3 

class MyAWSModule():
    def __init__(self):
        client = boto3.client('ec2')
        tags = client.describe_tags(DryRun=False)


class TestMyAWSModule(unittest.TestCase):
    @mock.patch("boto3.client.describe_tags")
    @mock.patch("boto3.client")
    def test_open_file_with_existing_file(self, mock_boto_client, mock_describe_tags):
        mock_describe_tags.return_value = mock_get_tags_response
        my_aws_module = MyAWSModule()
    
        mock_boto_client.assert_call_once('ec2')
        mock_describe_tags.assert_call_once_with(DryRun=False)

mock_get_tags_response = {
    'Tags': [
        {
            'ResourceId': 'string',
            'ResourceType': 'customer-gateway',
            'Key': 'string',
            'Value': 'string'
        },
    ],
'NextToken': 'string'
}

그게 도움이 되길 바랍니다.

간단하게 모토를 사용하면 어떨까요?

매우 편리한 장식기가 함께 제공됩니다.

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

나는 조롱해야 했습니다.boto3통합 테스트를 위해 고객을 찾았는데 조금 고통스러웠습니다!내가 가지고 있던 문제는moto지원하지 않음KMS아주 좋아요, 하지만 저는 제 자신의 조롱을 다시 쓰고 싶지 않았어요.S3양동이그래서 저는 모든 답의 형태를 만들었습니다.또한 그것은 전 세계적으로 작동하는데 꽤 멋집니다!

나는 그것을 2개의 파일로 설정했습니다.

는 첫번는째입니다.aws_mock.py의 .KMS boto3고객.

from unittest.mock import MagicMock

import boto3
from moto import mock_s3

# `create_key` response
create_resp = { ... }

# `generate_data_key` response
generate_resp = { ... }

# `decrypt` response
decrypt_resp = { ... }

def client(*args, **kwargs):
    if args[0] == 's3':
        s3_mock = mock_s3()
        s3_mock.start()
        mock_client = boto3.client(*args, **kwargs)

    else:
        mock_client = boto3.client(*args, **kwargs)

        if args[0] == 'kms':
            mock_client.create_key = MagicMock(return_value=create_resp)
            mock_client.generate_data_key = MagicMock(return_value=generate_resp)
            mock_client.decrypt = MagicMock(return_value=decrypt_resp)

    return mock_client

두 번째는 실제 테스트 모듈입니다.끝내꾸나라고 부르자.test_my_module.py는 코생니다습략했의 my_module테스트 중인 기능도 포함됩니다.을 그것들부르자을라고 부르자foo,bar기능들.

from unittest.mock import patch

import aws_mock
import my_module

@patch('my_module.boto3')
def test_my_module(boto3):
    # Some prep work for the mock mode
    boto3.client = aws_mock.client

    conn = boto3.client('s3')
    conn.create_bucket(Bucket='my-bucket')

    # Actual testing
    resp = my_module.foo()
    assert(resp == 'Valid')

    resp = my_module.bar()
    assert(resp != 'Not Valid')

    # Etc, etc, etc...

, 저는 그것을 되었습니다.moto사용자가 자격 증명 및 지역과 같은 환경 변수를 설정하지 않는 한 행복하지 않았습니다.실제 자격 증명일 필요는 없지만 설정해야 합니다.당신이 이것을 읽을 때쯤이면 고쳐질 가능성이 있습니다!하지만 필요할 때를 대비해 코드가 몇 개 있어요, 이번에는 셸 코드!

export AWS_ACCESS_KEY_ID='foo'
export AWS_SECRET_ACCESS_KEY='bar'
export AWS_DEFAULT_REGION='us-east-1'

나는 그것이 아마도 가장 예쁜 코드가 아니라는 것을 알지만 만약 당신이 보편적인 것을 찾고 있다면 그것은 꽤 잘 작동할 것입니다!

은 다음과 . boto boto와 함께 사용됩니다.pytestㅠㅠ 저는 제 프로젝트에서 'mturk'만 사용하고 있습니다.

후를 적용하는 의 하나였습니다.boto3.client미리 생성된 클라이언트를 반환하는 함수를 사용합니다.

@pytest.fixture(scope='session')
def patched_boto_client():
    my_client = boto3.client('mturk')

    def my_client_func(*args, **kwargs):
        return my_client

    with patch('bowels.of.project.other_module.boto3.client', my_client_func):
        yield my_client_func


def test_create_hit(patched_boto_client):    
    client = patched_boto_client()
    stubber = Stubber(client)
    stubber.add_response('create_hit_type', {'my_response':'is_great'})
    stubber.add_response('create_hit_with_hit_type', {'my_other_response':'is_greater'})
    stubber.activate()

    import bowels.of.project # this module imports `other_module`
    bowels.of.project.create_hit_function_that_calls_a_function_in_other_module_which_invokes_boto3_dot_client_at_some_point()

저는 또한 boto가 실수로 시스템의 다른 자격 증명 집합을 가져오지 않도록 더미 AWS 자격 증명을 설정하는 다른 고정 장치를 정의합니다.저는 문자 그대로 'foo'와 'bar'를 테스트에 대한 제 학점으로 설정했습니다. 그것은 수정이 아닙니다.

중요한 것은AWS_PROFILE그렇지 않으면 boto가 해당 프로필을 찾으러 가기 때문에 env 설정이 해제됩니다.

@pytest.fixture(scope='session')
def setup_env():
    os.environ['AWS_ACCESS_KEY_ID'] = 'foo'
    os.environ['AWS_SECRET_ACCESS_KEY'] = 'bar'
    os.environ.pop('AWS_PROFILE', None)

에 ' 그런다지정다니합음다니합지정▁and▁specify'를 지정합니다.setup_env 화가 나서.usefixtures모든 테스트 실행에 사용되는 항목입니다.

사용 사례가 약간 달랐는데, 이 경우에는 클라이언트가 설치됩니다.setup()클래스의 메서드는 대화 중인 AWS 서비스의 항목을 나열하는 것과 같은 몇 가지 작업을 수행합니다(나의 경우 Connect).위의 많은 접근법들은 제대로 작동하지 않았습니다. 그래서 이것이 미래의 구글 사용자들을 위한 제 작업 버전입니다.

모든 것이 제대로 작동하려면 다음과 같이 해야 했습니다.

에서 (시험 중)src/flow_manager.py):

class FlowManager:
    client: botocore.client.BaseClient
    
    def setup(self):
        self.client = boto3.client('connect')
    
    def set_instance(self):
        response = self.client.list_instances()
        ... do stuff ....

파일테스트파일(▁(파일▁in▁file()에.tests/unit/test_flow_manager.py):

@mock.patch('src.flow_manager.boto3.client')
def test_set_instance(self, mock_client):
    expected = 'bar'
    instance_list = {'alias': 'foo', 'id': 'bar'}
    mock_client.list_instances.return_value = instance_list
    actual = flow_manager.FlowManager("", "", "", "", 'foo')
    actual.client = mock_client
    actual.set_instance()
    self.assertEqual(expected, actual.instance_id)

이 답변에 대한 코드를 관련 비트로 잘라냈습니다.

언급URL : https://stackoverflow.com/questions/37143597/mocking-boto3-s3-client-method-python