Estoy tratando de burlarme de un método singular del objeto cliente boto3 s3 para lanzar una excepción. Pero necesito todos los demás 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

Primer 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

2º intento

Después de mirar el código fuente botocore.client.py descubrí que está haciendo algo inteligente y que el método upload_part_copy no existe. Descubrí que parece llamar a BaseClient._make_api_call en su lugar, así que intenté 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 arroja 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?

81
ptimson 10 may. 2016 a las 18:59

7 respuestas

La mejor respuesta

Tan pronto como publiqué aquí, logré encontrar una solución. Aquí es espero que ayude :)

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 el clase botocore.stub.Stubber. Si bien era una solución más limpia, no podía burlarme de operaciones específicas.

24
Community 23 may. 2017 a las 11:47

¿Qué pasa con el simple uso de moto?

Viene con un decorador muy útil:

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass
6
wikier 7 ago. 2018 a las 21:09

Botocore tiene un stubber de cliente que puede usar solo 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()

Aquí hay un ejemplo de cómo poner una respuesta normal. Además, el stubber ahora se puede usar en un contexto. Es importante tener en cuenta que el stubber verificará, en la medida de lo posible, que su respuesta proporcionada coincida con lo que el servicio realmente devolverá. Esto no es perfecto, pero lo protegerá de insertar respuestas totales 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
79
Jordon Phillips 3 jun. 2016 a las 18:13

Si no desea utilizar moto o el stubber de botocore (el stubber no evita que las solicitudes HTTP se realicen a los puntos finales de la API de AWS, al parecer), puede utilizar el testtest más detallado .mock way:

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)

4
c4urself 28 ago. 2019 a las 14:49

Aquí está mi solución para parchear un cliente boto usado en las entrañas de mi proyecto, con accesorios pytest. Solo estoy usando 'mturk' en mi proyecto.

El truco para mí fue crear mi propio cliente y luego parchear boto3.client con una función que devuelve ese cliente pre-creado.

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

También defino otro dispositivo que configura créditos falsos para que boto no recoja accidentalmente algún otro conjunto de credenciales en el sistema. Literalmente establecí 'foo' y 'bar' como mi crédito para las pruebas, eso no es una redacción.

Es importante que AWS_PROFILE env no esté configurado porque, de lo contrario, boto buscará ese perfil.

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

Y luego especifico setup_env como una entrada de pytest usefixtures para que se use en cada ejecución de prueba.

1
deargle 22 may. 2019 a las 19:13

Aquí hay un ejemplo de una simple prueba de unidad de Python que se puede utilizar para falsificar client = boto3.client ('ec2') llamada a la 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'
}

Ojalá eso ayude.

6
Aidan Melen 10 ene. 2020 a las 01:07

Tuve que burlarme del cliente boto3 para algunas pruebas de integración y ¡fue un poco doloroso! El problema que tuve es que moto no es muy compatible con KMS, pero no quería volver a escribir mi propio simulacro para los cubos S3. Así que creé este morfo de todas las respuestas. ¡También funciona a nivel mundial, lo cual es genial!

Lo tengo configurado con 2 archivos.

El primero es aws_mock.py. Para la burla de KMS obtuve algunas respuestas predefinidas que vinieron del cliente boto3 en vivo.

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

El segundo es el módulo de prueba real. Llamémoslo test_my_module.py. He omitido el código de my_module. Así como las funciones que están bajo prueba. Llamemos a esas foo, bar funciones.

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...

Una cosa más, no estoy seguro si eso se solucionó, pero descubrí que moto no estaba contento a menos que establezca algunas variables ambientales como credenciales y región. No tienen que ser credenciales reales, pero deben establecerse. ¡Existe la posibilidad de que se solucione cuando leas esto! Pero aquí hay un código en caso de que lo necesite, ¡código shell esta vez!

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

Sé que probablemente no sea el código más bonito, pero si estás buscando algo universal, ¡debería funcionar bastante bien!

3
Flimzy 5 abr. 2019 a las 08:57