Python Scenario Sample

from locust import HttpLocust, HttpUser, SequentialTaskSet, task, between
from locust.clients import HttpSession
import random
import json

user_count = 0


class ScenarioTask(SequentialTaskSet):
    def on_start(self):
        # paramsデータが存在したらon_startでload
        global user_count
        self.params_data = {}
        self.random_prefix = (
            f"{user_count}_"
            + "".join(
                [
                    random.choice("1234567890abcdefghijklmnopqrstuvwxyz")
                    for _ in range(10)
                ]
            )
            + "_"
        )
        user_count += 1

    # scenario_order: 1
    @task
    def req_post_signup(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
            "Content-Type": "application/json",
        }
        url = "%s%s" % ("/signup", req_params)
        body_params = {
            "name": "hoge",
            "email": "hoge@email.com",
            "password": "foobar",
            "password_confirmation": "foobar",
        }

        ### パラメーターのランダム化

        body_params["password"] = self.random_prefix + body_params["password"]

        body_params["password_confirmation"] = (
            self.random_prefix + body_params["password_confirmation"]
        )

        body_params["name"] = self.random_prefix + body_params["name"]

        body_params["email"] = self.random_prefix + body_params["email"]

        response = self.client.post(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

    # scenario_order: 2
    @task
    def req_post_auth_login(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
            "Content-Type": "application/json",
        }
        url = "%s%s" % ("/auth/login", req_params)
        body_params = {"email": "hoge@email.com", "password": "foobar"}

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        response = self.client.post(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

        self.adjust_after_param_reg_auth_token_reg = str(json_response["auth_token"])

    # scenario_order: 3
    @task
    def req_post_todos(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
            "Content-Type": "application/json",
        }
        url = "%s%s" % ("/todos", req_params)
        body_params = {"title": "Beethoven"}

        ### パラメーターのランダム化

        body_params["title"] = self.random_prefix + body_params["title"]

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        response = self.client.post(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

        self.レスポンスデータの再利用(URL)_reg_todo_id = str(json_response["id"])

    # scenario_order: 4
    @task
    def req_get_todos(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
        }
        url = "%s%s" % ("/todos", req_params)
        body_params = {}

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.get(url=url, headers=header_params)

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

    # scenario_order: 5
    @task
    def req_post_todos_param1_items(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
            "Content-Type": "application/json",
        }
        url = "%s%s" % ("/todos/#param1#/items", req_params)
        body_params = {"name": "create_music", "done": "false"}

        ### パラメーターのランダム化

        body_params["name"] = self.random_prefix + body_params["name"]

        body_params["done"] = self.random_prefix + body_params["done"]

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.post(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

        self.レスポンスデータの再利用(URL)_reg_item_id = str(json_response["id"])

    # scenario_order: 6
    @task
    def req_get_todos_param1_items(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
        }
        url = "%s%s" % ("/todos/#param1#/items", req_params)
        body_params = {}

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param2#", self.レスポンスデータの再利用(URL)_reg_item_id)

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.get(url=url, headers=header_params)

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

    # scenario_order: 7
    @task
    def req_put_todos_param1_items_param2(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
            "Content-Type": "application/json",
        }
        url = "%s%s" % ("/todos/#param1#/items/#param2#", req_params)
        body_params = {"done": "true"}

        ### パラメーターのランダム化

        body_params["done"] = self.random_prefix + body_params["done"]

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param2#", self.レスポンスデータの再利用(URL)_reg_item_id)

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.put(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

    # scenario_order: 8
    @task
    def req_delete_todos_param1_items_param2(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
        }
        url = "%s%s" % ("/todos/#param1#/items/#param2#", req_params)
        body_params = {}

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param2#", self.レスポンスデータの再利用(URL)_reg_item_id)

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.delete(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

    # scenario_order: 9
    @task
    def req_delete_todos_param1(self):
        req_params = ""
        header_params = {
            "HTTP_HOST": "localhost",
            "HTTP_X_REAL_IP": "127.0.0.1",
            "HTTP_ACCEPT": "*/*",
            "HTTP_USER_AGENT": "curl/7.29.0",
            "HTTP_X_FORWARDED_FOR": "127.0.0.1",
            "HTTP_AUTHORIZATION": "eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxLCJleHAiOjE2NzEwOTQ2Mzh9.EPd6HTYtEv3a6sjtqb2J2krrmn9sHYvXtMa9M7Ebt_c",
            "HTTP_VERSION": "HTTP/1.0",
            "HTTP_CONNECTION": "close",
        }
        url = "%s%s" % ("/todos/#param1#", req_params)
        body_params = {}

        ### レスポンスデータの再利用(ボディ・ヘッダ):set_request_param

        header_params["Authorization"] = self.adjust_after_param_reg_auth_token_reg

        ### レスポンスデータの再利用(URL):set_request_param

        url = url.replace("#param2#", self.レスポンスデータの再利用(URL)_reg_item_id)

        url = url.replace("#param1#", self.レスポンスデータの再利用(URL)_reg_todo_id)

        response = self.client.delete(
            url=url, data=json.dumps(body_params), headers=header_params
        )

        json_response = response.json()

        ### レスポンスデータの再利用(ボディ・ヘッダ)&レスポンスデータの再利用(URL):get_response_param

    @task
    def stop(self):
        self.interrupt()


class LocustUser(HttpUser):
    tasks = [ScenarioTask]
    # TODO: betweenはparam化する、当面はこのsettingで見てみる
    wait_time = between(0.2, 0.4)