Méthode d'imitation du client S3 de boto3 Python

Question

J'essaie de simuler une méthode singulière de l'objet client s3 de boto3 pour lancer une exception. Mais j'ai besoin que toutes les autres méthodes de cette classe fonctionnent normalement.

Cela me permet de tester une exception singulière lorsqu'une erreur se produit lors de l'exécution d'un test d'exception. upload_part_copy

1ère tentative

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

Cependant, l'erreur suivante se produit :

ImportError: No module named S3

2ème tentative

Après avoir regardé le code source de botocore.client.py, j'ai trouvé qu'il faisait quelque chose d'intelligent et que la méthode upload_part_copy n'existe pas. J'ai constaté qu'il semble appeler BaseClient._make_api_call au lieu de cela, j'ai donc essayé de me moquer de cela

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

Cela provoque une exception... mais sur la page get_object ce que je veux éviter.

Avez-vous des idées sur la façon dont je peux lancer l'exception uniquement sur l'élément upload_part_copy méthode ?

Answer 1

Botocore dispose d'un stubber client que vous pouvez utiliser dans ce but : documents .

Voici un exemple d'insertion d'une erreur :

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

Voici un exemple d'insertion d'une réponse normale. De plus, le stubber peut maintenant être utilisé dans un contexte. Il est important de noter que le stubber vérifiera, dans la mesure du possible, que la réponse que vous avez fournie correspond à ce que le service retournera réellement. Ce n'est pas parfait, mais cela vous évitera d'insérer des réponses totalement absurdes.

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

Answer 2

Dès que j'ai posté mon message ici, j'ai trouvé une solution. La voici, j'espère qu'elle vous aidera :)

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 a également proposé une solution intéressante en utilisant le botocore.stub.Stubber classe. Bien qu'il s'agisse d'une solution plus propre, je n'étais pas en mesure de simuler des opérations spécifiques.

Answer 3

Si vous ne voulez utiliser ni l'un ni l'autre moto ou le stubber botocore (le stubber fait pas (il semble que l'on puisse empêcher les requêtes HTTP d'être effectuées vers les points d'extrémité de l'API AWS), vous pouvez utiliser la méthode plus verbeuse de 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)

Answer 4

Voici un exemple d'unittest python simple qui peut être utilisé pour simuler client = boto3.client('ec2') appel 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.get_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'
}

J'espère que cela vous aidera.

Answer 5

Pourquoi ne pas simplement utiliser moto ?

Il est accompagné d'un décorateur :

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass