putobject - Burlándose de boto3 S3 cliente método Python

write a file to s3 python

Question

Estoy tratando de burlarme de un método singular del objeto cliente de boto3 s3 para lanzarlo y una excepción. Pero necesito todos los otros métodos para que esta clase funcione normalmente.

Esto es para que pueda probar una prueba de excepción singular cuando se produce un error al realizar una upload_part_copy

1er intento

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

Sin embargo, esto da el siguiente error:

ImportError: No module named S3

2do intento

Después de mirar el código fuente de botocore.client.py, descubrí que está haciendo algo inteligente y que el método upload_part_copy no existe. Encontré que parece llamar a BaseClient._make_api_call en BaseClient._make_api_call lugar, así que traté de burlarme de eso

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

Esto lanza una excepción pero lanza una excepción ... pero en el get_object que quiero evitar.

¿Alguna idea sobre cómo solo puedo lanzar la excepción en el método upload_part_copy ?

Answer 1

¿Qué hay de simplemente usar moto ?

Viene con un decorator muy práctico:

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass

Answer 2

Aquí hay un ejemplo de una prueba de unidad de python simple que se puede usar para falsificar la llamada api client = 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''
}

Esperemos que eso ayude.

Answer 3

Botocore tiene un cliente que puedes usar para este propósito: docs .

Aquí hay un ejemplo de poner un error en:

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

Este es un ejemplo de cómo poner una respuesta normal. Además, ahora se puede usar el stubber en un contexto. Es importante tener en cuenta que el rastreador verificará, en la medida de lo posible, que su respuesta proporcionada coincida con lo que realmente devolverá el servicio. Esto no es perfecto, pero lo protegerá de la inserción de respuestas sin sentido.

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 4

Tan pronto como publiqué aquí pude encontrar una solución. Aquí está la esperanza que ayuda :)

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 también publicó una gran solución utilizando la clase docs . Mientras que una solución más limpia no podía simular operaciones específicas.