LOCUST를 선택한 이유
-
최근에 부하테스트를 진행할 일이 있었다. 여러가지 부하 테스팅 툴이 있었지만, 나는 LOCUST를 선택 했다.
-
LOCUST를 선택한 이유는 설치가 간단해보이고, 파이썬으로 스크립트를 작성해 코드로 관리할 수 있다는 점도 마음에 들었다.
-
그리고 또한 웹 기반 UI를 제공하여 결과를 확인하거나, 조작하기도 편리하였다.
설치 방법
- 설치 방법은 그냥 터미널에서 다음 명령어를 입력하며 된다.
pip3 install locust
테스트 방법
- 원하는 폴더에 가서,
locustfile.py를 만들고 나서, 아래 코드를 작성한다.
from locust import HttpUser, task
class HelloWorldUser(HttpUser):
@task
def hello_world(self):
self.client.get("/hello")
self.client.get("/world")
-
가장 간단한 예제 코드이며, 해당 코드를 실행하면
HTTP요청을 서버에 전송한다. -
LOCUST를 실행하는locustfile.py를 작성한 디렉터리에서 터미널에 아래 명령어를 입력하는 것이다.
locust --host="server address"
- 실행 하면, 사용자 수와 사용자가 증가하는 기간을 설정할 수 있고, 초당 응답 시간 및 전체적인 지표를 확인할 수 있다.
from random import randint
from locust import HttpUser, task
class UserBehavior(HttpUser):
def on_start(self):
self.login()
def on_stop(self):
self.logout()
def login(self):
response = self.client.post("/api/v1/auth/login", json={"user-id": "id",
"user-pwd": "pass"})
access_token = response.json()['data']['access_token']
self.client.headers.update({'Authorization': 'Bearer ' + access_token})
# # user info
self.user_company_key = response.json()['data']['user_info']['company_key']
def logout(self):
self.client.delete("/api/v1/auth/logout_access_token")
# 조회
@task
def view_dashboard(self):
"""
HOME - DASH BOARD
"""
# get list
response = self.client.get('/api/v1/dashboard/summary-info?role=ADMIN&company_key=26')
response = self.client.get('/api/v1/notice/dashboard-notices?company_key=26')
response = self.client.get('/api/v1/group/groups?show_deleted=0&company_key=26&by_requestor=0')
response = self.client.get('/api/v1/dashboard/compliance-list?offset=0&count=12&group_list=&company_key=26')
@task
def view_home_notice(self):
"""
HOME - NOTICE
"""
# get list
response = self.client.get(
'/api/v1/notice/notices?offset=0&count=20&company_key=26&show_deleted=0&is_cm=1&is_admin=1&is_service_admin=0')
@task
def view_training_my_training_to_do_list(self):
"""
TRAINING - MY TRAINING - TO DO LIST
"""
# get list
response = self.client.get(
'/api/v1/contents/my-lecture/to-do-list?show_deleted=0&offset=0&count=20&category_list=&my_status=NOT+OPEN,+IN+PROGRESS,+OVERDUE&status_list=ACTIVE,+EXPIRED&after_active_date=1&exclude_previous=1&validity=1&company_key=26')
@task
def view_training_my_training_optional_list(self):
"""
TRAINING - MY TRAINING - OPTIONAL
"""
# get list
response = self.client.get(
'/api/v1/contents/lectures?show_deleted=0&offset=0&count=20&status=ACTIVE&training_type=OPTIONAL&category_list=&attended=0&validity=1&my_lecture=0&sorting_mode=due_date_desc&exclude_previous=1&type=LECTURE&company_key=26')
@task
def view_training_my_training_all_lectures(self):
"""
TRAINING - MY TRAINING - ALL
"""
# get list
response = self.client.get(
'/api/v1/contents/my-lecture/to-do-list?show_deleted=0&offset=0&count=20&category_list=238,239,267,268,240,242,489,366,367,368,488,343,369,370,472,473,371,312,382,467,476,483,490,482,495,491,493,496,497&validity=1&company_key=26')
@task
def view_all_training_list(self):
"""
TRAINING - ALL TRAINING
"""
# get list
response = self.client.get('/api/v1/user/all_user_list?company_key=26&user_status_list=ACTIVE')
response = self.client.get('/api/v1/job-title/all-job-titles?company_key=26')
response = self.client.get('/api/v1/contents/categories-from-dic?exclude_previous=1&company_key=26')
response = self.client.get(
'/api/v1/contents/lectures?show_deleted=0&offset=0&count=20&category_list=&exclude_previous=1&company_key=26')
@task
def view_user_management_group(self):
"""
USER_MANAGEMENT - GROUP
"""
# get list
response = self.client.get('/api/v1/group/groups?show_deleted=0&company_key=26&by_requestor=0')
@task
def view_user_management_job_title(self):
"""
USER_MANAGEMENT - JOB TITLE
"""
# get list
response = self.client.get('/api/v1/job-title/job-titles?show_deleted=0&offset=0&count=20&company_key=26')
@task
def view_user_management_user(self):
"""
USER_MANAGEMENT - USER
"""
# get list
response = self.client.get('/api/v1/user/all_user_list?company_key=26&user_status_list=ACTIVE')
response = self.client.get('/api/v1/job-title/all-job-titles?company_key=26')
response = self.client.get('/api/v1/group/groups?show_deleted=0&company_key=26&by_requestor=0')
response = self.client.get(
'/api/v1/user/users?status=ACTIVE,INACTIVE&offset=0&count=20&roles=&group_list=&show_deleted=0&company_key=26')
@task
def view_contents_management_category(self):
"""
CONTENTS MANAGEMENT - CATEGORY
"""
# get list
response = self.client.get('/api/v1/contents/categories?show_deleted=0&exclude_previous=0&company_key=26')
@task
def view_contents_management_lecture(self):
"""
CONTENTS MANAGEMENT - LECTURE
"""
# get list
response = self.client.get('/api/v1/job-title/all-job-titles?company_key=26')
response = self.client.get('/api/v1/user/all_user_list?company_key=26&user_status_list=ACTIVE')
response = self.client.get(
'/api/v1/contents/lectures?show_deleted=0&offset=0&count=20&category_list=&exclude_previous=1&type=LECTURE&company_key=26')
@task
def view_contents_management_quiz(self):
"""
CONTENTS MANAGEMENT - QUIZ
"""
# get list
response = self.client.get(
'/api/v1/contents/quizzes?show_deleted=0&offset=0&count=20&status=ACTIVE,+INACTIVE&company_key=26')
@task
def view_contents_management_course(self):
"""
CONTENTS MANAGEMENT - COURSE
"""
# get list
response = self.client.get('/api/v1/job-title/all-job-titles?company_key=26')
response = self.client.get('/api/v1/user/all_user_list?company_key=26&user_status_list=ACTIVE')
response = self.client.get('/api/v1/contents/categories-from-dic?exclude_previous=1&company_key=26')
response = self.client.get(
'/api/v1/contents/lectures?show_deleted=0&offset=0&count=20&category_list=&exclude_previous=1&type=COURSE&company_key=26')
@task
def view_learning_progress_training(self):
"""
LEARNING PROGRESS - TRAINING
"""
# get list
response = self.client.get('/api/v1/contents/categories-from-dic?company_key=26')
response = self.client.get('/api/v1/progress/lectures?offset=0&count=20&category_list=&company_key=26')
@task
def view_learning_progress_user(self):
"""
LEARNING PROGRESS - USER
"""
# get list
response = self.client.get('/api/v1/user/all_user_list_by_requester?company_key=26&user_status_list=ACTIVE')
response = self.client.get('/api/v1/group/groups?show_deleted=0&company_key=26&by_requestor=1')
response = self.client.get('/api/v1/contents/categories-from-dic?company_key=26')
response = self.client.get(
'/api/v1/progress/users?offset=0&count=20&show_deleted=0&category_list=&show_inactive_user=false&company_key=26')
@task
def view_system_management(self):
"""
SYSTEM MANAGEMENT
"""
response = self.client.get('/api/v1/sys/system?company_key=26')
# 생성
@task
def add_category(self):
"""
CONTENTS MANAGEMENT - CATEGORY - ADD
"""
num1 = randint(0, 5000)
num2 = randint(0, 5000)
response = self.client.post('/api/v1/contents/categories', json={
"parent_category_name": "NEW COMPANY 1",
"parent_category_key": 0,
"category_name": "test_category_" + str(num1) + "_" + str(num2),
"note": None,
"training_type": "MANDATORY",
"due_days": None,
"alarm": 0,
"alarm_days": 0,
"deleted": 0,
"status": "ACTIVE",
"trainee_list": [],
"mandatory_trainee_list": [],
"optional_trainee_list": [],
"category_key": -1,
"inactive_date": None,
"will_inactive": 0,
"name": "test_category_1",
"company_key": 26
})
@task
def add_lecture(self):
"""
CONTENTS MANAGEMENT - LECTURE - ADD
"""
num1 = randint(0, 5000)
num2 = randint(0, 5000)
response = self.client.post('/api/v1/contents/lectures', json={
"category_key": 497,
"name": "lecture_" + str(num1) + "_" + str(num2),
"practitioner_type": None,
"training_year": None,
"note": None,
"training_type": "MANDATORY",
"lecture_length": 0,
"active_date": "2021-10-20",
"inactive_date": "9999-12-31",
"due_date": None,
"due_days": 1,
"alarm": 0,
"alarm_days": 0,
"certificate": 0,
"complete_date_mode": "AUTO_ASSIGN",
"trainee_list": [
{"trainee_key": 1013, "value": "dwlee", "trainee_type": "USER", "training_type": "MANDATORY"}],
"mandatory_trainee_list": [
{"trainee_key": 1013, "value": "dwlee", "trainee_type": "USER", "training_type": "MANDATORY"}],
"optional_trainee_list": [], "include_contents": 0, "contents_type": None, "contents_name": None,
"contents_url": None,
"contents_link": "", "completion_period": 0, "visibility": False, "add_on_file_name": None,
"add_on_file_url": None,
"include_quiz": 0, "quiz_type": None, "quiz_key": None, "number_of_quiz": 0, "passing_score": 0,
"start_date": None,
"end_date": None, "has_quiz_time_limit": 0, "quiz_time_limit": 0, "company_key": 26
})
-
위는 내가 작성한 스크립트이다. 특이한 점으로는 부하 테스트를 실행하고 끝낼 때
on_start()메서드와,on_stop()메서드를 이용하여 로그인과 로그아웃을 하는 것이다. -
그리고 로그인 할 때 반환되는
JWT토큰정보를self.client.headers.update를 이용하여 저장하는 것을 확인할 수 있다. -
이렇게 하면, API 호출을 하기 위해서 토큰이 필요하더라도 문제 없이 호출할 수 있다.
참고 문헌
>> Home