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("")
class LocustSystem(auto_chaos.base_system.BaseSystem):
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

SPAWN_RATE = '5'
RUN_TIME = '5'
availability_route
csv_report_names
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.
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

tasks: List[Union[locust.user.task.TaskSet, Callable]] = [<function LocustUser.stress_api>]

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