Belastungstest mit Locust. Teil 2

  • Tutorial
Für diejenigen, die meinen vorherigen Artikel mochten , teile ich weiterhin meine Eindrücke mit dem Tool zum Testen von Locust.

Ich werde versuchen, die Vorteile des Schreibens eines Python-Load-Test-Codes visuell darzustellen, in dem Sie bequem sowohl Daten für den Test vorbereiten als auch die Ergebnisse verarbeiten können.


Serverantworten bearbeiten


Bei Lasttests reicht es manchmal nicht aus, nur ein HTTP 200-OK vom Server zu erhalten. Es kommt vor, dass Sie immer noch den Inhalt der Antwort überprüfen müssen, um sicherzustellen, dass der Server unter Last die korrekten Daten bereitstellt oder genaue Berechnungen durchführt. Nur für solche Fälle hat Locust die Möglichkeit hinzugefügt, die Parameter für den Erfolg der Serverantwort neu zu definieren. Betrachten Sie das folgende Beispiel:

from locust import HttpLocust, TaskSet, task
import random as rnd
classUserBehavior(TaskSet):   @task(1)defcheck_albums(self):
       photo_id = rnd.randint(1, 5000)
       with self.client.get(f'/photos/{photo_id}', catch_response=True, name='/photos/[id]') as response:
           if response.status_code == 200:
               album_id = response.json().get('albumId')
               if album_id % 10 != 0:
                   response.success()
               else:
                   response.failure(f'album id cannot be {album_id}')
           else:
               response.failure(f'status code is {response.status_code}')
classWebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

Es gibt nur eine Anforderung, die im folgenden Szenario geladen wird:
Vom Server fordern wir Fotos mit einer zufälligen ID im Bereich von 1 bis 5000 an und überprüfen die Album-ID in diesem Objekt, sofern es sich nicht um ein Vielfaches von 10 handelt
Hier können Sie sofort einige Erklärungen geben:

  • awesome mit request () - Konstruktion als Antwort: Sie können erfolgreich mit response = request () ersetzen und leise mit dem Antwortobjekt arbeiten
  • Die URL wird mit der in Python 3.6 hinzugefügten Zeichenkettenformatsyntax generiert, wenn ich mich nicht irre - f '/ photos / {photo_id}' . In früheren Versionen funktioniert dieses Design nicht!
  • Das neue Argument catch_response = True zeigt Locust an, dass wir selbst den Erfolg der Serverantwort bestimmen werden. Wenn Sie es nicht angeben, erhalten wir das Antwortobjekt auf dieselbe Weise und können seine Daten verarbeiten, das Ergebnis jedoch nicht neu definieren. Unten ist ein detailliertes Beispiel.
  • Ein weiteres Argument name = '/ photos / [id]' . Sie wird zum Gruppieren von Anforderungen in Statistiken benötigt. Der Name kann ein beliebiger Text sein, die URL muss nicht wiederholt werden. Andernfalls wird jede Anforderung mit einer eindeutigen Adresse oder Parametern separat aufgezeichnet. So funktioniert es:


Mit demselben Argument können Sie einen anderen Trick ausführen - manchmal kommt es vor, dass ein Dienst mit unterschiedlichen Parametern (z. B. unterschiedlicher Inhalt von POST-Anforderungen) eine andere Logik ausführt. Um zu testen , die Ergebnisse sind nicht gemischt, ist es möglich , einige spezifische Aufgaben zu schreiben, für jeden sein Argument Angabe Namen .

Als nächstes machen wir die Prüfungen. Ich habe zwei davon: Zuerst prüfen wir, ob der Server die Antwort an uns zurückgesendet hat, wenn response.status_code == 200 :

Wenn ja, überprüfen wir, ob die Album-ID ein Vielfaches von 10 ist. Wenn nicht mehrere, dann markieren wir diese Antwort als erfolgreiche response.success ()

in anderen In Fällen geben wir an, warum die Antwort fehlgeschlagen ist. Antwort.failure ('fehlertext') . Dieser Text wird während der Testausführung auf der Seite Fehler angezeigt.



Auch aufmerksame Leser könnten das Fehlen von Ausnahmebehandlern (Ausnahmen) feststellen, die für den Code charakteristisch sind, der mit Netzwerkschnittstellen funktioniert. Im Falle eines Timeouts, eines Verbindungsfehlers und anderer unvorhergesehener Ereignisse wird Locust die Fehler selbst behandeln und die Antwort zurückgeben. Der Status des Antwortcodes ist jedoch 0.

Wenn der Code immer noch eine Ausnahme generiert, wird er zur Laufzeit auf der Registerkarte Ausnahmen aufgezeichnet so können wir es verarbeiten. Die typischste Situation ist, dass in json'e der Antwort der von uns gesuchte Wert nicht zurückkehrte, aber wir führen bereits die folgenden Operationen aus.

Schließen Sie vorher das Thema - im Beispiel verwende ich den Json-Server zur Klarheit, da die Antworten einfacher zu handhaben sind. Mit HTML, XML, FormData, Dateianhängen und anderen Daten, die von HTTP-basierten Protokollen verwendet werden, können Sie jedoch mit demselben Erfolg arbeiten.

Arbeiten Sie mit komplexen Szenarien


Nahezu jedes Mal, wenn eine Aufgabe für die Durchführung von Lasttests einer Webanwendung festgelegt ist, wird schnell klar, dass es nicht möglich ist, eine ausreichende Abdeckung allein mit GET-Diensten bereitzustellen, die lediglich Daten zurückgeben.

Ein klassisches Beispiel: Um einen Online-Store zu testen, ist es wünschenswert, dass der Benutzer dies tut

  1. Öffnete den Hauptspeicher
  2. Ich habe nach Waren gesucht
  3. Öffnet die Details der Ware
  4. Artikel zum Warenkorb hinzufügen
  5. Bezahlt

Im Beispiel können wir davon ausgehen, dass es nicht möglich ist, Dienste in einer zufälligen Reihenfolge nur sequentiell aufzurufen. Darüber hinaus können Waren, Körbe und Zahlungsarten eindeutige Kennzeichnungen für jeden Benutzer haben.

Mit dem vorherigen Beispiel können Sie mit geringfügigen Änderungen das Testen eines solchen Szenarios leicht implementieren. Passen Sie das Beispiel an unseren Testserver an:

  1. Benutzer schreibt einen neuen Beitrag
  2. Der Benutzer schreibt einen Kommentar in den neuen Beitrag.
  3. Der Benutzer liest den Kommentar

from locust import HttpLocust, TaskSet, task
classFlowException(Exception):passclassUserBehavior(TaskSet):   @task(1)defcheck_flow(self):# step 1
       new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'}
       post_response = self.client.post('/posts', json=new_post)
       if post_response.status_code != 201:
           raise FlowException('post not created')
       post_id = post_response.json().get('id')
       # step 2
       new_comment = {
           "postId": post_id,
           "name": "my comment",
           "email": "test@user.habr",
           "body": "Author is cool. Some text. Hello world!"
       }
       comment_response = self.client.post('/comments', json=new_comment)
       if comment_response.status_code != 201:
           raise FlowException('comment not created')
       comment_id = comment_response.json().get('id')
       # step 3
       self.client.get(f'/comments/{comment_id}', name='/comments/[id]')
       if comment_response.status_code != 200:
           raise FlowException('comment not read')
classWebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

In diesem Beispiel habe ich eine neue Klasse FlowException hinzugefügt . Wenn es nicht wie erwartet gelaufen ist, führe ich nach jedem Schritt diese Ausnahmeklasse aus, um das Skript zu unterbrechen. Wenn der Beitrag nicht funktioniert hat, gibt es nichts zu kommentieren usw. Falls gewünscht, kann die Konstruktion durch eine normale Rückkehr ersetzt werden. In diesem Fall wird jedoch während der Ausführung und bei der Analyse der Ergebnisse nicht so deutlich erkannt, in welchem ​​Schritt das Ausführungsskript auf die Registerkarte Ausnahmen fällt. Aus dem gleichen Grund verwende ich den Versuch nicht ... außer der Konstruktion .

Die Last realistisch machen


