Verspottung der boto3 S3-Clientmethode Python

74

Ich versuche, eine Singluar-Methode aus dem boto3 s3-Clientobjekt zu verspotten, um eine Ausnahme auszulösen. Aber ich brauche alle anderen Methoden, damit diese Klasse normal funktioniert.

Auf diese Weise kann ich einen einzelnen Ausnahmetest testen, wenn beim Ausführen einer upload_part_copy ein Fehler auftritt

1. Versuch

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

Dies führt jedoch zu folgendem Fehler:

ImportError: No module named S3

2. Versuch

Nachdem ich mir den Quellcode botocore.client.py angesehen hatte, stellte ich fest, dass er etwas Kluges tut und die Methode upload_part_copynicht existiert. Ich fand, dass es BaseClient._make_api_callstattdessen zu rufen scheint, also versuchte ich das zu verspotten

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

Dies löst eine Ausnahme aus ... aber auf get_objectdie ich vermeiden möchte.

Irgendwelche Ideen, wie ich nur die Ausnahme auf die upload_part_copyMethode werfen kann ?

ptimson
quelle

Antworten:

108

Botocore verfügt über einen Client-Stubber, den Sie nur für diesen Zweck verwenden können: docs .

Hier ist ein Beispiel für das Einfügen eines Fehlers:

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

Hier ist ein Beispiel für das Einfügen einer normalen Antwort. Außerdem kann der Stubber jetzt in einem Kontext verwendet werden. Es ist wichtig zu beachten, dass der Stubber so weit wie möglich überprüft, ob Ihre bereitgestellte Antwort mit der tatsächlichen Rückgabe des Dienstes übereinstimmt. Dies ist nicht perfekt, schützt Sie jedoch vor dem Einfügen von vollständigen Unsinnsantworten.

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
Jordon Phillips
quelle
1
Nun, da es sich um Botocore handelt, hätten Sie in den Botocore-Dokumenten nachsehen müssen, und nicht viele tun dies. Es ist auch ziemlich neu.
Jordon Phillips
2
Warum löst client.upload_part_copy () einen ClientError aus?
Aidan Melen
1
@AidanMelen, weil ich der Antwortwarteschlange explizit einen Fehler hinzugefügt habe. Sie können auch normale Serviceantworten hinzufügen. Ich werde aktualisieren, um beide zu zeigen.
Jordon Phillips
7
Muss clientin das zu testende Gerät injiziert werden? Mein Verständnis von Pythonic-Unit-Tests war, dass Tester so etwas verwenden unittest.mock, um importierte Abhängigkeiten zu verspotten. Verspottet dieser Ansatz Boto-Clients, die in andere Dateien importiert werden?
Carl G
Sie würden diese mehr oder weniger genauso verwenden wie ein Scheinobjekt. Sie richten sie ein und geben sie dann an Ihren Code weiter. Wie Sie sie weitergeben, liegt bei Ihnen.
Jordon Phillips
38

Sobald ich hier gepostet habe, habe ich eine Lösung gefunden. Hier ist es hoffentlich hilft es :)

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 hat auch eine großartige Lösung mit der Klasse botocore.stub.Stubber veröffentlicht . Während eine sauberere Lösung war ich nicht in der Lage, bestimmte Operationen zu verspotten.

ptimson
quelle
4
Das ist sehr hilfreich. Es hat eine Weile gedauert, bis mir klar wurde, dass viele der boto3-Clients zur Laufzeit effektiv generiert werden und daher nicht direkt verspottet werden können.
Rumdrums
1
Dies ist die Lösung, die für mich als Stubber funktioniert hat, und viele andere Mocking-Tools können Boto3-Anpassungsfunktionen wie das Hochladen von Dateien oder generierte vorgegebene URLs nicht stubben.
JD D
1
Diese Antwort war großartig. Ich habe zuerst versucht, Stubber zu verwenden, aber es schien nur für einen sofortigen Anruf zu funktionieren. Ich konnte es aus irgendeinem Grund nicht für einen Anruf innerhalb einer Unterfunktion zum Laufen bringen. Dies hat andererseits perfekt funktioniert und ist sehr einfach zu implementieren, also danke!
Jake Boomgaarden
9

Hier ist ein Beispiel für eine einfache Python-Unittest, mit der der API-Aufruf client = boto3.client ('ec2') gefälscht werden kann ...

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'
}

hoffentlich hilft das.

Aidan Melen
quelle
Wie kann ich ein globales Client- oder Ressourcenobjekt verwalten? Dies kann nicht verspottet werden, da der Aufruf vor dem Verspotten-Setup erfolgt.
Pt12lol
3
Die erste Zeile von 'test_open_file_with_existing_file' sollte nicht 'mock_describe_tags.return_value = mock_get_tags_response' sein. anstelle von 'mock_boto_client'?
cloudy_weather
Wie kann man daraus schließen, dass @ mock.patch ("boto3.client.get_tags") verspottet wird
Shivangi Singh
1
Dieser Artikel ist sehr gut. Sie sollten sich darauf beziehen, wenn Sie weitere Fragen haben. toptal.com/python/an-introduction-to-mocking-in-python
Aidan Melen
7

