Forked from adamszeptycki/gist:935cd14e46aab9049b9dfc1847d2beee
Created
August 5, 2019 12:48
-
-
Save revmischa/2b0da8e7dd0069b73bf09afac879b073 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from random import uniform, randint | |
from locust import TaskSequence, HttpLocust | |
from locust import seq_task | |
import random | |
import string | |
import os | |
def randomStringDigits(stringLength=6): | |
"""Generate a random string of letters and digits """ | |
lettersAndDigits = string.ascii_letters + string.digits | |
return ''.join(random.choice(lettersAndDigits) for i in range(stringLength)) | |
def _debt_info_factory(): | |
return dict( | |
ext_debt_id=randomStringDigits(20), | |
ctf_pay_client_id=randomStringDigits(20), | |
fee_basis=15, | |
original_debt=round(uniform(1, 100000), 2), | |
ext_pay_client_id=randomStringDigits(20), | |
) | |
password = "testpass" | |
email = "[email protected]" | |
class SubUserTaskSet(TaskSequence): | |
def on_start(self): | |
""" on_start is called when a Locust start before any task is scheduled """ | |
self.login() | |
def on_stop(self): | |
""" on_stop is called when the TaskSet is stopping """ | |
self.client.headers = {} | |
def login(self): | |
resp = self.client.post("/api/auth/login", json={"email": email, "password": password}) | |
print(resp.json) | |
self.client.headers = { | |
"Authorization": f"Bearer {resp.json()['access_token']}" | |
} | |
class CreateDebt(SubUserTaskSet): | |
debts = [] | |
@seq_task(1) | |
def create_debt(self): | |
debt_info = _debt_info_factory() | |
self.client.post("/api/debt", json=debt_info) | |
self.debts.append(debt_info) | |
@seq_task(2) | |
def update_debt(self): | |
debt = self.debts.pop() | |
debt["fee_basis"] = randint(0, 10) | |
debt["original_debt"] = round(uniform(1, 100000), 2) | |
self.client.patch(f"/api/debt/{debt['ext_debt_id']}", name="/api/debt/<ext_debt_id>", json=debt) | |
class CheckDebt(SubUserTaskSet): | |
@seq_task(1) | |
def get_debts(self): | |
self.client.get("/api/debt") | |
class SubUserCreateDebt(HttpLocust): | |
task_set = CreateDebt | |
min_wait = 1 | |
max_wait = 5 | |
class SubUserCheckDebt(HttpLocust): | |
task_set = CheckDebt | |
min_wait = 1 | |
max_wait = 5 | |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment