Utilisation de Locust pour tester la charge d'une application FastAPI avec des users simultanées.
Écrivez des tests pour les user dans du code Python ordinaire, puis exécutez les tests avec des user concurrents.
![post main image](/static/blog/posts/media/images/blog-post-cd2fe5fda837478da5013d9578b0457d-8083-using-locust-to-test-a-fastapi-app-with-concurrent-users.jpg)
Je viens de terminer ma première application FastAPI . Cette application permet aux user d'avoir leurs propres éléments, ce qui signifie que les modèles de données user ont tous un champ user_id. Rien de spécial, mais comme FastAPI introduit certaines choses nouvelles pour moi, comme Dependency Injection, je n'étais pas sûr que mon application fonctionnerait comme je le voulais.
Ma question était la suivante : si je teste la charge de la API avec de nombreuses user simultanées, est-ce que cela fonctionne toujours ?
Avertissement. Ne lancez un test de charge sur votre ordinateur que si vous êtes sûr qu'il peut le supporter.
Définition du test
J'avais besoin d'un outil qui puisse :
- Initialiser plusieurs users
- Exécuter simultanément des tests pour chaque user.
- Faire cela aussi vite que possible
Pour initialiser une user , je dois enregistrer la user avec une adresse électronique unique. Ensuite, je récupère le jeton d'accès pour cette user, et je définis l'en-tête pour les opérations GET et POST .
Je veux le test suivant pour chaque user :
- Ajouter un élément avec un nom unique (POST)
Cela renvoie l'élément ajouté, y compris son identifiant. - Lire l'élément par son id (GET)
Renvoie l'élément. - Vérifier si le nom dans l'élément retourné est égal au nom unique
Sélection de Locust pour le test
Pour les applications web triviales, j'utilise Apache Bench (ab), un outil très facile à utiliser pour les tests de stress. Mais il n'est pas vraiment utile ici car nous devons effectuer une initialisation et écrire un test (simple).
J'ai déjà écrit de nombreux tests pour mon application FastAPI avec Pytest, et il n'y avait aucun problème. Je me suis renseigné sur pytest-xdist mais il semble qu'il ait des problèmes et/ou qu'il soit trop lent( ?).
J'ai décidé d'utiliser Locust dans ce cas. Il est facile à utiliser et vous pouvez coder vos tests dans Python. Merveilleux.
Le code
Voici le code du test. Quelques notes :
import logging
Je consigne également tout dans un fichier.
def unique_str()
Dans le test, j'ai besoin d'adresses e-mail et de noms de ville uniques. On pourrait aussi faire un UUID4 ici.
network_timeout, connection_timeout
Je ne veux pas que le test échoue mais que le client attende.
wait_time
Je veux que le test suivant commence immédiatement lorsque le test en cours se termine.
dummy task
J'active cette tâche et désactive toutes les autres tâches lors du développement de la méthode 'on_start'.
(POST/GET) catch_response=True
Avec la construction 'with', cela nous permet de marquer une requête comme ayant échoué en appelant response.failure().
(GET) name=<URL>
Locust montre toutes les URL mais ici GET a une URL différente à chaque fois à cause de l'ID. En utilisant 'name=', seule cette URL apparaît.
Vérification des codes d'état
Locust considère qu'une requête est réussie si le code de réponse est <400, mais je veux vérifier explicitement la valeur.
RescheduleTask()
Lorsqu'une erreur se produit, il n'est pas nécessaire de poursuivre une tâche, donc nous levons une exception. Cela pourrait être fait de manière plus avancée.
# test_cities_wrc.py
import logging
import random
import time
from locust import HttpUser, task, between
from locust.exception import RescheduleTask
test_email_host = 'my-local-test.test'
def unique_str():
return '{:.10f}.{}'.format(time.time(), random.randint(1000, 9999))
def unique_email():
return 'e.' + unique_str() + '@' + test_email_host
def unique_city_name():
return 'c.' + unique_str()
class WebsiteTestUser(HttpUser):
network_timeout = 30.0
connection_timeout = 30.0
#wait_time = between(0.5, 3.0)
def on_start(self):
base_url = 'http://127.0.0.1:8020/api/v1'
# set up urls
register_url = base_url + '/auth/register'
get_token_url = base_url + '/auth/token'
# urls used in task
self.cities_create_url = base_url + '/cities'
self.cities_get_by_id_url = base_url + '/cities/'
# get unique email
email = unique_email()
password = 'abcdefghi'
# register
response = self.client.post(
register_url,
json={'email': email, 'password': password},
)
if response.status_code != 201:
error_msg = 'register: response.status_code = {}, expected 201'.format(response.status_code)
logging.error(error_msg)
# get_token
# - username instead of email
# - x-www-form-urlencoded (instead of json)
response = self.client.post(
get_token_url,
data={'username': email, 'password': password},
)
access_token = response.json()['access_token']
logging.debug('get_token: for email = {}, access_token = {}'.format(email, access_token))
# set headers with access token
self.headers = {'Authorization': 'Bearer ' + access_token}
def on_stop(self):
pass
# enable this dummy task to develop 'on_start'
#@task
def dummy(self):
pass
@task
def cities_write_read_check(self):
# add city to api
city_name = unique_city_name()
logging.debug('cities_create: city_name = {}'.format(city_name))
with self.client.post(
self.cities_create_url,
json={'name': city_name},
headers=self.headers,
catch_response=True,
) as response:
if response.status_code != 201:
error_msg = 'cities_create: response.status_code = {}, expected 201, city_name = {}'.format(response.status_code, city_name)
logging.error(error_msg)
response.failure(error_msg)
raise RescheduleTask()
response_dict = response.json()
if 'data' not in response_dict:
error_msg = 'cities_create: data not in response_dict, city_name = {}'.format(city_name)
logging.error(error_msg)
response.failure(error_msg)
raise RescheduleTask()
city_id = response_dict['data']['id']
logging.debug('cities_create: for city_name = {}, city_id = {}'.format(city_name, city_id))
# get city from api and check
with self.client.get(
self.cities_get_by_id_url + city_id,
headers=self.headers,
name=self.cities_get_by_id_url + 'uuid',
catch_response=True,
) as response:
if response.status_code != 200:
error_msg = 'cities_get_by_id: response.status_code = {}, expected 200, city_name = {}'.format(response.status_code, city_name)
logging.error(error_msg)
response.failure(error_msg)
raise RescheduleTask()
if 'data' not in response_dict:
error_msg = 'cities_get_by_id: data not in response_dict, city_name = {}'.format(city_name)
logging.error(error_msg)
response.failure(error_msg)
raise RescheduleTask()
city_name_returned = response_dict['data']['name']
logging.debug('cities_get_by_id: for city_id = {}, city_name_returned = {}'.format(city_id, city_name_returned))
if city_name_returned != city_name:
error_msg = 'cities_get_by_id: city_name_returned = {} not equal city_name = {}'.format(city_name_returned, city_name)
logging.error(error_msg)
response.failure(error_msg)
raise RescheduleTask()
Exécution du test
AVERTISSEMENT : Lorsque vous exécutez un test comme celui-ci, vous stressez votre CPU !
Pour exécuter le test, et enregistrer dans un fichier 'lc.log', ouvrez un terminal et entrez la commande :
locust -L DEBUG --logfile lc.log -f ./test_cities_wrc.py
Et ensuite dans votre navigateur tapez :
http://localhost:8089/
Puis par exemple entrez :
- Nombre total de user à simuler : 20
- Taux de spawn (users spawned/second) : 20
- Hôte : http:127.0.0.1
et appuyez sur le bouton 'Start swarming'. Immédiatement, vous devriez entendre les ventilateurs de votre ordinateur se mettre à tourner à la vitesse maximale...
Quel a été le résultat ?
J'ai effectivement trouvé un problème avec mon code. Ce problème n'est pas apparu avec Pytest. Il est apparu que quelque part dans ma AccessControlManager 4_6791_TNEMECALPER_4Q utilisée avec la Dependency Injection, une variable pouvait être écrasée par une autre requête. En conséquence, un mauvais user_id était renvoyé et le test échouait, pas tout le temps mais lorsque deux ou plusieurs user accédaient à la API au même moment.
Résumé
Je n'étais pas sûr que ma première application FastAPI fonctionnerait sous une charge avec de nombreux user simultanés. Locust m'a permis de simuler cette situation de manière très simple, en codant le test dans Python. Locust dispose de nombreuses options et peut également fonctionner sans tête, c'est-à-dire sans navigateur.
Ce test m'a permis non seulement de détecter le problème mais aussi de l'isoler. J'ai corrigé le problème, mais cela m'a également incité à en savoir plus sur Dependency Injection en FastAPI.
Liens / crédits
Locust
https://docs.locust.io/en/stable/
Laissez un commentaire
Commentez anonymement ou connectez-vous pour commenter.
Commentaires (1)
Laissez une réponse
Répondez de manière anonyme ou connectez-vous pour répondre.
![avatar](/static/img/avatars/default.png)
Hello ! This is a nice article. But I was wondering one thing :
Imagine you have setup your engine like this :
engine = create_engine(SQLALCHEMY_DATABASE_URI, pool_size=5, max_overflow=5)
And I have setup the middleware just like you said, when you try to start 11 requests at the same time, it just hangs. There's not really a queue or nothing. Is this because my routes are async ?
Thanks !
Récent
- Masquer les clés primaires de la base de données UUID de votre application web
- Don't Repeat Yourself (DRY) avec Jinja2
- SQLAlchemy, PostgreSQL, nombre maximal de lignes par user
- Afficher les valeurs des filtres dynamiques SQLAlchemy
- Transfert de données sécurisé grâce au cryptage à Public Key et à pyNaCl
- rqlite : une alternative à haute disponibilité et dist distribuée SQLite
Les plus consultés
- Utilisation des Python's pyOpenSSL pour vérifier les certificats SSL téléchargés d'un hôte
- Utiliser UUIDs au lieu de Integer Autoincrement Primary Keys avec SQLAlchemy et MariaDb
- Connexion à un service sur un hôte Docker à partir d'un conteneur Docker
- Utiliser PyInstaller et Cython pour créer un exécutable Python
- SQLAlchemy : Utilisation de Cascade Deletes pour supprimer des objets connexes
- Flask RESTful API validation des paramètres de la requête avec les schémas Marshmallow