Wenn Sie weder motoden Botocore-Stubber noch den Botocore-Stubber verwenden möchten (der Stubber verhindert anscheinend nicht, dass HTTP-Anforderungen an AWS-API-Endpunkte gesendet werden ), können Sie die ausführlichere Methode unittest.mock verwenden:

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)

c4urself
quelle
6

Was ist mit einfach Moto ?

Es kommt mit einem sehr praktischen Dekorateur :

from moto import mock_s3

@mock_s3
def test_my_model_save():
    pass
wikier
quelle
4

Ich musste den boto3Client für einige Integrationstests verspotten und es war ein bisschen schmerzhaft! Das Problem, das ich hatte, ist, dass motoes nicht KMSsehr gut unterstützt, aber ich wollte mein eigenes Modell für die S3Eimer nicht umschreiben . Also habe ich diesen Morph aller Antworten erstellt. Es funktioniert auch global, was ziemlich cool ist!

Ich habe es mit 2 Dateien eingerichtet.

Der erste ist aws_mock.py. Für die KMSVerspottung habe ich einige vordefinierte Antworten erhalten, die vom Live- boto3Client kamen.

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

Das zweite ist das eigentliche Testmodul. Nennen wir es test_my_module.py. Ich habe den Code von weggelassen my_module. Sowie Funktionen, die getestet werden. Lassen Sie uns die Teilnehmer anrufen foo, barFunktionen.

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

Eine weitere Sache, nicht sicher, ob das behoben ist, aber ich fand heraus, dass dies motonicht glücklich war, es sei denn, Sie haben einige Umgebungsvariablen wie Anmeldeinformationen und Region festgelegt. Sie müssen keine tatsächlichen Anmeldeinformationen sein, aber sie müssen festgelegt werden. Es besteht die Möglichkeit, dass dies behoben ist, wenn Sie dies lesen! Aber hier ist ein Code, falls Sie ihn brauchen, diesmal Shell-Code!

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

Ich weiß, dass es wahrscheinlich nicht der schönste Code ist, aber wenn Sie nach etwas Universellem suchen, sollte es ziemlich gut funktionieren!

Barmaley
quelle
Dies kommt meinem eigenen Anwendungsfall sehr nahe - ich muss mich mit Organisationsaufrufen von boto3 anstelle von KMS befassen. Da jedoch alle coolen Kids es jetzt verwenden, versuche ich, pytest (und pytest-mock) zu verwenden, und ich kann es nicht dazu bringen, Ihre Client-Funktion in das MagicMock zu patchen. Haben Sie dies mit pytest statt unittest versucht? HINWEIS: Ich selbst habe erst kürzlich von unittest gewechselt, sodass pytest immer noch ein Rätsel ist.
Marakai
UPDATE: Ich habe es buchstäblich gerade geschafft, mit Pytest gut zu spielen. Könnte eine Antwort darauf posten, wenn ich diesen Stall habe.
Marakai
@ Marakai, ich benutze tatsächlich pytest, um meine Tests durchzuführen. Ich denke, ich bin etwas neu im Unit-Test und habe nicht bemerkt, dass Pytest eine eigene Scheinimplementierung hat. Hoffentlich ist die Implementierung nicht allzu schwierig!
Barmaley
1
Wenn ich es richtig verstehe (und ich bin kein Experte in diesem Bereich), ist das Mocking-Framework in Pytest lediglich ein Wrapper um das unittest Mock-Framework. Ich fand, dass ich verwenden kann @pytest.fixtureund @mock.patchund es funktioniert. Ich wünschte, ich könnte Ihre Antwort mehr als einmal positiv bewerten. Es hat mir enorm geholfen, Boto3-Stubs auch für Kunden, die Stubs (noch) nicht unterstützen, konsequent zu verwenden.
Marakai
@ Marakai, ich bin froh, dass du meinen Beitrag nützlich fandest! Ich bin auch froh, dass ich wieder zur Stack Overflow Community beitragen kann!
Barmaley
3

Hier ist meine Lösung zum Patchen eines Boto-Clients, der im Darm meines Projekts verwendet wird, mit pytestFixtures. Ich verwende nur 'mturk' in meinem Projekt.

Der Trick für mich bestand darin, einen eigenen Client zu erstellen und dann boto3.clientmit einer Funktion zu patchen , die diesen vorab erstellten Client zurückgibt.

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

Ich definiere auch ein anderes Gerät, das Dummy-Aws-Creds einrichtet, damit Boto nicht versehentlich andere Anmeldeinformationen auf dem System abruft. Ich habe buchstäblich 'foo' und 'bar' als meine Creds zum Testen festgelegt - das ist keine Redaktion.

Es ist wichtig, dass AWS_PROFILEenv nicht gesetzt ist, da Boto sonst nach diesem Profil sucht.

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

Und dann setup_envgebe ich als Pytest- usefixturesEintrag an, damit er für jeden Testlauf verwendet wird.

Liebling
quelle