Equivalent of nose's --process-restartworker

[jiasli]

In our project, we have a process-level cache. On Windows with nose test framework, we can use --process-restartworker to isolate each ScenarioTest in a different process. However, with pytest if multiple ScenarioTests are assigned to the same process, there will be conflicts in the cache.

With pytest, is there an equivalent of --process-restartworker? If --forked doesn't work on Windows, is it possible to just restart the process instead?

Also see:
https://stackoverflow.com/questions/48234032/run-py-test-test-in-different-process
https://stackoverflow.com/questions/45462374/mark-test-to-be-run-in-independent-process
https://stackoverflow.com/questions/51187188/pytest-run-each-test-in-a-separate-process

[Zac-HD]

https://github.com/pytest-dev/pytest-forked is where this functionality would live, but nobody has volunteered to maintain it.

[nicoddemus]

Hi @jiasli,

Seems like just resetting the cache after each test would suffice? This can easily be accomplished by just using a autouse fixture on your top-level conftest.py file:

@pytest.fixture(autouse=True)
def reset_cache():
    yield
    ScenarioTest.cache.clear()

This is simple and will be faster than executing each test in its own process.

---

Having said that, out of curiosity, I wrote a simple plugin which executes each test in a separate process:

# content of conftest.py
import os
from concurrent.futures.process import ProcessPoolExecutor


def pytest_configure(config):
    config.process_pool = ProcessPoolExecutor()

def pytest_unconfigure(config):
    config.process_pool.shutdown()
    del config.process_pool
    
def pytest_report_header():
    return [f'pytest pid: {os.getpid()}']    


def pytest_pyfunc_call(pyfuncitem):
    testfunction = pyfuncitem.obj
    funcargs = pyfuncitem.funcargs
    testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}

    with ProcessPoolExecutor() as executor:
        future = executor.submit(testfunction, **testargs)
        future.result()
    return True



# content of test_foo.py
import os
import pytest

@pytest.fixture
def some_fixture():
    return {1, 2}

def test_foo(some_fixture):
    print(f'test_foo pid: {os.getpid()}')
    assert some_fixture == {1, 2}

def test_bar(some_fixture):
    print(f'test_bar pid: {os.getpid()}')
    assert some_fixture == {10, 20}

This kind of works:

λ pytest -s
======================== test session starts ========================
platform win32 -- Python 3.8.1, pytest-5.3.3.dev15+g06206bcf3.d20191226, py-1.8.1, pluggy-0.13.1
pytest pid: 25264
collected 2 items

test_foo.py test_foo pid: 27452
.test_bar pid: 25864
F

============================= FAILURES ==============================
_____________________________ test_bar ______________________________
concurrent.futures.process._RemoteTraceback:
"""
Traceback (most recent call last):
  File "C:\Users\bruno\AppData\Local\Programs\Python\Python38\lib\concurrent\futures\process.py", line 239, in _process_worker
    r = call_item.fn(*call_item.args, **call_item.kwargs)
  File "D:\projects\pytest\.tmp\subproc\test_foo.py", line 16, in test_bar
    assert some_fixture == {10, 20}
AssertionError
"""

The above exception was the direct cause of the following exception:

pyfuncitem = <Function test_bar>

    def pytest_pyfunc_call(pyfuncitem):
        testfunction = pyfuncitem.obj
        funcargs = pyfuncitem.funcargs
        testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}

        with ProcessPoolExecutor() as executor:
            future = executor.submit(testfunction, **testargs)
>           future.result()

conftest.py:23:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\..\AppData\Local\Programs\Python\Python38\lib\concurrent\futures\_base.py:439: in result
    return self.__get_result()
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <Future at 0x1d9ed63dc40 state=finished raised AssertionError>

    def __get_result(self):
        if self._exception:
>           raise self._exception
E           AssertionError

..\..\..\AppData\Local\Programs\Python\Python38\lib\concurrent\futures\_base.py:388: AssertionError
==================== 1 failed, 1 passed in 0.81s ====================

But it has a lot of caveats:

