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性能压测
把两类代码完全拆分,不要写在同一个类中:
- pytest测试类仅做功能校验,不继承任何Locust类,逻辑和场景1一致
- 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
相关产品推荐
相关产品推荐

