diff options
Diffstat (limited to 'tests/conftest.py')
| -rw-r--r-- | tests/conftest.py | 54 |
1 files changed, 5 insertions, 49 deletions
diff --git a/tests/conftest.py b/tests/conftest.py index c8ca21f2..23d6b213 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -7,7 +7,6 @@ Pytest auto configuration. This module is run automatically by pytest, to define and enable fixtures. """ -import itertools import os import sys import sysconfig @@ -20,7 +19,6 @@ from coverage import env from coverage.exceptions import _StopEverything from coverage.files import set_relative_directory - # Pytest will rewrite assertions in test modules, but not elsewhere. # This tells pytest to also rewrite assertions in coveragetest.py. pytest.register_assert_rewrite("tests.coveragetest") @@ -29,6 +27,9 @@ pytest.register_assert_rewrite("tests.helpers") # Pytest can take additional options: # $set_env.py: PYTEST_ADDOPTS - Extra arguments to pytest. +pytest_plugins = "tests.balance_xdist_plugin" + + @pytest.fixture(autouse=True) def set_warnings(): """Configure warnings to show while running tests.""" @@ -92,17 +93,11 @@ def reset_filesdotpy_globals(): WORKER = os.environ.get("PYTEST_XDIST_WORKER", "none") -TRACK_TESTS = False -TEST_TXT = "/tmp/tests.txt" - def pytest_sessionstart(): """Run once at the start of the test session.""" - if TRACK_TESTS: # pragma: debugging - with open(TEST_TXT, "w") as testtxt: - print("Starting:", file=testtxt) - - # Create a .pth file for measuring subprocess coverage. + # Only in the main process... if WORKER == "none": + # Create a .pth file for measuring subprocess coverage. pth_dir = find_writable_pth_directory() assert pth_dir (pth_dir / "subcover.pth").write_text("import coverage; coverage.process_startup()\n") @@ -118,53 +113,14 @@ def pytest_sessionfinish(): if pth_file.exists(): pth_file.unlink() - -def write_test_name(prefix): - """For tracking where and when tests are running.""" - if TRACK_TESTS: # pragma: debugging - with open(TEST_TXT, "a") as testtxt: - test = os.environ.get("PYTEST_CURRENT_TEST", "unknown") - print(f"{prefix} {WORKER}: {test}", file=testtxt, flush=True) - - @pytest.hookimpl(hookwrapper=True) def pytest_runtest_call(item): """Run once for each test.""" - write_test_name(">") - # Convert _StopEverything into skipped tests. outcome = yield if outcome.excinfo and issubclass(outcome.excinfo[0], _StopEverything): # pragma: only jython pytest.skip(f"Skipping {item.nodeid} for _StopEverything: {outcome.excinfo[1]}") - write_test_name("<") - - -def interleaved(firsts, rest, n): - """Interleave the firsts among the rest so that they occur each n items.""" - num = sum(len(l) for l in firsts) + len(rest) - lists = firsts + [rest] * (n - len(firsts)) - listcycle = itertools.cycle(lists) - - while num: - alist = next(listcycle) # pylint: disable=stop-iteration-return - if alist: - yield alist.pop() - num -= 1 - - -def pytest_collection_modifyitems(items): - """Re-order the collected tests.""" - # Trick the xdist scheduler to put all of the VirtualenvTest tests on the - # same worker by sprinkling them into the collected items every Nth place. - virt = set(i for i in items if "VirtualenvTest" in i.nodeid) - rest = [i for i in items if i not in virt] - nworkers = int(os.environ.get("PYTEST_XDIST_WORKER_COUNT", 4)) - items[:] = interleaved([virt], rest, nworkers) - if TRACK_TESTS: # pragma: debugging - with open("/tmp/items.txt", "w") as f: - print("\n".join(i.nodeid for i in items), file=f) - def possible_pth_dirs(): """Produce a sequence of directories for trying to write .pth files.""" |