1. Test parameters and exceptions will be sent/received to the child process, and so they need to be serializable.
2. Changes in fixture values won't be reflected back to the parent.
3. This might break in several other ways other than my quick-and-dirty test can show.

This might work fine for very simple test suites with simple fixtures, but caution is warranted.

[jiasli]

Another scenario is that some dirty tests may use os.chdir to switch the working directory but don't switch back, such as (tempdir is created in advance)

class MyTestCase1(unittest.TestCase):

    def test_step2(self):
        os.chdir("tempdir")
        print(os.getcwd())


class MyTestCase2(unittest.TestCase):

    def test_step1(self):
        os.chdir("tempdir")
        print(os.getcwd())

By running them in the same process, the second one will fail:

>pytest D:\py\testdemo\testdemo.py::MyTestCase1 D:\py\testdemo\testdemo.py::MyTestCase2 -n 1
================================= test session starts =================================
platform win32 -- Python 3.8.1, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: D:\py
plugins: forked-1.1.3, timeout-1.3.4, xdist-1.31.0
gw0 [2]
.F                                                                               [100%]
====================================== FAILURES =======================================
_______________________________ MyTestCase2.test_step2 ________________________________
[gw0] win32 -- Python 3.8.1 d:\cli\env38\scripts\python.exe

self = <testdemo.MyTestCase2 testMethod=test_step2>

    def test_step2(self):
>       os.chdir("tempdir")
E       FileNotFoundError: [WinError 2] The system cannot find the file specified: 'tempdir'

..\testdemo\testdemo.py:14: FileNotFoundError
============================= 1 failed, 1 passed in 0.73s =============================

This can be solved with --forked on Linux:

$ pytest ~/py/testdemo/testdemo.py::MyTestCase1 ~/py/testdemo/testdemo.py::MyTestCase2 -n 1 --forked
=================================== test session starts ===================================
platform linux -- Python 3.8.1, pytest-4.4.2, py-1.8.1, pluggy-0.13.1
rootdir: /home/xxx/py/testdemo
plugins: timeout-1.3.4, xdist-1.31.0, forked-1.1.3
gw0 [2]
..
================================ 2 passed in 0.42 seconds =================================

On Windows, this can be overcome by nosetests with --process-restartworker:

>nosetests D:\py\testdemo\testdemo.py:MyTestCase1 D:\py\testdemo\testdemo.py:MyTestCase2 --process-restartworker --processes=1
..
----------------------------------------------------------------------
Ran 2 tests in 1.183s

OK

This will be a feature nice to have on Windows. 😃

[blueyed]

@jiasli
Your last use case works with pytest-forked:

import os


init = os.getcwd()


def test_step1():
    os.chdir("/tmp")


def test_step2():
    assert os.getcwd() == init

[jiasli]

Thanks @blueyed, yes, --forked does solve the problem on Linux.

But it doesn't work with Windows and I got:

INTERNALERROR> AssertionError: ('testdemo.py::MyTestCase1::test_step1', <WorkerController gw0>)
INTERNALERROR> assert not 'testdemo.py::MyTestCase1::test_step1'
Exception ignored in: <function ForkedFunc.__del__ at 0x0000023A47573310>
Traceback (most recent call last):
  File "d:\cli\env38\lib\site-packages\py\_process\forkedfunc.py", line 110, in __del__
    if self.pid is not None:  # only clean up in main process
AttributeError: 'ForkedFunc' object has no attribute 'pid'

[nicoddemus]

But it doesn't work with Windows and I got

Yeah it would require to spawn a new worker for every test on Windows, which would make it unnaceptable slow I think.

[jiasli]

Hi @nicoddemus, thank you for the reply. Most of our tests have network transactions which takes much longer than restarting the process. Maybe pytest can do the same as what nose does?

[nicoddemus]

Unfortunately the pytest internals was not designed be used with threads in mind, I'm afraid.

[nicoddemus]

Wait, spoke too soon: that uses a new subprocess to run a new test, is that correct? If that's the case, it would still be somewhat costly, although it seems doable by a plugin.

[jiasli]

I think so, though I am not an expert on process management. The cost is affordable to us, compared to the effort we need to make each test case nice and clean.