Improve compliance with `pytest-xdist`

[smarie]

trafficstars

Maybe this is not so bad: maybe there is a way to tell pytest-xdist that the group of steps have to be executed in the same distributed process ?

[aaronsmith1234]

FYI @smarie , love the project, but I just ran into this problem today. I use pytest-xdist in CI to speed things up a bit, but in my case this actually broke my tests, which locally, not using pytest-xdist, worked fine.

One thing that worked for my use case was adding the flag --dist=loadfile to my pytest call. This distributes the tests by file, and for my use case it solved the problem.

[smarie]

That's so nice of you @aaronsmith1234 to share this workaround !

It is persons like you who make open source so great :) I very much appreciate.

I'll leave the ticket open but rename it accordingly.

[linusheck]

A good way to do this is to write a new scheduler for your tests. Here's some example code (feel free to do it differently)

from xdist.remote import Producer
from xdist.scheduler.loadscope import LoadScopeScheduling

class YourScheduling(LoadScopeScheduling):
    """Implements a scheduler based on the LoadScopeScheduling. If a test using
    pytest-steps is executed, it groups the nodeids such that the tests
    depending on each other are executed on the same node. If not, it just
    defaults to executing tests on any node.
    """

    TESTS_USING_STEPS = ["test_using_steps"]

    def _split_scope(self, nodeid):
        """Determine the scope (grouping) of a nodeid.
        The nodeids for the model tests look like:
            path/to/your/test.py::test_using_steps[function_name-step]
        This function will group tests with the scope determined by splitting
        at 'model' plus the decimal number following.
         In the above example, scopes will be::
            path/to/your/test.py::test_using_steps[function_name
        """

        if any([x in nodeid for x in self.TESTS_USING_STEPS]):
            split = nodeid.split("-")
            return split[0]
        # Default: Each test is its own scope => Each test can run on any node
        return nodeid

Then use this scheduler by implementing this hook in conftest.py:

@pytest.hookimpl(trylast=True)
def pytest_xdist_make_scheduler(config, log):
    dist = config.getoption("dist")
    # Replace load (default) scheduler with YourScheduling which is a load
    # scheduler that also works for stepped tests
    schedulers = {
        "each": xdist.scheduler.EachScheduling,
        "load": YourScheduling,
        "loadscope": xdist.scheduler.LoadScopeScheduling,
        "loadfile": xdist.scheduler.LoadFileScheduling,
        "loadgroup": xdist.scheduler.LoadGroupScheduling,
    }
    return schedulers[dist](config, log)