Use pytest tests as locust tasks

[prestomation]

Description of issue / feature request

Automatic interpretation of pytest tests as locust tasks
I already have a comprehensive pytest setup that tests a service API. I would like to be able to annotate these tests with locust tasks or similar so I can share this code base for both functional and performance testing

Is this fundamentally possible due to locust's event system?

Expected behavior

A simple set of instructions exist that allow me to configure locust to use my existing pytest tests

Actual behavior

I have to duplicate code between my functional and locust tests.

[cgoldberg]

A simple set of instructions exist

sorry, no such instructions exist. please open a PR if you create some.

[Arching-Noradan]

Can someone provide more information? I need this too - i don't want to write each test twice - in locust and pytest+allure.

[manishaery]

This would be a great add-on, we have all our tests written in pytest and will be great if they can be consumed within locust tasks

[SpicySyntax]

+1 for this

[xinzhen2015]

+1

[cyberw]

If someone were to build this (implemented as some special User class that discovers and registers test cases as tasks?), it would be very welcome (maybe as part of locust-plugins though)

[robsonpeixoto]

This is a great feature.
I'd like to use locust to do stress test, but in some cases I'd like to only check if the API is working without stress it.

[github-actions]

This issue is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 10 days.

[github-actions]

This issue was closed because it has been stalled for 10 days with no activity. This does not necessarily mean that the issue is bad, but it most likely means that nobody is willing to take the time to fix it. If you have found Locust useful, then consider contributing a fix yourself!

[divakar260183]

I am trying to integrate pytest and locust and facing the issue PytestCollectionWarning: cannot collect test class because it has a init constructor

[anuragchauhan1989]

I am trying to integrate pytest and locust and facing the issue PytestCollectionWarning: cannot collect test class because it has a init constructor

@prestomation can you check this issue, and share a solution for this. @divakar260183 please share the repo if possible

[anuragchauhan1989]

@manishaery is there any solution on "This would be a great add-on, we have all our tests written in pytest and will be great if they can be consumed within locust tasks"

[divakar260183]

@prestomation The repo is posted at https://github.com/divakar260183/APIAutomation.git

[cyberw]

I think it is great that you are exploring this. I dont think that it is easy to do though. I think you would need to hook in to pytest on a slightly lower level.

The "cannot collect test class because it has a init constructor" is because pytest expects test classes to have a default constructor, but Locust Users (and TaskSets) require parameters (a reference to the Environment instance). I dont see an obvious solution to this so you'll need to to a bit of digging. Or maybe you could use this approach (calling pytest from inside a regular User class): https://docs.pytest.org/en/6.2.x/usage.html#calling-pytest-from-python-code

[divakar260183]

@cyberw I have already done this in my file where I am collecting all my test as a Locust task

import pytest
from locust import TaskSet, between, task, HttpUser

class TestCollector:

def __init__(self):
    self.collected = []

def pytest_collection_modifyitems(self, items):
    for item in items:
        self.collected.append(item.nodeid)

test_collector = TestCollector()

pytest.main(['LocustTest', '--collect-only'], plugins=[test_collector])

class TestTaskSet(TaskSet):

@task
def task_gen(self):
    for test in test_collector.collected:
        self.schedule_task(pytest.main(["-x", "LocustTest"]))

class WebsiteUser(HttpUser):
tasks = [TestTaskSet]
wait_time = between(1, 5)

When executing this using locust -f main.py, I am getting the error, "PytestCollectionWarning: cannot collect test class because it has a init constructor"

My Test Class is :

import json

import pytest
import requests
from locust import SequentialTaskSet

request_url = ""
module_name = "shifts"
class TestShifts(SequentialTaskSet):

def get_shifts_data(self, login, request_data):
    global request_url
    name = "Get Shifts Data"
    request_url = login.get_base_url() + request_data['path']
    with self.client.get(request_url, catch_response=True, name=name, headers=login.get_header_get_request()) \
            as get_shift_response:
        if get_shift_response.status_code == 200:
            get_shift_response.success()
            return json.loads(get_shift_response.content)
        else:
            get_shift_response.failure("Get Shifts Failure")

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

