Locust自定义测试场景逻辑复杂？用Python模块化封装实现可复用的测试用例

Luo Wei 2026-09-19 00:52

一、为什么你的测试场景会乱成一锅粥

好多朋友刚开始用 Locust，觉得特别简单。写个类，加两个 @task，跑起来就能看到并发数蹭蹭涨。可一旦业务逻辑复杂了，比如要登录、要加购物车、要下单、还要处理优惠券，代码就开始失控。任务一个叠一个，你中有我，我中有你。今天想改一下接口地址，得翻半天文件；明天想加一个新场景，又得复制一大段代码。最后 Locust 脚本变成了一锅粥，谁也不敢动。

1.1 从最简单的压测脚本说起

先看一个最基础的 Locust 脚本。它只做一件事：访问首页。

# 技术栈：Python 3.8 + Locust 2.x
from locust import HttpUser, task

class DemoUser(HttpUser):
    # 每个虚拟用户会执行下面的任务
    @task
    def visit_home(self):
        self.client.get("/")

这个脚本看起来挺清爽。但如果业务里有两个接口，一个首页，一个购物车，你可能就会写成这样：

# 技术栈：Python 3.8 + Locust 2.x
from locust import HttpUser, task

class DemoUser(HttpUser):
    @task(3)
    def visit_home(self):
        self.client.get("/")

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

括号里的数字是权重，3 和 1 表示首页被访问的概率是购物车的三倍。到这里，一切还都在控制之内。

1.2 场景复杂后遇到的问题

真正让人头疼的场景，往往长这样：用户打开首页，搜索商品，看详情，加购物车，下订单，最后支付。每个接口都需要带上登录后的 token，订单号又要从前一个接口的返回里取出来，还有可能用到优惠券。如果你把这些逻辑全塞进一个类里，问题就会冒出来：

登录逻辑写了三四遍，每个任务里都来一份。
token 怎么在任务之间传递？直接放全局变量又不安全。
付费用户和普通用户行为不一样，代码里全是 if else。
测试数据写死在代码里，从联调环境切到压测环境，要改一堆东西。
只想跑下单这一个场景，却发现所有任务都会执行。

这些问题不是 Locust 不够好，而是我们缺少一种组织代码的思路。思路就是模块化。

二、用模块化思维拆解压测任务

模块化的核心思想，就是把大问题切成小零件，每个零件只干一件事。比如把下单流程拆成“选商品”、“加购物车”、“提交订单”、“支付”这四个独立动作。每个动作是一个函数，函数只管自己的逻辑，对外提供参数和返回值。

2.1 把用户行为拆成独立函数

下面这段代码，把商品详情浏览封装成了一个函数。它接收一个 client 对象和商品 ID，返回 SKU 编号。

# 技术栈：Python 3.8 + Locust 2.x
def select_product(client, product_id):
    # 请求商品详情接口
    resp = client.get(f"/api/product/{product_id}")
    # 如果接口返回异常，主动抛出错误，方便压测结束后定位
    assert resp.status_code == 200, f"商品{product_id}加载失败"
    # 从JSON响应里取出SKU编号，后续加购物车会用到
    return resp.json()["sku_id"]

这里有个小技巧：把对响应结果的判断放到函数内部。这样不管谁来调用，都能保证拿到的数据是有效的，后面的流程不会被脏数据带偏。

2.2 把数据准备和业务操作分开

测试数据其实也是一类“零件”。账号、密码、商品 ID、优惠券 ID，这些东西跟真正的业务操作没有关系。把它们单独放到一个文件里，换测试环境的时候就只改这个文件。

# 技术栈：Python 3.8 + Locust 2.x
class TestData:
    # 被测系统的基地址
    BASE_URL = "https://test-api.example.com"
    # 账号信息
    USERNAME = "stress_tester"
    PASSWORD = "123456"
    # 商品信息
    PRODUCT_ID = "P10086"
    # 任务间等待时间（单位：秒）
    WAIT_MIN_SEC = 0.5
    WAIT_MAX_SEC = 2.0

有人可能会想，为什么不直接把数据写在函数里？因为一旦写在函数里，你就把“数据”和“动作”耦合在一起了。比如你想换成另外一个商品，你得去改函数的实现，这明显不划算。

2.3 使用自定义类封装场景

