Posts Tagged ‘pytest’
pytest development reorganization, adopt pytest month!
Today I went live with shifting pytest development to be more community driven. During a discussion at FOSDEM 2015 a significant subset of pytest contributors decided to create two organisations at github and bitbucket and shift core pytest and several plugins to them, on the initial suggestion of Anatoly Bubenkoff. See the new pytest contribution page for details. The teams currently have a dozen members and we are looking forward to integrate more contributors which all get full access and commit rights to all repositories, can push to the website and release to pypi.
Also at FOSDEM 2015, core pytest contributor Brianna Laugher suggested and started the Adopt pytest month initiative which will bring together pytest contributors, users and Open Source projects interested to use pytest in their project. Many pytest contributors and practioners will participate which means you get excellent support for bringing your testing efforts up to speed with pytest.
The pytest team is also working towards a pytest-2.7 release, take a peak at the current changelog. If you like to get something in, now is a good time to submit a pull request. Or, if you are working with a company you may contract merlinux which in turns contracts contributors to quickly resolve any issues you might have or organises in-house training or consulting. Apart from the direct benefit for your company it’s also a good way to support a sustained and ever-improving testing infrastructure commons for Python (pytest, tox, devpi projects in particular).
pytest core contributors at FOSDEM 2015 in Bruxelles, left to right: Andreas Pelme, Floris Bruynooghe, Ronny Pfannschmidt, Brianna Laugher, Holger Krekel, Anatoly Bubenkoff
Running tests against multiple devices/resources (in parallel)
How to best distribute tests against multiple devices or resources with pytest? This interesting question came up during my training in Lviv (Ukraine) at an embedded systems company. Distributing tests to processes can serve two purposes:
- running the full test suite against each device to verify they all work according to the test specification
- distributing the test load to several devices of the same type in order to minimize overall test execution time.
The solution to both problems is easy if you use two pytest facilities:
- the general fixture mechanism: we write a fixture function which provides a device object which is pre-configured for use in tests.
- the pytest-xdist plugin: we use it to run subprocesses and communicate configuration data for the device fixture from the master process to the subprocesses.
To begin with, let’s configure three devices that are each reachable by a separate IP address. We create a list of ip addresses in a file:
# content of devices.json ["192.168.0.1", "192.168.0.2", "192.168.0.3"]
We now create a local pytest plugin which reads the configuration data, implements a per-process device fixture and the master-to-slave communication to configure each subprocess according to our device list:
# content of conftest.py
import pytest
def read_device_list():
import json
with open("devices.json") as f:
return json.load(f)
def pytest_configure(config):
# read device list if we are on the master
if not hasattr(config, "slaveinput"):
config.iplist = read_device_list()
def pytest_configure_node(node):
# the master for each node fills slaveinput dictionary
# which pytest-xdist will transfer to the subprocess
node.slaveinput["ipadr"] = node.config.iplist.pop()
@pytest.fixture(scope="session")
def device(request):
slaveinput = getattr(request.config, "slaveinput", None)
if slaveinput is None: # single-process execution
ipadr = read_device_list()[0]
else: # running in a subprocess here
ipadr = slaveinput["ipadr"]
return Device(ipadr)
class Device:
def __init__(self, ipadr):
self.ipadr = ipadr
def __repr__(self):
return "<Device ip=%s>" % (self.ipadr)
We can now write tests that simply make use of the device fixture by using its name as an argument to a test function:
# content of test_device.py
import time
def test_device1(device):
time.sleep(2) # simulate long test time
assert 0, device
def test_device2(device):
time.sleep(2) # simulate long test time
assert 0, device
def test_device3(device):
time.sleep(2) # simulate long test time
assert 0, device
Let’s first run the tests in a single-process, only using a single device (also using some reporting option to shorten output):
$ py.test test_device.py -q --tb=line FFF ================================= FAILURES ================================= /tmp/doc-exec-9/test_device.py:5: AssertionError: <Device ip=192.168.0.1> /tmp/doc-exec-9/test_device.py:9: AssertionError: <Device ip=192.168.0.1> /tmp/doc-exec-9/test_device.py:13: AssertionError: <Device ip=192.168.0.1> 3 failed in 6.02 seconds
As to be expected, we get six seconds execution time (3 tests times 2 seconds each).
Now let’s run the same tests in three subprocesses, each using a different device:
$ py.test --tx 3*popen --dist=each test_device.py -q --tb=line gw0 I / gw1 I / gw2 I gw0 [3] / gw1 [3] / gw2 [3] scheduling tests via EachScheduling FFFFFFFFF ================================= FAILURES ================================= E AssertionError: <Device ip=192.168.0.1> E AssertionError: <Device ip=192.168.0.3> E AssertionError: <Device ip=192.168.0.2> E AssertionError: <Device ip=192.168.0.1> E AssertionError: <Device ip=192.168.0.3> E AssertionError: <Device ip=192.168.0.2> E AssertionError: <Device ip=192.168.0.3> E AssertionError: <Device ip=192.168.0.1> E AssertionError: <Device ip=192.168.0.2> 9 failed in 6.52 seconds
We just created three subprocesses each running three tests. Instead of 18 seconds execution time (9 tests times 2 seconds per test) we roughly got 6 seconds, a 3-times speedup. Each subprocess ran in parallel three tests against “its” device.
Let’s also run with load-balancing, i.e. distributing the tests against three different devices so that each device executes one test:
$ py.test --tx 3*popen --dist=load test_device.py -q --tb=line gw0 I / gw1 I / gw2 I gw0 [3] / gw1 [3] / gw2 [3] scheduling tests via LoadScheduling FFF ================================= FAILURES ================================= E AssertionError: <Device ip=192.168.0.3> E AssertionError: <Device ip=192.168.0.2> E AssertionError: <Device ip=192.168.0.1> 3 failed in 2.50 seconds
Here each test runs in a separate process against its device, overall more than halfing the test time compared to what it would take in a single-process (3*2=6 seconds). If we had many more tests than subproceses than load-scheduling would distribute tests in real-time to the process which has finished executing other tests.
Note that the tests themselves do not need to be aware of the distribution mode. All configuration and setup is contained in the conftest.py file.
To summarize the behaviour of the hooks and fixtures in conftest.py:
- pytest_configure(config) is called both on the master and each subprocess. We can distinguish where we are by checking for presence of config.slaveinput.
- pytest_configure_node(node) is called for each subprocess. We can fill the slaveinput dictionary which the subprocess slave can then read via its config.slaveinput dictionary.
- the device fixture only is called when a test needs it. In distributed mode, tests are only collected and executed in a subprocess. In non-distributed mode, tests are run single-process. The Device class is just a stub — it will need to grow methods for actual device communication. The tests can then simply use those device methods.
I’d like to thank Anton and the participants of my three day testing training in Lviv (Ukraine) for bringing up this and many other interesting questions.
I am giving another such professional testing course 25-27th November at the Python Academy in Leipzig. There are still two seats available. Me and other trainers can also be booked for on-site/in-house trainings worldwide.
metaprogramming and politics 