@pytest.mark.usefixtures("login", "request_data")
def test_create_shift(self, login, request_data):
    shift_config_request_url = login.get_base_url() + request_data['configPath']
    with self.client.get(shift_config_request_url, catch_response=True, name="Get Shift Config",
                         params=login.get_param(),
                         headers=login.get_header_get_request()) as shift_config_response:
        if shift_config_response.status_code == 200:
            shift_config_response_json_dict = json.loads(shift_config_response.content)
            if not shift_config_response_json_dict['isEnabled']:
                shift_config_response = requests.post(shift_config_request_url + ':enable',
                                                      params=login.get_param(),
                                                      headers=login.get_header_post_request())
                with self.client.post(shift_config_request_url + ':enable', catch_response=True,
                                      name="Enable Shift", params=login.get_param(),
                                      headers=login.get_header_get_request()) as shift_config_enable_response:
                    if shift_config_response.status_code == 200:
                        shift_config_response_json_dict = json.loads(shift_config_response.content)
                        if shift_config_response_json_dict['isEnabled'] is True:
                            shift_config_enable_response.success()
                        else:
                            shift_config_enable_response.failure("Shift Enable Failure")
    request_data['requestBody']['agentIds'][0] = login.get_agent_id()
    agent_away_statuses_request_url = login.get_base_url() + request_data['agentAwayStatusesPath']
    with self.client.get(agent_away_statuses_request_url, catch_response=True, name="Get Agent Away Statuses",
                         params=login.get_param(), headers=login.get_header_get_request()) \
            as agent_away_statuses_response:
        if agent_away_statuses_response.status_code == 200:
            agent_away_statuses_dict = json.loads(agent_away_statuses_response.content)
            for x in range(len(request_data['requestBody']['shiftWorkingHours'])):
                request_data['requestBody']['shiftWorkingHours'][x]['agentAwayStatusId'] = \
                    agent_away_statuses_dict[0]['id']
    name = "Create Shift with name :" + request_data['requestBody']['name']
    with self.client.post(request_url, catch_response=True, name=name, data=request_data['requestBody'],
                          headers=login.get_header_post_request(), params=login.get_param()) \
            as shift_creation_response:
        if shift_creation_response.status_code == 201:
            shift_creation_response_dict = json.loads(shift_creation_response.content)
            shifts_data_dict = self.get_shifts_data(login, request_data)
            request_data['requestBody']['agentIds'][0] = login.get_agent_id()
            for shift in shifts_data_dict['shifts']:
                if shift['name'] == request_data['requestBody']['name'] and shift_creation_response_dict['id'] \
                        == shift['id'] and shift['agentIds'] == request_data['requestBody']['agentIds'] \
                        and shift['timeZone'] == request_data['requestBody']['timeZone']:
                    shift_creation_response.success()
                else:
                    shift_creation_response.failure("Shift Creation Failure")

[cyberw]

I can only repeat what I said: pytest test classes (TestShifts in your case) are not allowed to have a constructor (because pytest will instantiate the classes from within its framework, and it has no idea what parameters to pass) and User/TaskSet has (and needs) a parametrized constructor. I dont know the best way to resolve this.

[cyberw]

I think the most straight forward way would be to define the User/TaskSet separately from the Test class. Of course you would then need some way to pass the HttpSession into the Test class somehow (https://docs.locust.io/en/stable/testing-requests-based%20SDK%27s.html), which might not be easy.

[divakar260183]

@cyberw Can you pleas let me know in which file the test collection strategy is defined for pytest? Do we have option to override/modify that?

[cyberw]

Sorry, you are asking the wrong person, I am in no way an expert on pytest :)

[divakar260183]

Any one have idea, please let me know.

[anuragchauhan1989]

@cyberw has anyone used the pytest api tests as locust tasks. We want to use the same pytest tests for performance testing.

[cyberw]

@cyberw has anyone used the pytest api tests as locust tasks.

No, not that I know of.

[antont]

@anuragchauhan1989

@cyberw has anyone used the pytest api tests as locust tasks. We want to use the same pytest tests for performance testing.

Some folks report or at least draft usage there on SO, but there's also a comment to one of the answers saying "This code may be outdated due to changes in locust. Locust made same breaking changes when it was upgraded to 1.0 release"

https://stackoverflow.com/questions/55914502/how-can-i-perform-load-testing-with-locust-using-pytest