Locust gebruiken om een FastAPI app met gelijktijdige user's te testen

Ik heb zojuist mijn eerste FastAPI app voltooid. Deze app staat users toe om hun eigen items te hebben, wat betekent dat de user data modellen allemaal een user_id veld hebben. Niets bijzonders, maar omdat FastAPI een aantal dingen introduceert die nieuw voor mij zijn, zoals Dependency Injection, was ik niet zeker of mijn app zou werken zoals ik wilde.

Mijn vraag was: als ik de API loadtest met veel gelijktijdige users, werkt het dan nog steeds?

Waarschuwing. Voer alleen een belastingstest uit op uw computer als u zeker weet dat hij het aankan.

De test definiëren

Ik had een programma nodig dat:

• Initialiseer meerdere users
• Gelijktijdig testen uitvoert voor elke user
• Doe dit zo snel mogelijk

Om een user te initialiseren moet ik de user registreren met een uniek email adres. Daarna krijg ik het access token voor deze user, en stel ik de header in voor de GET en POST operaties.

Ik wil de volgende test voor elke user:

• Voeg een item toe met een unieke naam (POST)
  Dit geeft het toegevoegde item terug, inclusief zijn id.
• Leest het item met id terug (GET)
  Dit geeft het item terug.
• Controleer of de naam in het teruggegeven item gelijk is aan de unieke naam

Selecteren van Locust voor de test

Voor triviale webapplicaties gebruik ik Apache Bench (ab), een zeer eenvoudig te gebruiken tool om te stress-testen. Maar hier niet echt bruikbaar omdat we een initialisatie moeten uitvoeren en een (eenvoudige) test moeten schrijven.

Ik heb al veel tests geschreven voor mijn FastAPI app met Pytest, en er waren geen problemen. Ik las over pytest-xdist maar het lijkt erop dat het problemen heeft en/of te traag is(?).

Ik heb besloten om in dit geval Locust te gebruiken. Het is eenvoudig te gebruiken en je kunt je tests in Python coderen. Prachtig.

De code

Hieronder staat de code voor de test. Enkele opmerkingen:

import logging
Ik log alles ook naar een bestand.

def unique_str()
In de test heb ik unieke email adressen en plaatsnamen nodig. We zouden hier ook een UUID4 kunnen maken.

network_timeout, connection_timeout
Ik wil niet dat de test mislukt, maar dat de klant moet wachten.

wait_time
Ik wil dat de volgende test onmiddellijk begint als de huidige klaar is.

dummy task
Ik schakel deze in en schakel alle andere task uit bij het ontwikkelen van de 'on_start' methode.

(POST/GET) catch_response=True
Samen met de "with"-constructie stelt dit ons in staat een verzoek als mislukt te markeren door response.failure() aan te roepen.

(GET) name=<URL>
Locust toont alle URL's maar hier heeft GET telkens een andere URL wegens het ID. Met de 'name=' laat hij alleen deze URL zien.

Status_code checks
Locust beschouwt een request als succesvol als de response code <400 is, maar ik wil de waarde expliciet controleren.

RescheduleTask()
Als er een fout optreedt, is het niet nodig een taak voort te zetten, dus we heffen een uitzondering op. Dit zou geavanceerder kunnen.

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

De test uitvoeren

WAARSCHUWING: Wanneer je een test als deze uitvoert, belast je je CPU!

Om de test uit te voeren, en te loggen naar een bestand 'lc.log', open je een terminal en voer je het commando in:

locust  -L  DEBUG  --logfile  lc.log  -f ./test_cities_wrc.py

En typ dan in je browser:

http://localhost:8089/

Voer dan bijvoorbeeld in:

• Aantal totaal te simuleren user's: 20
• Spawn rate (user's gespawned/seconde): 20
• Gastheer: http:127.0.0.1

en druk op de 'Start swarming' knop. Onmiddellijk zou u de ventilatoren van uw computer op maximum snelheid moeten horen draaien ...

Wat was het resultaat?

Ik heb inderdaad een probleem gevonden met mijn code. Dit probleem deed zich niet voor bij Pytest. Het bleek dat ergens in mijn AccessControlManager Class gebruikt met Dependency Injection, een variabele overschreven kon worden door een ander verzoek. Als gevolg daarvan werd een verkeerde user_id geretourneerd en mislukte de test, niet altijd maar wanneer twee of meer user's op hetzelfde moment toegang hadden tot de API .

Samenvatting

Ik was onzeker of mijn eerste FastAPI app zou werken onder belasting met veel gelijktijdige users. Locust liet me deze situatie op een zeer eenvoudige manier simuleren, door de test in Python te coderen. Locust heeft veel opties en kan ook headless draaien, dat wil zeggen, zonder browser.

Deze test hielp me niet alleen om het probleem op te sporen, maar ook om het te isoleren. Ik heb het probleem verholpen, maar dit vertelde me ook om meer te lezen over Dependency Injection in FastAPI.

Links / credits

Locust
https://docs.locust.io/en/stable/