Locust 提供了一个叫 TaskSet 的类，专门用来组织任务流程。你可以继承它，把一组相关的任务放进同一个类里。下面这个例子展示了一个带登录态的下单任务集。

# 技术栈：Python 3.8 + Locust 2.x
from locust import TaskSet, task
from actions import select_product, add_to_cart, submit_order, pay_order
from test_data import TestData

class OrderTask(TaskSet):
    # 每个虚拟用户开始跑任务前，会先执行on_start
    def on_start(self):
        self.login()

    def login(self):
        # 登录接口，拿到访问令牌
        resp = self.client.post("/api/login", json={
            "username": TestData.USERNAME,
            "password": TestData.PASSWORD
        })
        assert resp.status_code == 200, "登录失败"
        token = resp.json()["access_token"]
        # 把令牌放到请求头里，之后的所有请求都会自动带上
        self.client.headers.update({"Authorization": f"Bearer {token}"})

    @task(2)
    def quick_buy(self):
        # 快速购买：选商品、加购、下单、支付
        sku_id = select_product(self.client, TestData.PRODUCT_ID)
        cart_id = add_to_cart(self.client, sku_id)
        order_id = submit_order(self.client, cart_id)
        pay_order(self.client, order_id)

    @task(1)
    def buy_with_coupon(self):
        # 带优惠券的购买流程
        sku_id = select_product(self.client, TestData.PRODUCT_ID)
        cart_id = add_to_cart(self.client, sku_id)
        # 这里临时获取一张优惠券
        coupon_resp = self.client.get("/api/coupon")
        coupon_id = coupon_resp.json()[0]["id"]
        order_id = submit_order(self.client, cart_id, coupon_id=coupon_id)
        pay_order(self.client, order_id)

这段代码把登录动作放到了 on_start 里。on_start 是 Locust 的钩子方法，每个虚拟用户开始干活之前会调用一次。这样做的好处是，同一个虚拟用户的大多数请求都共享同一个 token，不会每次请求都重复登录，也更接近真实用户的行为。

三、实战：一个下单流程的模块化封装

下面我们把上面的思路组合起来，做一个完整的小项目。你会看到，每个文件都很短，但加起来能支撑一个复杂的压测场景。

3.1 定义任务集

有了前面的 OrderTask，我们再把它补全一下，加上标签功能，方便在启动时筛选场景。

# 技术栈：Python 3.8 + Locust 2.x
from locust import TaskSet, task, tag
from actions import select_product, add_to_cart, submit_order, pay_order
from test_data import TestData

class OrderTask(TaskSet):
    """电商下单主流程"""

    def on_start(self):
        self.login()

    def login(self):
        # 初始化登录，提取token并注入请求头
        resp = self.client.post(
            "/api/login",
            json={"username": TestData.USERNAME, "password": TestData.PASSWORD}
        )
        assert resp.status_code == 200, "登录接口异常"
        token = resp.json().get("access_token")
        self.client.headers["Authorization"] = f"Bearer {token}"

    @task(2)
    @tag("quick")
    def quick_buy(self):
        # 无优惠券快速下单
        sku_id = select_product(self.client, TestData.PRODUCT_ID)
        cart_id = add_to_cart(self.client, sku_id)
        order_id = submit_order(self.client, cart_id)
        pay_order(self.client, order_id)

    @task(1)
    @tag("coupon")
    def buy_with_coupon(self):
        # 使用优惠券下单
        sku_id = select_product(self.client, TestData.PRODUCT_ID)
        cart_id = add_to_cart(self.client, sku_id)
        coupon_resp = self.client.get("/api/coupon")
        coupon_id = coupon_resp.json()[0]["id"]
        order_id = submit_order(self.client, cart_id, coupon_id=coupon_id)
        pay_order(self.client, order_id)

标签（@tag）是 Locust 提供的筛选工具。运行压测时，你想只跑 quick 场景，就可以用 --tags quick 参数。后面我们会看到完整命令。

3.2 封装公共请求方法

现在把业务动作集中放到 actions.py 里。这样任务集类只需要调用这些函数，不用关心每个接口的细节。

# 技术栈：Python 3.8 + Locust 2.x
def select_product(client, product_id):
    # 浏览商品详情，返回SKU编号
    resp = client.get(f"/api/product/{product_id}")
    assert resp.status_code == 200, f"商品{product_id}详情加载失败"
    return resp.json()["sku_id"]

