編寫 locust file

Locust 文件是一個普通的 python 文件。唯一的要求是它必須聲明至少一個從 User 類繼承的類。

User 用戶類

一個用戶類代表一個用戶。Locust 將為每個被模擬的用戶產(chǎn)生一個 User 類的實例南片。User 類可以定義一些常見的屬性慢叨。

wait_time 屬性

User 類的 wait _ time 方法是一個可選屬性族扰，用于確定模擬用戶在執(zhí)行任務(wù)之間應(yīng)該等待多長時間种柑。如果沒有指定等待時間岗仑，則一個新任務(wù)完成后將立即執(zhí)行。

有三個內(nèi)置的等待時間函數(shù):

• constant 固定時間
• between在最小值和最大值之間的隨機時間
• constant_pacing一個自適應(yīng)時間聚请，確保任務(wù)每 x 秒運行一次

例如荠雕，讓每個用戶在每次任務(wù)執(zhí)行之間等待0.5到10秒:

from locust import User, task, between

class MyUser(User):
    @task
    def my_task(self):
        print("executing my_task")

    wait_time = between(0.5, 10)

也可以直接在類上聲明自己的 wait _ time 方法。例如良漱，下面的 User 類將休眠一秒鐘舞虱，然后兩秒鐘，然后三秒鐘母市，等等矾兜。

class MyUser(User):
    last_wait_time = 0

    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

    ...

weight 屬性

如果文件中存在多個用戶類，可以通過命令行參數(shù)來指定在同一個 locustfile 中使用哪個用戶類:

$ locust -f locust_file.py WebUser MobileUser

如果希望模擬某種類型的更多用戶患久，可以在這些用戶類上設(shè)置一個 weight 屬性椅寺。例如，WebUser 比 MobileUser 的可能性高三倍：

class WebUser(User):
    weight = 3
    ...

class MobileUser(User):
    weight = 1
    ...

host 屬性

Host 屬性是目標(biāo)系統(tǒng)的 URL 前綴(比如“ http://google.com””)蒋失。通常返帕，這是在 Locust 的 web UI 或命令行中指定的。

如果在 User 類中聲明了一個 host 屬性篙挽，那么在命令行或 web 請求中未指定 host 的情況下將使用該host荆萤。

environment 屬性

對用戶運行environment的引用。使用它來與環(huán)境或其中包含的runner進行交互铣卡。例如链韭，從某個task方法中停止runner:

self.environment.runner.quit()

如果運行一個獨立的Locust 實例，這將停止整個運行煮落。如果在worker節(jié)點上運行敞峭，它將停止該節(jié)點。

on_start 和 on_stop 方法

User(和 TaskSet)可以聲明 on _ start 方法蝉仇、 on _ stop 方法旋讹。用戶在開始運行時調(diào)用 on _ start 方法，在停止運行時調(diào)用 on _ stop 方法轿衔。對于 TaskSet沉迹，當(dāng)模擬用戶開始執(zhí)行 TaskSet 時調(diào)用 on _ start 方法，當(dāng)模擬用戶停止執(zhí)行 TaskSet 時調(diào)用 on _ stop害驹。

Tasks 任務(wù)

當(dāng)負載測試啟動時鞭呕，將為每個模擬用戶創(chuàng)建一個 User 類的實例，在自己的協(xié)程中運行裙秋。當(dāng)這些用戶運行時琅拌，他們會一直重復(fù)“選擇待執(zhí)行任務(wù)執(zhí)行-等待-再次選擇待執(zhí)行任務(wù)執(zhí)行”這樣的過程。

@task 裝飾器

使用@task 裝飾器為 User 添加任務(wù)摘刑。

from locust import User, task, constant

class MyUser(User):
    wait_time = constant(1)

    @task
    def my_task(self):
        print("User instance (%r) executing my_task" % self)

@task 接受一個可選的 weight 參數(shù)进宝，該參數(shù)可用于指定任務(wù)的執(zhí)行權(quán)重。在下面的例子中枷恕，task2被選中的幾率是 task1的兩倍:

from locust import User, task, between

class MyUser(User):
    wait_time = between(5, 15)

    @task(3)
    def task1(self):
        pass

    @task(6)
    def task2(self):
        pass

tasks 屬性

另一種定義 User 任務(wù)的方法是設(shè)置 tasks 屬性党晋。

tasks 屬性是 Tasks 的列表，或者是一個 < Task: int > 字典徐块，其中 Task 是 python 可調(diào)用的類或 TaskSet 類未玻。

from locust import User, constant

def my_task(user):
    pass

class MyUser(User):
    tasks = [my_task]
    wait_time = constant(1)

如果將 tasks 屬性指定為列表，則每次執(zhí)行任務(wù)時胡控，都將從 tasks 屬性中隨機選擇任務(wù)扳剿。但是，如果任務(wù)是 字典——使用 callables 作為鍵昼激，使用整型 int 作為值——則將隨機選擇要執(zhí)行的任務(wù)庇绽，但使用 int 作為權(quán)重。如下所示:

{my_task: 3, another_task: 1}

my_task 被執(zhí)行的可能性是 another_task 的3倍橙困。

@ tag 裝飾器

通過使用@tag 裝飾器標(biāo)記任務(wù)瞧掺，您可以使用 -- tags 和 -- exclude-tags 參數(shù)來挑選測試期間執(zhí)行的任務(wù)。例子:

from locust import User, constant, task, tag

class MyUser(User):
    wait_time = constant(1)

    @tag('tag1')
    @task
    def task1(self):
        pass

    @tag('tag1', 'tag2')
    @task
    def task2(self):
        pass

    @tag('tag3')
    @task
    def task3(self):
        pass

    @task
    def task4(self):
        pass

如果您使用 -- tag1開始這個測試凡傅，那么在測試期間只會執(zhí)行 task1和 task2辟狈。如果以 -- tag2 tag3開始，那么只執(zhí)行 task2和 task3夏跷。

--exclude-tags 則相反哼转，是做排除選擇。因此拓春，如果以 --exclude-tags tag3開始測試释簿，則只執(zhí)行 task1、 task2和 task4硼莽。排除總是優(yōu)先于包含庶溶，因此如果一個任務(wù)包含一個您已經(jīng)包含的標(biāo)記和一個您已經(jīng)排除的標(biāo)記，那么它將不會被執(zhí)行懂鸵。

Events

如果您想運行一些設(shè)置代碼作為測試的一部分偏螺，通常只需將其放在 locustfile 的模塊級別，但有時您需要在運行過程中的特定時間做一些事情匆光。為此套像，Locust 提供了events hooks 鉤子。

test_start 和 test_stop

如果您需要在負載測試的開始或停止時運行一些代碼（注意區(qū)別里user task級別的 on_start on_stop）终息，則應(yīng)該使用 test_start 和 test_stop 事件夺巩。您可以在Locust 文件的模塊級設(shè)置這些事件的偵聽器:

from locust import events

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("A new test is starting")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("A new test is ending")

當(dāng) Locust 分布式（master-worker）運行時贞让，test _ start 和 test _ stop 事件將只在master節(jié)點中觸發(fā)。

init 初始化

init 事件在每個 Locust 進程開始時觸發(fā)柳譬。這在分布式模式中特別有用喳张，因為每個worker(不是每個用戶)都需要一次機會來進行一些初始化。例如:

from locust import events
from locust.runners import MasterRunner

@events.init.add_listener
def on_locust_init(environment, **kwargs):
    if isinstance(environment.runner, MasterRunner):
        print("I'm on master node")
    else:
        print("I'm on a worker or standalone node")

other events

... ...

HttpUser 類

HttpUser 是最常用的User美澳，它添加了一個用于發(fā)出 HTTP 請求的client屬性销部。

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(5, 15)

    @task(4)
    def index(self):
        self.client.get("/")

    @task(1)
    def about(self):
        self.client.get("/about/")

client 屬性 / HttpSession

client是 HttpSession 的一個實例。HttpSession 是requests.Session的子類/包裝器制跟。因此如果熟悉requests的小伙伴就會對此很熟悉舅桩。

client 包含所有 HTTP 方法: get、 post雨膨、 put擂涛、 ..。

就像requests.Session一樣哥放。它在請求之間會保存 cookie歼指，所以它可以很容易地用于需要登錄的使用場景。

發(fā)出一個 POST 請求甥雕，查看響應(yīng)并在第二個請求時使用已經(jīng)獲得的 session cookie

response = self.client.post("/login", {"username":"testuser", "password":"secret"})
print("Response status code:", response.status_code)
print("Response text:", response.text)
response = self.client.get("/my-profile")

捕獲任何請求踩身。Session 拋出的 RequestException (由連接錯誤、超時或類似情況引起) 社露，而是返回一個虛擬的 Response 對象挟阻，其 status _ code 設(shè)置為0，內(nèi)容設(shè)置為 None峭弟。

斷言響應(yīng)

如果 HTTP 響應(yīng)代碼是 OK 的(< 400) 附鸽，那么請求被認為是成功的，但是對響應(yīng)進行一些額外的驗證通常是有用的瞒瘸。

通過使用 catch_response 參數(shù)坷备、 with-statement 和對 response.failure ()的調(diào)用，可以將請求標(biāo)記為 failed

with self.client.get("/", catch_response=True) as response:
    if response.text != "Success":
        response.failure("Got wrong response")
    elif response.elapsed.total_seconds() > 0.5:
        response.failure("Request took too long")

你也可以將一個請求標(biāo)記為成功情臭，即使響應(yīng)代碼不正確:

with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

您甚至可以通過拋出異常省撑，然后在 with-block 外捕獲異常，從而完全避免記錄請求俯在【癸或者你可以拋出一個 Locust 異常，就像下面的例子一樣跷乐，然后讓 Locust 捕捉它肥败。

from locust.exception import RescheduleTask
...
with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        raise RescheduleTask()

使用動態(tài)參數(shù)對 URL 請求進行分組

比如一些網(wǎng)站網(wǎng)址包含某種動態(tài)參數(shù)，可以通過向 HttpSession 的request方法傳遞一個 name 參數(shù)來實現(xiàn)請求分組。

例子:

# Statistics for these requests will be grouped under: /blog/?id=[id]
for i in range(10):
    self.client.get("/blog?id=%i" % i, name="/blog?id=[id]")

代理設(shè)置

... ...

Tasksets

Tasksets 是一種結(jié)構(gòu)化等級網(wǎng)站/系統(tǒng)測試的方法馒稍。

如何組織測試代碼

對于小型測試皿哨，可以將所有測試代碼保存在一個 locustfile.py ，但是對于較大的測試套件纽谒，您可能希望將代碼分割成多個文件和目錄往史。

如何構(gòu)造測試源代碼當(dāng)然完全取決于您，但是我們建議您遵循 Python 最佳實踐佛舱。下面是一個想象中Locust 項目的文件結(jié)構(gòu)示例:

項目根目錄
│  locustfile.py
│  requirements.txt # 外部 Python 依賴項通常保存在 requirements.txt 中
│
└─common
        auth.py
        config.py
        __init__.py

有多個不同Locust 文件的項目也可以將它們保存在一個單獨的子目錄中:

項目根目錄
│  requirements.txt
│
├─common
│      auth.py
│      config.py
│      __init__.py
│
└─locustfiles
        api.py
        website.py