import pytest class TestHooks: @pytest.fixture(autouse=True) def create_test_file(self, testdir): testdir.makepyfile(""" import os def test_a(): pass def test_b(): pass def test_c(): pass """) def test_runtest_logreport(self, testdir): """Test that log reports from pytest_runtest_logreport when running with xdist contain "node", "nodeid" and "worker_id" attributes. (#8) """ testdir.makeconftest(""" def pytest_runtest_logreport(report): if hasattr(report, 'node'): if report.when == "call": workerid = report.node.workerinput['workerid'] if workerid != report.worker_id: print("HOOK: Worker id mismatch: %s %s" % (workerid, report.worker_id)) else: print("HOOK: %s %s" % (report.nodeid, report.worker_id)) """) res = testdir.runpytest('-n1', '-s') res.stdout.fnmatch_lines([ '*HOOK: test_runtest_logreport.py::test_a gw0*', '*HOOK: test_runtest_logreport.py::test_b gw0*', '*HOOK: test_runtest_logreport.py::test_c gw0*', '*3 passed*', ]) def test_node_collection_finished(self, testdir): """Test pytest_xdist_node_collection_finished hook (#8). """ testdir.makeconftest(""" def pytest_xdist_node_collection_finished(node, ids): workerid = node.workerinput['workerid'] stripped_ids = [x.split('::')[1] for x in ids] print("HOOK: %s %s" % (workerid, ', '.join(stripped_ids))) """) res = testdir.runpytest('-n2', '-s') res.stdout.fnmatch_lines_random([ '*HOOK: gw0 test_a, test_b, test_c', '*HOOK: gw1 test_a, test_b, test_c', ]) res.stdout.fnmatch_lines([ '*3 passed*', ])