def add_to_cart(client, sku_id):
    # 把指定SKU加入购物车
    resp = client.post("/api/cart", json={"sku_id": sku_id, "num": 1})
    assert resp.status_code == 201, "加购失败"
    return resp.json()["cart_id"]

def submit_order(client, cart_id, coupon_id=None):
    # 提交订单，如果有优惠券一并传入
    payload = {"cart_id": cart_id}
    if coupon_id:
        payload["coupon_id"] = coupon_id
    resp = client.post("/api/orders", json=payload)
    assert resp.status_code == 201, "提交订单失败"
    return resp.json()["order_id"]

def pay_order(client, order_id):
    # 对指定订单进行支付
    resp = client.post(f"/api/orders/{order_id}/pay")
    assert resp.status_code == 200, f"订单{order_id}支付失败"

注意这里每个函数都接收 client 参数，这是 Locust 的 HttpUser 提供的请求对象。你可以理解为每个虚拟用户自己的浏览器会话。

3.3 配置文件单独放

为了避免把环境相关的内容散落在各个文件里，我们把数据统一放进 test_data.py。

# 技术栈：Python 3.8 + Locust 2.x
class TestData:
    # 被测系统地址
    BASE_URL = "https://test-api.example.com"
    # 测试账号
    USERNAME = "stress_tester"
    PASSWORD = "123456"
    # 测试商品
    PRODUCT_ID = "P10086"
    # 用户操作间隔（秒），模拟真实用户思考时间
    WAIT_MIN_SEC = 0.5
    WAIT_MAX_SEC = 2.0

你可能注意到，这里的 BASE_URL 其实没有直接用到。因为 Locust 启动时可以用 --host 参数指定目标地址。那我们为什么还要写在这里？主要是为了给测试同学一个直观的入口。真正运行时，用下面的命令启动：

# 技术栈：Python 3.8 + Locust 2.x
locust -f locustfile.py --host https://test-api.example.com --tags quick

这样只跑带 quick 标签的任务。如果你想跑全部的订单场景，就去掉 --tags 参数。

3.4 用户类入口

最后是 locustfile.py。它把前面的零件组装成一个可以运行的压测脚本。

# 技术栈：Python 3.8 + Locust 2.x
from locust import HttpUser, between
from tasks import OrderTask
from test_data import TestData

class WebsiteUser(HttpUser):
    # 指定这个用户要执行的任务集
    tasks = [OrderTask]
    # 两次任务之间的随机等待时间
    wait_time = between(TestData.WAIT_MIN_SEC, TestData.WAIT_MAX_SEC)

这里有一个容易踩的坑：between 函数接收的是秒数，不是毫秒。我们之前已经在 TestData 里写成了秒，所以没问题。如果你习惯用毫秒，记得除以 1000。

四、应用场景与优缺点分析

光说模块化好不够，我们还得清楚它适合解决什么样的问题，以及有什么副作用。

4.1 适合哪些场景

业务流程长，比如电商下单、支付、金融交易，这类场景需要多个接口配合。
有多角色用户，比如普通用户和 VIP 用户，他们的操作路径不一样。
需要频繁调整接口参数，比如 token 过期后要重新登录，或者要依赖上一步的返回值。
团队一起维护压测脚本，希望代码可读、可评审。

4.2 模块化封装的好处

模块化之后，最明显的好处是代码好读了。你打开 tasks.py，直接能看到业务步骤；想看接口细节，去 actions.py；想改数据，去 test_data.py。各找各妈。

另外，复用性也强了。假设你又要压测退货流程，你不需要把下单流程再写一遍，直接在任务里调用 select_product 和 submit_order 就可以了。甚至可以把这些函数打包成 Python 包，放到其他项目里用。

模块化的另一个好处是，你可以很容易地把自己常用的函数积累成一个工具库。比如我写了一个封装登录的模块，下次另一个项目也能用，只要复制过去就行。这样一来，团队里就不需要每个人从零开始造轮子了。

4.3 需要注意的坑

模块化不是银弹，用不好也会给自己找麻烦。

第一，别过度设计。如果你的场景只有两个 GET 请求，那就老老实实写在一个类里，别为了“优雅”强行拆成四个文件。抽象是需要成本的。

