boto3 – Passer de monkeypatch à moto

Aujourd’hui, à la suggestion d’un collège (Zachary Paden), j’ai contemplé l’idée de migrer mes tests utilisant pytest/monkeypatch à moto. Ici je vais partager mon parcours et observations.

Pour fin de simplicité, voici une version très minimaliste du code et test de départ.

# main.py

import boto3

session = boto3.session.Session()
client = session.client(service_name="firehose", region_name="us-east-1")

def func() -> None:
    client.put_record(DeliveryStreamName="some-stream", Record={"key": "value"})

# test.py
from typing import Any

import pytest
from _pytest.monkeypatch import MonkeyPatch

import main

@pytest.fixture
def mock_aws(monkeypatch: MonkeyPatch) -> list[dict[str, Any]]:
    """Mock the AWS components.
    Returns:
        A reference to a list where firehose events are added.
    """
    called_with: list[dict[str, Any]] = []

    def mock_put_record(DeliveryStreamName: str, Record: dict[str, Any]) -> None:
        called_with.append(Record)

    monkeypatch.setattr(main.client, "put_record", mock_put_record)

    return called_with

def test_main(mock_aws: list[dict[str, Any]]) -> None:
    main.func()
    assert mock_aws[0] == {"key": "value"}

Passons maintenant à la version du code refait avec moto.

Tout d’abord, il faut installer moto en spécifiant comme «extra»les composantes qui seront utilisées. Avec poetry: poetry add --group dev moto -E s3 -E firehose

# main.py
from typing import Final, Optional

import boto3
from mypy_boto3_firehose.client import FirehoseClient

REGION_NAME: Final = "us-east-1"

_SESSION: Optional[boto3.session.Session] = None
_FH_CLIENT: Optional[FirehoseClient] = None


def get_firehose_client() -> FirehoseClient:
    global _FH_CLIENT
    if not _FH_CLIENT:
        _SESSION = boto3.session.Session()
        _FH_CLIENT = _SESSION.client(service_name="firehose", region_name=REGION_NAME)
    return _FH_CLIENT


def func() -> None:
    get_firehose_client().put_record(DeliveryStreamName="some-stream", Record={"key": "value"})

# test.py
import boto3
import moto

import main

@moto.mock_s3
@moto.mock_firehose
def test_main() -> None:
    # setup
    s3_client = boto3.client("s3", region_name="us-east-1")
    s3_client.create_bucket(Bucket="patate")
    fh_client = boto3.client("firehose", region_name="us-east-1")
    fh_client.create_delivery_stream(
        DeliveryStreamName="some-stream",
        ExtendedS3DestinationConfiguration={"BucketARN": "patate", "RoleARN": "FakeRole"},
    )

    # execution
    main.func()

    # validation
    objs = s3_client.list_objects(Bucket="patate")
    obj = s3_client.get_object(Bucket="patate", Key=objs["Contents"][0]["Key"])
    data = json.loads(obj["Body"].read().decode())
    assert data == {"key": "value"}

Explorons la raison derrière chacun de ces changements.

1. session et client sont maintenant « lazy loaded »

Pour que moto et ses simulacres fonctionnent, il faut que moto soit chargé AVANT la création de session et de client. Ils ne peuvent donc plus être initialisés au scope global. (Techniquement ils le pourraient, si index était importé à l’intérieur de chacun des tests plutôt qu’au niveau global de test.py, mais ça créerait beaucoup de répétition.)

2. La fixture mock_aws est remplacée par les deux décorateurs @moto.mock_s3 et @moto.mock_firehose

Ces décorateurs font en sorte que les clients s3 et firehose seront simulés pour la durée du test.

3. Le test doit créer un simulacre complet des composantes AWS impliquées

Même si main.func n’utilise qu’un firehose, le test doit tout de même d’abord créer un compartiment s3 afin d’y faire pointer ledit firehose, car un firehose moto, tout comme un vrai, doit avoir une destination.

4. Pour accéder au résultat, il faut continuer de prétendre d’interagir avec AWS

Puisque main.func écrit dans un firehose, et que celui-ci a pour destination s3, pour consulter le résultat de l’invocation de put_record, il faut aller le chercher dans s3.

Mes observations

« Lazy loader » session et client, n’est pas si mal, mais. Dépendamment de quand a lieu cette initialisation, elle peut aller à l’encontre du principe « fail fast« . Cette technique nécessite également plus de code.

D’un côté, l’approche monkeypatch ne demande à peu près aucune connaissance d’AWS. De l’autre, l’approche moto demande des connaissances plus grandes que celles nécessaires pour écrire le code qui est testé. Dans la situation ci-présente, il faut savoir qu’un firehose doit avoir une destination et que s3 est supporté. Ensuite il faut savoir comment utiliser le client s3 pour créer un compartiment, récupérer un objet et en lire le contenu. On peut se demander jusqu’où cette complexité peut s’étendre avec des applications plus complexes.

L’approche monkeypatch demande de remplacer toutes les invocations, une à une, peu importe où elles sont. L’approche moto les remplace toutes d’un seul trait.

Sachant que la plupart du temps, il faut réutiliser les simulacres d’un test à l’autre, l’approche implémenté ci-dessus n’est pas idéale, car il faudra copier/coller la portion # setup d’un test à l’autre. Pour pallier à cette situation, voici une réimplémentation de test.py.

# test.py
from typing import Generator

import boto3
import moto

from mypy_boto3_firehose.client import FirehoseClient
from mypy_boto3_s3.client import S3Client

import main

@pytest.fixture
def s3_client() -> Generator[S3Client, None, None]:
    with moto.mock_s3():
        conn = boto3.client("s3", region_name="us-east-1")
        yield conn


@pytest.fixture
def firehose_client() -> Generator[FirehoseClient, None, None]:
    with moto.mock_firehose():
        conn = boto3.client("firehose", region_name="us-east-1")
        yield conn

@pytest.fixture(autouse=True)
def aws_setup(s3_client: S3Client, firehose_client: FirehoseClient):
    s3_client.create_bucket(Bucket="patate")
    firehose_client.create_delivery_stream(
        DeliveryStreamName="api-tracking",
        ExtendedS3DestinationConfiguration={"BucketARN": "patate", "RoleARN": "FakeRole"},
    )


def test_v3_api_key_support(s3_client: S3Client) -> None:
    # execution
    main.func()

    # validation
    objs = s3_client.list_objects(Bucket="patate")
    obj = s3_client.get_object(Bucket="patate", Key=objs["Contents"][0]["Key"])
    data = json.loads(obj["Body"].read().decode())
    assert data == {"key": "value"}

Ici, la fixture aws_setup est définie avec le paramètre autouse=True, ce qui fait qu’elle est invoquée automatiquement à chaque test. Ensuite, chaque test peut avoir comme argument s3_client pour obtenir un client s3 plutôt que de devoir le créer.

Conclusion

Je ne vois pas d’approche clairement gagnante. Je crois que le choix est vraiment contextuel.

En ce qui me concerne, je n’avais qu’une seule invocation à firehose.put_record à remplacer, ce qui fait que le monkeypatch était simple. Aussi, de par mes essais je n’avais migré qu’une seule fonction de test et j’aurais dû passer plus de temps pour n’avoir, à mon avis, aucun gain. J’ai donc opté pour garder le monkeypatch en place.

Billet connexe