기본 예제

from locust import HttpUser, task

class SampleUser(HttpUser):
    @task
    def sample_task(self):
        self.client.get("/greeting")
        self.client.get("/info")

최적화 팁

• 응답 검증: 2xx 상태 코드만 성공으로 간주되며, 다른 코드는 실패로 처리됩니다.
• SequentialTaskSet vs TaskSet: TaskSet은 무작위 실행을 지원하고, SequentialTaskSet은 순차적으로 실행됩니다.
• 태스크 정의: @task 데코레이터를 사용해 사용자 행동을 지정합니다.
• 부하 형태 제어:
  1. 자동 제어 모드: LoadTestShape 클래스를 상속해 tick 메서드 재정의하여 사용자 수와 요청 주기를 정의합니다.
  2. WebUI 수동 조절: UI에서 '+'/'-' 버튼으로 실시간으로 사용자 수를 조절할 수 있습니다.
• HttpUser 설정:
  • host: 기본 URL 지정
  • wait_time: 태스크 간 대기 시간
  • tasks: 실행할 태스크 목록
• name 속성: 동적인 URL을 통합하여 통계 분석에 유용하게 활용합니다.
• 히버 함수:
  • on_test_start: 테스트 시작 시 실행
  • on_test_stop: 테스트 종료 시 실행

from locust import HttpUser, task, between, LoadTestShape, SequentialTaskSet, constant_throughput, events

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("테스트 시작")

class UserBehavior(SequentialTaskSet):
    auth_token = None

    def on_init(self):
        print("사용자 초기화")

    @task
    def authenticate(self):
        with self.client.post("/auth", name="인증", catch_response=True) as response:
            if response.status_code == 200:
                self.auth_token = response.json().get("token")
                response.success()
            else:
                response.failure("비정상 응답: " + response.text)

    @task(3)
    def create_item(self):
        if not self.auth_token:
            return

        payload = {"itemName": "테스트 상품"}
        headers = {"Authorization": f"Bearer {self.auth_token}"}
        with self.client.post("/create", name="상품 생성", json=payload, headers=headers, catch_response=True) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure("비정상 응답: " + response.text)

    @task
    def logout(self):
        if not self.auth_token:
            return

        with self.client.post("/logout", name="로그아웃", catch_response=True) as response:
            if response.status_code == 200:
                self.auth_token = None
                response.success()
            else:
                response.failure("비정상 응답: " + response.text)


class TimeBasedLoadShape(LoadTestShape):
    def tick(self):
        run_time = self.get_run_time()
        if run_time > 600:
            return None
        return (100, 10)


class UserSimulation(HttpUser):
    tasks = [UserBehavior]
    wait_time = constant_throughput(2)

부하 테스트 구현

class StepwiseLoadShape(LoadTestShape):
    step_duration = 10
    step_users = 10
    spawn_rate = 10
    max_users = 100

    def tick(self):
        run_time = self.get_run_time()
        current_step = int(run_time / self.step_duration) + 1
        users = current_step * self.step_users
        if users > self.max_users:
            return None
        return (users, self.spawn_rate)

class PhasedLoadShape(LoadTestShape):
    phases = [
        {"duration": 10, "users": 50, "spawn_rate": 10},
        {"duration": 20, "users": 50, "spawn_rate": 10},
        {"duration": 30, "users": 100, "spawn_rate": 10},
        {"duration": 40, "users": 150, "spawn_rate": 10},
        {"duration": 45, "users": 100, "spawn_rate": 20},
        {"duration": 50, "users": 50, "spawn_rate": 20},
        {"duration": 55, "users": 0, "spawn_rate": 20},
    ]

    def tick(self):
        run_time = self.get_run_time()
        for phase in self.phases:
            if run_time < phase["duration"]:
                return (phase["users"], phase["spawn_rate"])
        return None

고급 기능

1. 분산 환경에서의 동시 실행
2. 동적 데이터 파라미터화
3. TaskSet 계층 구조 구성