Jetzt kann ich vorgeworfen werden - im Fall des Ladens ist alles wirklich linear, aber das Beispiel für Beiträge und Kommentare ist zu weit hergeholt - sie lesen Beiträge zehnmal öfter, als sie erstellen. Lassen Sie uns das Beispiel vernünftiger machen. Und es gibt mindestens zwei Ansätze:

  1. Sie können die Liste der von den Benutzern gelesenen Beiträge "hardcore" machen und den Testcode vereinfachen, wenn eine solche Möglichkeit besteht und die Backend-Funktionalität nicht von bestimmten Beiträgen abhängt
  2. Speichern Sie die erstellten Beiträge und lesen Sie sie, wenn die Liste der Beiträge nicht voreingestellt werden kann oder die realistische Belastung stark davon abhängt, welche Beiträge gelesen werden (Ich habe die Erstellung von Kommentaren aus dem Beispiel entfernt, um den Code kleiner und klarer zu machen).

from locust import HttpLocust, TaskSet, task
import random as r
classUserBehavior(TaskSet):
   created_posts = []
   @task(1)defcreate_post(self):
       new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'}
       post_response = self.client.post('/posts', json=new_post)
       if post_response.status_code != 201:
           return
       post_id = post_response.json().get('id')
       self.created_posts.append(post_id)
   @task(10)defread_post(self):if len(self.created_posts) == 0:
           return
       post_id = r.choice(self.created_posts)
       self.client.get(f'/posts/{post_id}', name='read post')
classWebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

In der UserBehavior- Klasse habe ich eine Created_posts- Liste erstellt . Seien Sie besonders aufmerksam - dies ist ein Objekt, das nicht im Klassenkonstruktor __init __ () erstellt wurde. Daher ist diese Liste im Gegensatz zur Clientsitzung für alle Benutzer gleich. Die erste Aufgabe erstellt einen Beitrag und schreibt seine ID in die Liste. Der zweite - zehnmal häufiger - liest einen zufällig ausgewählten Beitrag aus der Liste. Eine weitere Bedingung der zweiten Aufgabe ist die Überprüfung, ob Beiträge erstellt wurden.

Wenn jeder Benutzer nur mit seinen eigenen Daten arbeiten muss, können wir diese im Konstruktor wie folgt deklarieren:

classUserBehavior(TaskSet):def__init__(self, parent):
       super(UserBehavior, self).__init__(parent)
       self.created_posts = list()

Einige weitere Funktionen


Für den sequentiellen Start von Tasks wird in der offiziellen Dokumentation vorgeschlagen, dass auch die Taskanmerkung @seq_task (1) verwendet wird, die die Sequenznummer der Task im Argument angibt

classMyTaskSequence(TaskSequence):    @seq_task(1)deffirst_task(self):pass    @seq_task(2)defsecond_task(self):pass    @seq_task(3)    @task(10)defthird_task(self):pass

In diesem Beispiel führt jeder Benutzer zuerst first_task , dann second_task und dann zehnmal drittes_task aus .

Ehrlich gesagt, das Vorhandensein einer solchen Gelegenheit gefällt, aber im Gegensatz zu den vorherigen Beispielen ist nicht klar, wie die Ergebnisse der ersten Aufgabe bei Bedarf auf die zweite übertragen werden sollen.

Für besonders komplexe Szenarien ist es auch möglich, verschachtelte Aufgabengruppen zu erstellen, indem mehrere TaskSet-Klassen erstellt und miteinander verbunden werden.

from locust import HttpLocust, TaskSet, task
classTodo(TaskSet):   @task(3)defindex(self):
       self.client.get("/todos")
   @task(1)defstop(self):
       self.interrupt()
classUserBehavior(TaskSet):
   tasks = {Todo: 1}
   @task(3)defindex(self):
       self.client.get("/")
   @task(2)defposts(self):
       self.client.get("/posts")
classWebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

Im obigen Beispiel wird das Todo- Skript mit einer Wahrscheinlichkeit von 1 bis 6 ausgeführt , und es wird ausgeführt, bis es mit einer Wahrscheinlichkeit von 1 bis 4 zum UserBehavior- Skript zurückkehrt . Das Vorhandensein des self.interrupt () - Aufrufs ist hier sehr wichtig - ohne ihn wird das Testen der Unteraufgabe wiederholt.

Danke fürs Lesen. Im letzten Artikel werde ich über verteiltes Testen und Testen ohne Benutzeroberfläche schreiben, sowie über die Schwierigkeiten, die beim Testen mit Locust aufgetreten sind und wie man sie umgehen kann.

Jetzt auch beliebt: