You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Locust与Pytest集成触发PytestCollectionWarning警告如何解决

问题成因
  • pytest的默认收集规则:名称以Test开头的类会被识别为测试类,这类测试类不允许存在__init__构造方法,否则就会触发PytestCollectionWarning警告,跳过该类的测试用例收集。
  • 你定义的TestShifts类继承了Locust的SequentialTaskSet父类,该父类本身已经内置了__init__构造方法,哪怕你没有在子类中手动定义构造方法,继承过来的构造方法也会触发pytest的校验规则。
  • 额外逻辑错误:Locust的SequentialTaskSet是性能测试的任务编排类,不能直接作为pytest的测试类使用,你给TaskSet实例方法加的@pytest.mark.usefixtures标记完全无效,pytest不会处理Locust任务类内的fixture注入。
解决方案

根据你的实际需求选择对应方案即可:

场景1:仅需要编写pytest功能测试,不需要跑Locust性能压测

直接去掉SequentialTaskSet继承,把请求逻辑替换为普通的requests调用即可,调整后代码示例:

import json
import pytest
import requests

request_url = ""
module_name = "shifts"

# 去掉Locust父类继承
class TestShifts:
    def get_shifts_data(self, login, request_data):
        global request_url
        request_url = login.get_base_url() + request_data['path']
        get_shift_response = requests.get(request_url, headers=login.get_header_get_request())
        if get_shift_response.status_code == 200:
            return json.loads(get_shift_response.content)
        pytest.fail("Get Shifts Failure")

    @pytest.mark.usefixtures("login", "request_data")
    def test_delete_shift(self, login, request_data):
        global request_url
        shifts_data_dict = self.get_shifts_data(login, request_data)
        for shift in shifts_data_dict['shifts']:
            if shift['name'] == request_data['requestBody']['name']:
                shift_delete_response = requests.delete(
                    request_url + "/" + shift['id'], 
                    headers=login.get_header_get_request()
                )
                assert shift_delete_response.status_code == 204, "Shift Delete Failure"

场景2:需要同时支持pytest功能测试和Locust性能压测

把两类代码完全拆分,不要写在同一个类中:

  1. pytest测试类仅做功能校验,不继承任何Locust类,逻辑和场景1一致
  2. Locust任务类单独编写,类名不要以Test开头避免被pytest收集,压测需要的登录、参数信息在Locust的生命周期方法中初始化,不要使用pytest的fixture
    代码结构示例:
import json
import pytest
import requests
from locust import SequentialTaskSet, task, HttpUser

# ---------- pytest功能测试部分 ----------
class TestShifts:
    # 功能测试逻辑,和场景1写法一致
    def get_shifts_data(self, login, request_data):
        pass
    def test_delete_shift(self, login, request_data):
        pass

# ---------- Locust性能测试部分 ----------
# 类名不要以Test开头,避免被pytest识别为测试类
class ShiftsTaskSet(SequentialTaskSet):
    # 压测启动前初始化登录、参数
    def on_start(self):
        self.login_info = custom_login_logic()
        self.request_data = custom_get_param_logic()
    
    def get_shifts_data(self):
        request_url = self.login_info.get_base_url() + self.request_data['path']
        with self.client.get(request_url, catch_response=True, name="Get Shifts Data", headers=self.login_info.get_header_get_request()) as resp:
            if resp.status_code == 200:
                resp.success()
                return json.loads(resp.content)
            resp.failure("Get Shifts Failure")
    
    @task
    def delete_shift(self):
        name = "Delete Shift with name :" + self.request_data['requestBody']['name']
        shifts_data_dict = self.get_shifts_data()
        for shift in shifts_data_dict['shifts']:
            if shift['name'] == self.request_data['requestBody']['name']:
                with self.client.delete(
                    f"{self.login_info.get_base_url() + self.request_data['path']}/{shift['id']}",
                    catch_response=True, name=name, headers=self.login_info.get_header_get_request()
                ) as resp:
                    resp.success() if resp.status_code == 204 else resp.failure("Shift Delete Failure")

class RunUser(HttpUser):
    tasks = [ShiftsTaskSet]
    min_wait = 1000
    max_wait = 3000

临时规避方案

如果该TestShifts类仅用于Locust压测,不需要被pytest收集,可以直接在类中加__test__ = False标记,告诉pytest跳过该类的收集:

class TestShifts(SequentialTaskSet):
    __test__ = False # 新增该行即可
    # 其余原有代码保持不变

内容的提问来源于stack exchange,提问作者Divakar Mishra

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.02 21:54:03