第二，别把太重的东西塞进 on_start。比如登录接口本身就是一个业务请求，如果每个虚拟用户都要登录，那登录接口的压力会特别大。这在某些场景下是合理的，但如果你只想测试下单接口，登录压力反而会掩盖真实结果。可以用 Locust 的 --skip-login 之类的思路来绕过，或者在代码里加一个开关控制是否执行登录。

第三，断言别写太多。压测追求的是吞吐和耗时，不是单元测试。每个接口都来五个断言，响应分析器会浪费 CPU，影响压测结果的准确性。适当断言关键字段就够了。

第四，注意数据共享的并发问题。Locust 默认是协程并发，不是线程并发。协程之间共享同一个进程，所以全局变量是共享的。如果你在代码里写了类似 users = [] 的全局列表，多协程同时 append，可能会产生意想不到的结果。最好把数据存在每个虚拟用户自己的实例里。

第五，记得给每个任务加上合适的名字。任务名会显示在压测报告里，方便你区分各个场景的延迟和失败率。如果都用默认的类名，报告看起来会很费劲。

4.4 关联技术：参数化测试数据

真实世界里面，不可能所有用户都买同一个商品。我们可以用 CSV 文件来准备一批商品 ID，然后在任务里随机取一个。这也是模块化的一种延伸。

# 技术栈：Python 3.8 + Locust 2.x
import csv
import random
from pathlib import Path

def load_product_ids():
    # 读取CSV文件中的商品ID列表
    csv_path = Path(__file__).parent / "products.csv"
    with open(csv_path, "r", encoding="utf-8") as fp:
        reader = csv.DictReader(fp)
        return [row["product_id"] for row in reader]

# 模块加载时读一次，运行时直接取，减少IO开销
PRODUCT_IDS = load_product_ids()

def random_product():
    # 随机返回一个商品ID
    return random.choice(PRODUCT_IDS)

然后在任务里，把 TestData.PRODUCT_ID 换成 random_product()：

# 技术栈：Python 3.8 + Locust 2.x
sku_id = select_product(self.client, random_product())

这样一来，压测数据就更贴近真实了。这个技巧在复杂场景里非常实用。

五、总结

Locust 本身灵活度很高，但灵活不意味着可以乱写。通过 Python 的模块化封装，我们可以把复杂的业务场景拆成一个个细小的函数和类：动作封装成函数，场景组织成 TaskSet，数据独立成模块，再用标签控制运行范围。整个流程下来，压测脚本就像一份干净的业务代码，好读、好改、好扩展。

下次再遇到复杂的压测需求，别急着堆代码。先想想能不能拆成几个小零件，然后再动手写。你会发现，原本乱糟糟的脚本，很快就变得清清楚楚。

围绕“Locust自定义测试场景逻辑复杂？用Python模块化封装实现可复用的测试用例”参与讨论

0 条

Locust自定义测试场景逻辑复杂？用Python模块化封装实现可复用的测试用例

一、为什么你的测试场景会乱成一锅粥

1.1 从最简单的压测脚本说起

1.2 场景复杂后遇到的问题

二、用模块化思维拆解压测任务

2.1 把用户行为拆成独立函数

2.2 把数据准备和业务操作分开

2.3 使用自定义类封装场景

三、实战：一个下单流程的模块化封装

3.1 定义任务集

3.2 封装公共请求方法

3.3 配置文件单独放

3.4 用户类入口

四、应用场景与优缺点分析

4.1 适合哪些场景

4.2 模块化封装的好处

4.3 需要注意的坑

4.4 关联技术：参数化测试数据

五、总结

评论

关联文章

AI助手

一、为什么你的测试场景会乱成一锅粥

1.1 从最简单的压测脚本说起

1.2 场景复杂后遇到的问题

二、用模块化思维拆解压测任务

2.1 把用户行为拆成独立函数

2.2 把数据准备和业务操作分开

2.3 使用自定义类封装场景

三、实战：一个下单流程的模块化封装

3.1 定义任务集

3.2 封装公共请求方法

3.3 配置文件单独放

3.4 用户类入口

四、应用场景与优缺点分析

4.1 适合哪些场景

4.2 模块化封装的好处

4.3 需要注意的坑

4.4 关联技术：参数化测试数据

五、总结

同分类面试题

关联文章

AI助手