LOCUST를 이용한 부하 테스트

이동욱

2021/10/21

Categories: 개발

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