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.