auto_chaos.locust_system
🦞 Locust system 🦞
1""" 2🦞 3Locust system 4🦞 5""" 6import os 7import subprocess 8import requests 9import csv 10from locust import HttpUser, task 11from auto_chaos.base_system import BaseSystem 12 13 14class LocustSystem(BaseSystem): 15 """ 16 Locust system 17 """ 18 19 SPAWN_RATE = "5" 20 RUN_TIME = "5" 21 22 def __init__(self): 23 """ 24 Init 25 """ 26 self.availability_route = os.getenv("AVAILABILITY_ROUTE") 27 self.csv_report_names = [ 28 "_stats.csv" 29 ] # , "_failures.csv", "_exceptions.csv", "_stats_history.csv"] 30 super().__init__() 31 32 def stress_api(self, args: list[str] = None): 33 """ 34 Stress api by running locust 35 36 Args: 37 args (list[str], optional): List of arguments to use. Defaults to None. 38 """ 39 if not "http" in args[0]: 40 args[0] = "http://localhost:8080" + args[0] 41 42 # Run locust 43 result = subprocess.run( 44 [ 45 "locust", 46 "-f", 47 os.path.join(os.path.dirname(__file__), "locust_system.py"), 48 "--headless", 49 "-u", 50 args[1], 51 "--spawn-rate", 52 self.SPAWN_RATE, 53 "--run-time", 54 self.RUN_TIME, 55 "--host", 56 args[0], 57 "--csv=stress", 58 ], 59 # stdout=subprocess.PIPE, 60 ) 61 62 # Get CSV reports 63 stress_report = [] 64 for name in self.csv_report_names: 65 csv_file = "stress" + name 66 with open(csv_file, "r") as my_input_file: 67 stress_report += [ 68 (" ".join(row) + "\n") for row in csv.reader(my_input_file) 69 ] 70 # Add the result to results 71 self.results.append(str(stress_report)) 72 73 74class LocustUser(HttpUser): 75 """ 76 A simple locust user 77 """ 78 79 @task 80 def stress_api(self): 81 """ 82 Request on the url given at locust starting 83 """ 84 self.client.get("")
15class LocustSystem(BaseSystem): 16 """ 17 Locust system 18 """ 19 20 SPAWN_RATE = "5" 21 RUN_TIME = "5" 22 23 def __init__(self): 24 """ 25 Init 26 """ 27 self.availability_route = os.getenv("AVAILABILITY_ROUTE") 28 self.csv_report_names = [ 29 "_stats.csv" 30 ] # , "_failures.csv", "_exceptions.csv", "_stats_history.csv"] 31 super().__init__() 32 33 def stress_api(self, args: list[str] = None): 34 """ 35 Stress api by running locust 36 37 Args: 38 args (list[str], optional): List of arguments to use. Defaults to None. 39 """ 40 if not "http" in args[0]: 41 args[0] = "http://localhost:8080" + args[0] 42 43 # Run locust 44 result = subprocess.run( 45 [ 46 "locust", 47 "-f", 48 os.path.join(os.path.dirname(__file__), "locust_system.py"), 49 "--headless", 50 "-u", 51 args[1], 52 "--spawn-rate", 53 self.SPAWN_RATE, 54 "--run-time", 55 self.RUN_TIME, 56 "--host", 57 args[0], 58 "--csv=stress", 59 ], 60 # stdout=subprocess.PIPE, 61 ) 62 63 # Get CSV reports 64 stress_report = [] 65 for name in self.csv_report_names: 66 csv_file = "stress" + name 67 with open(csv_file, "r") as my_input_file: 68 stress_report += [ 69 (" ".join(row) + "\n") for row in csv.reader(my_input_file) 70 ] 71 # Add the result to results 72 self.results.append(str(stress_report))
Locust system
LocustSystem()
23 def __init__(self): 24 """ 25 Init 26 """ 27 self.availability_route = os.getenv("AVAILABILITY_ROUTE") 28 self.csv_report_names = [ 29 "_stats.csv" 30 ] # , "_failures.csv", "_exceptions.csv", "_stats_history.csv"] 31 super().__init__()
Init
def
stress_api(self, args: list[str] = None):
33 def stress_api(self, args: list[str] = None): 34 """ 35 Stress api by running locust 36 37 Args: 38 args (list[str], optional): List of arguments to use. Defaults to None. 39 """ 40 if not "http" in args[0]: 41 args[0] = "http://localhost:8080" + args[0] 42 43 # Run locust 44 result = subprocess.run( 45 [ 46 "locust", 47 "-f", 48 os.path.join(os.path.dirname(__file__), "locust_system.py"), 49 "--headless", 50 "-u", 51 args[1], 52 "--spawn-rate", 53 self.SPAWN_RATE, 54 "--run-time", 55 self.RUN_TIME, 56 "--host", 57 args[0], 58 "--csv=stress", 59 ], 60 # stdout=subprocess.PIPE, 61 ) 62 63 # Get CSV reports 64 stress_report = [] 65 for name in self.csv_report_names: 66 csv_file = "stress" + name 67 with open(csv_file, "r") as my_input_file: 68 stress_report += [ 69 (" ".join(row) + "\n") for row in csv.reader(my_input_file) 70 ] 71 # Add the result to results 72 self.results.append(str(stress_report))
Stress api by running locust
Arguments:
- args (list[str], optional): List of arguments to use. Defaults to None.
Inherited Members
class
LocustUser(locust.user.users.HttpUser):
75class LocustUser(HttpUser): 76 """ 77 A simple locust user 78 """ 79 80 @task 81 def stress_api(self): 82 """ 83 Request on the url given at locust starting 84 """ 85 self.client.get("")
A simple locust user
@task
def
stress_api(self):
80 @task 81 def stress_api(self): 82 """ 83 Request on the url given at locust starting 84 """ 85 self.client.get("")
Request on the url given at locust starting
abstract =
False
If abstract is True, the class is meant to be subclassed, and users will not choose this locust during a test
Collection of python callables and/or TaskSet classes that the Locust user(s) will run.
If tasks is a list, the task to be performed will be picked randomly.
If tasks is a (callable,int) list of two-tuples, or a {callable:int} dict, the task to be performed will be picked randomly, but each task will be weighted according to its corresponding int value. So in the following case, ThreadPage will be fifteen times more likely to be picked than write_post::
class ForumPage(TaskSet):
tasks = {ThreadPage:15, write_post:1}
Inherited Members
- locust.user.users.HttpUser
- HttpUser
- pool_manager
- client
- locust.user.users.User
- host
- min_wait
- max_wait
- wait_time
- wait_function
- weight
- fixed_count
- environment
- on_start
- on_stop
- run
- wait
- start
- stop
- group
- greenlet
- context
- fullname