iT邦幫忙

2025 iThome 鐵人賽

DAY 4
2

在前面的章節中,我們介紹了如何定義使用者行為 (HttpUser)、分配任務權重 (@task) 以及設定等待時間 (wait_time)。我們也都是透過 locust -f <filename> 啟動服務,再到 Web UI (http://localhost:8089) 進行設定並開始測試。

今天,我們將深入兩個核心主題:

  1. 命令列 (CLI) 執行:如何完全脫離 Web UI,直接透過命令列參數來配置並執行負載測試,這對於自動化測試至關重要。
  2. 生命週期鉤子 (Lifecycle Hooks):如何在測試開始前與結束後,為模擬使用者執行特定的準備或清理工作。

使用命令列 (CLI) 執行 Locust

雖然 Web UI 非常直觀,但在自動化流程中(例如 CI/CD pipeline),我們不可能手動去點擊介面。因此,Locust 提供了強大的 CLI 功能,讓我們可以完全在背景執行測試。這需要使用 --headless 參數,並搭配下面這些常用 CLI 參數一起執行。

常用 CLI 參數

參數 (長) 參數 (短) 說明 範例
--locustfile -f 指定要執行的 locustfile。 -f demo.py
--host 指定被測試的主機。 --host="http://api.example.com"
--users -u 設定要模擬的總使用者數量。 -u 100
--spawn-rate -r 設定每秒產生的使用者數量。 -r 10
--run-time -t 設定測試的總執行時間。 -t 1m30s (1分30秒)
--headless 無頭模式,不啟動 Web UI,直接在終端機執行。 --headless
--html 在測試結束後,產生一份 HTML 格式的報告。 --html=report.html
--csv 將統計數據即時寫入 CSV 檔案。 --csv=stats

CLI 執行範例

假設我們要模擬 50 個使用者,每秒產生 5 個,對 http://localhost:8080 進行 1 分鐘的測試,並在結束後產生一份名為 test_report.html 的報告。

指令如下:

locust -f demo.py -u 50 -r 5 -t 1m --headless --html=test_report.html

而測試的範例如下:

from locust import HttpUser, task, constant_pacing

class WebsiteUser(HttpUser):
    wait_time = constant_pacing(5)
    host = "http://localhost:8080"

    @task(4)  # 權重設為 4
    def test_hello_world_endpoint(self):
        self.client.get("/hello-world")

執行後,您會在終端機看到即時的統計數據,類似下圖:

https://ithelp.ithome.com.tw/upload/images/20250821/201440247EXwONBgnZ.png

測試時間一到,程式會自動停止,並在專案目錄下生成 test_report.html 檔案,其內容與 Web UI 的報告頁面完全相同。

https://ithelp.ithome.com.tw/upload/images/20250821/20144024ssBVozId1J.png

生命週期鉤子:on_starton_stop

在許多測試場景中,我們需要在模擬使用者開始執行 @task 之前,先完成一些前置作業,例如登入以取得 token,或準備測試數據。同樣地,測試結束後可能需要執行登出或清理資源等操作。Locust 提供了 on_starton_stop 這兩個函式來應對這種需求。

這兩個函式都定義在 UserHttpUser 類別底下,並且每個模擬使用者 (virtual user) 在其生命週期中,都只會執行這兩個函式一次

on_start:測試開始前的準備

on_start 函式會在一個模擬使用者產生 (spawned) 後、開始執行任何 @task 之前被呼叫。

常見用途:

  • 使用者登入,並將 token 或 session cookie 存儲起來供後續任務使用。
  • 讀取測試資料。
  • 建立測試環境所需的資源。

範例:模擬使用者登入

假設我們的 API 需要 Bearer Token 進行驗證。我們可以在 on_start 中呼叫登入 API,並將獲取的 token 存到 self.token,然後在後續的 @task 中使用它。

# demo.py
from locust import HttpUser, task, between

class ApiUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080" # 假設登入和業務 API 在同一個 host

    def on_start(self):
        """在每個模擬使用者開始測試前執行"""
        print("使用者開始執行 on_start,準備登入...")
        response = self.client.post("/login", json={"username": "test_user", "password": "password"})
        
        if response.status_code == 200:
            self.token = response.json().get("token")
            print("登入成功,取得 token!")
        else:
            print("登入失敗!")
            self.token = None

    @task
    def get_profile(self):
        if not self.token:
            print("沒有 token,無法取得個人資料")
            return
            
        headers = {"Authorization": f"Bearer {self.token}"}
        self.client.get("/profile", headers=headers)
        print("正在使用 token 請求個人資料...")

on_stop:測試結束後的清理

on_stop 函式則會在測試停止時,針對每個模擬使用者執行一次。

常見用途:

  • 使用者登出。
  • 清理 on_start 中建立的數據或資源。
  • 回報或記錄該使用者的測試結果。

範例:模擬使用者登出

接續上面的例子,我們可以在測試結束時,呼叫登出 API 來清理 session。

# demo.py (接續)
class ApiUser(HttpUser):
    # ... on_start 和 @task 的程式碼 ...

    def on_stop(self):
        """在每個模擬使用者結束測試時執行"""
        print("使用者測試結束,執行 on_stop,準備登出...")
        if self.token:
            headers = {"Authorization": f"Bearer {self.token}"}
            self.client.post("/logout", headers=headers)
            print("已傳送登出請求。")

『請執行一個包含 on_starton_stop 的測試腳本。先啟動測試,讓 on_start 的 print 訊息顯示在終端機,然後點擊 Web UI 的 "STOP" 按鈕,讓 on_stop 的 print 訊息也顯示出來。請將這個過程的終端機輸出截圖。』

全域生命週期鉤子:test_starttest_stop

除了針對每個使用者的 on_starton_stop,Locust 還提供了全域的 test_starttest_stop 事件。它們的特點是:

  • 作用範圍:針對整個測試任務,而非單一使用者。
  • 執行次數:在一次完整的測試中,各自只會被執行一次。
  • 執行時機test_start 在第一個使用者產生前執行;test_stop 在所有使用者都停止後執行。
  • 定義位置:監聽函式需要定義在 User Class 之外,在 locustfile 的模組層級。

常見用途:

  • test_start: 連接資料庫、讀取大型共享設定檔、重設測試環境。
  • test_stop: 關閉資料庫連接、清理測試數據、發送測試完成通知。

範例:

# demo.py
from locust import events, HttpUser, task, between

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    """
    在測試開始時觸發,只執行一次
    """
    print("--- 全域測試已開始,準備共享資源 ---")
    # environment.runner.stats.reset_all() # 例如:手動重設統計數據

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    """
    在測試結束時觸發,只執行一次
    """
    print("--- 全域測試已結束,清理共享資源 ---")

class ApiUser(HttpUser):
    wait_time = between(1, 3)
    host = "http://localhost:8080"

    def on_start(self):
        print("一個使用者已產生 (on_start)")

    def on_stop(self):
        print("一個使用者已停止 (on_stop)")

    @task
    def some_task(self):
        print("執行任務中...")


明天,我們將深入 TaskSet,學習如何組織與控制複雜的使用者行為。我們將探索循序、巢狀的 TaskSet,比較不同的任務定義方式,並了解如何用 self.interrupt() 來中斷流程,打造更真實的測試場景。敬請期待!


上一篇
Day03 - 深入理解 Locust 的使用者行為模擬
下一篇
Day05 - 深入淺出 Task 以及 TaskSet 的使用
系列文
Vibe Coding 後的挑戰:Locust x Loki 負載及監控13
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言