IPM-Load-Test/locustfile.py at master · nusdbsystem/IPM-Load-Test · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
from locust import HttpUser, TaskSet, task, between
import random

# class User10kOnly(TaskSet):
#     # Load tokens from a file
#     with open('tokens.txt') as f:
#         tokens = f.read().splitlines()

#     def on_start(self):
#         # Assign a unique token to each user
#         self.id, self.token = User10kOnly.tokens.pop().split(" ")
#         self.headers = {"Authorization": f"Bearer {self.token}"}
#     @task
#     def get_apis(self):
#         APIs = [
#             "/api/v1/org?page=1&condition=undefined&value=undefined",
#             "/api/v1/dashboard/manager/upcoming_event",
#             "/api/v1/dashboard/manager/ip/count",
#             "/api/v1/dashboard/manager/patent/count",
#             "/api/v1/dashboard/manager/agr/count",
#             "/api/v1/dashboard/manager/agr/revenue",
#             "/api/v1/dashboard/manager/ip/expense",
#             "/api/v1/dashboard/manager/ip/tag/count",
#             "/api/v1/dashboard/manager/patent/count_by_country",
#             "/api/v1/dashboard/manager/patent/count_by_type",
#             "/api/v1/dashboard/manager/trl_chart",
#             "/api/v1/dashboard/manager/crl_chart"
#         ]

#         for api in APIs:
#             self.client.get(api, headers=self.headers)

#         # get IPs
#         api = "/api/v1/ip?page=1&page_size=5&condition=ManagerId,SortedBy,SortOrder&value={},modified_on,desc".format(self.id)
#         self.client.get(api, headers=self.headers)

class FullDatabaseUserBehavior(TaskSet):
     # Load tokens from a file
    with open('tokens.txt') as f:
        tokens = f.read().splitlines()

    def on_start(self):
        # Assign a unique token to each user
        self.id, self.token = FullDatabaseUserBehavior.tokens.pop(0).split(" ")
        self.headers = {"Authorization": f"Bearer {self.token}"}


    @task
    def update_patent(self):
        # get patents by created_by
        api = "/api/v1/patent?page=1&page_size=30&condition=CreatedBy&value={}".format(self.id)
        response = self.client.get(api, headers=self.headers)

        # random patent_id
        random_index = random.randint(0, len(response.json()['list']) - 1)
        patent_id = response.json()['list'][random_index]['patent_id']

        # update patent
        api = "/api/v1/patent/{}".format(patent_id)
        payload = {
            "title": "Updated Title",
            "modified_by": self.id
        }
        self.client.put(api, data=payload, headers=self.headers)

    @task(9)
    def get_patent(self):
        # get patents by created_by
        api = "/api/v1/patent?page=1&page_size=20&condition=CreatedBy&value={}".format(self.id)
        response = self.client.get(api, headers=self.headers)

        # random patent_id
        random_index = random.randint(0, len(response.json()['list']) - 1)
        patent_id = response.json()['list'][random_index]['patent_id']

        # get patent
        api = "/api/v1/patent/{}".format(patent_id)
        self.client.get(api, headers=self.headers)


class WebsiteUser(HttpUser):
    tasks = [FullDatabaseUserBehavior]
    wait_time = between(2, 5)