python मॉकिंग बोटो 3 एस 3 क्लाइंट विधि पायथन




mocking boto (6)

मैं एक अपवाद को फेंकने के लिए boto3 s3 क्लाइंट ऑब्जेक्ट से एक विलक्षण विधि का मजाक उड़ाने की कोशिश कर रहा हूं। लेकिन मुझे इस वर्ग को सामान्य रूप से काम करने के लिए अन्य सभी तरीकों की आवश्यकता है।

ऐसा तब होता है जब मैं एक singular Exception परीक्षण कर सकता हूं, जब एक 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 कॉल लगता है, इसलिए मैंने 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 पद्धति पर अपवाद कैसे फेंक सकता हूं?


बोटोकोर का एक क्लाइंट स्टबर है जिसका उपयोग आप इस उद्देश्य के लिए कर सकते हैं: docs

यहाँ एक त्रुटि डालने का एक उदाहरण है:

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

यहाँ मेरी परियोजना के pytest में इस्तेमाल किए गए एक बोटो क्लाइंट को 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()

मैं एक और स्थिरता को भी परिभाषित करता हूं जो डमी एव्स क्रेडिट को सेट करता है ताकि सिस्टम पर कुछ अन्य क्रेडेंशियल्स का बोटो गलती से न उठे। मैंने शाब्दिक रूप से 'फू' और 'बार' को परीक्षण के लिए अपनी साख के रूप में निर्धारित किया है - यह एक नया रूप नहीं है।

यह महत्वपूर्ण है कि AWS_PROFILE env AWS_PROFILE हो क्योंकि अन्यथा AWS_PROFILE उस प्रोफ़ाइल की तलाश में जाएगा।

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

और फिर मैं setup_env को एक पाइस्टेस्ट usefixtures एंट्री के रूप में निर्दिष्ट करता setup_env ताकि इसे हर टेस्ट रन के लिए उपयोग किया जाए।


यहाँ एक सरल अजगर का एक उदाहरण है जो नकली ग्राहक = boto3.client ('ec2') एपीआई कॉल के लिए इस्तेमाल किया जा सकता है ...

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.get_tags")
    @mock.patch("boto3.client")
    def test_open_file_with_existing_file(self, mock_boto_client, mock_describe_tags):
        mock_boto_client.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'
}

उम्मीद है कि मदद करता है।


मुझे कुछ एकीकरण परीक्षण के लिए boto3 क्लाइंट का मजाक boto3 पड़ा और यह थोड़ा दर्दनाक था! मेरे पास जो समस्या थी वह यह है कि moto KMS बहुत अच्छा समर्थन नहीं करता है, फिर भी मैं S3 बाल्टियों के लिए अपने स्वयं के नकली को फिर से लिखना नहीं चाहता था। इसलिए मैंने सभी उत्तरों का यह रूप बनाया। इसके अलावा यह विश्व स्तर पर काम करता है जो बहुत अच्छा है!

मैंने इसे 2 फाइलों के साथ सेटअप किया है।

पहले एक aws_mock.pyKMS boto3 मुझे कुछ पूर्वनिर्धारित प्रतिक्रियाएँ boto3 जो कि लाइव boto3 क्लाइंट से आई 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'

मुझे पता है कि यह शायद कोड का सबसे सुंदर टुकड़ा नहीं है, लेकिन अगर आप किसी सार्वभौमिक चीज़ की तलाश कर रहे हैं तो उसे बहुत अच्छा काम करना चाहिए!


यदि आप या तो moto या बोटोकोर स्टबर का उपयोग नहीं करना चाहते हैं (स्टबर को कॉल नहीं रोका जा रहा है तो HTTP एडब्ल्यूएस एपीआई एंडपॉइंट्स के लिए किए जा रहे अनुरोधों को लगता है), आप अधिक वर्बोज़ का उपयोग कर सकते हैं।

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)

जैसे ही मैंने यहां पोस्ट किया मैं एक समाधान के साथ आने में कामयाब रहा। यहाँ यह आशा है कि यह मदद करता है :)

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

जॉर्डन फिलिप्स ने भी docs क्लास का उपयोग करके एक शानदार समाधान पोस्ट किया । एक क्लीनर समाधान के दौरान मैं विशिष्ट कार्यों का मजाक उड़ाने में सक्षम था।





botocore