diff --git a/setup.cfg b/setup.cfg index a80ac39..d09d11a 100644 --- a/setup.cfg +++ b/setup.cfg @@ -53,7 +53,6 @@ pytest11 = [options.extras_require] testing = filelock - pytest psutil = psutil>=3.0 setproctitle = setproctitle diff --git a/src/xdist/workermanage.py b/src/xdist/workermanage.py index 2c4f1a6..7f0bce2 100644 --- a/src/xdist/workermanage.py +++ b/src/xdist/workermanage.py @@ -423,9 +423,9 @@ def unserialize_warning_message(data): kwargs = {"message": message, "category": category} # access private _WARNING_DETAILS because the attributes vary between Python versions - for attr_name in warnings.WarningMessage._WARNING_DETAILS: + for attr_name in warnings.WarningMessage._WARNING_DETAILS: # type: ignore[attr-defined] if attr_name in ("message", "category"): continue kwargs[attr_name] = data[attr_name] - return warnings.WarningMessage(**kwargs) + return warnings.WarningMessage(**kwargs) # type: ignore[arg-type] diff --git a/testing/acceptance_test.py b/testing/acceptance_test.py index 3e30e45..3280aa1 100644 --- a/testing/acceptance_test.py +++ b/testing/acceptance_test.py @@ -1,90 +1,93 @@ import os import re +import shutil +from typing import Dict +from typing import List +from typing import Tuple -import py import pytest import xdist class TestDistribution: - def test_n1_pass(self, testdir): - p1 = testdir.makepyfile( + def test_n1_pass(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ def test_ok(): pass """ ) - result = testdir.runpytest(p1, "-n1") + result = pytester.runpytest(p1, "-n1") assert result.ret == 0 result.stdout.fnmatch_lines(["*1 passed*"]) - def test_n1_fail(self, testdir): - p1 = testdir.makepyfile( + def test_n1_fail(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ def test_fail(): assert 0 """ ) - result = testdir.runpytest(p1, "-n1") + result = pytester.runpytest(p1, "-n1") assert result.ret == 1 result.stdout.fnmatch_lines(["*1 failed*"]) - def test_n1_import_error(self, testdir): - p1 = testdir.makepyfile( + def test_n1_import_error(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ import __import_of_missing_module def test_import(): pass """ ) - result = testdir.runpytest(p1, "-n1") + result = pytester.runpytest(p1, "-n1") assert result.ret == 1 result.stdout.fnmatch_lines( ["E *Error: No module named *__import_of_missing_module*"] ) - def test_n2_import_error(self, testdir): + def test_n2_import_error(self, pytester: pytest.Pytester) -> None: """Check that we don't report the same import error multiple times in distributed mode.""" - p1 = testdir.makepyfile( + p1 = pytester.makepyfile( """ import __import_of_missing_module def test_import(): pass """ ) - result1 = testdir.runpytest(p1, "-n2") - result2 = testdir.runpytest(p1, "-n1") + result1 = pytester.runpytest(p1, "-n2") + result2 = pytester.runpytest(p1, "-n1") assert len(result1.stdout.lines) == len(result2.stdout.lines) - def test_n1_skip(self, testdir): - p1 = testdir.makepyfile( + def test_n1_skip(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ def test_skip(): import pytest pytest.skip("myreason") """ ) - result = testdir.runpytest(p1, "-n1") + result = pytester.runpytest(p1, "-n1") assert result.ret == 0 result.stdout.fnmatch_lines(["*1 skipped*"]) - def test_manytests_to_one_import_error(self, testdir): - p1 = testdir.makepyfile( + def test_manytests_to_one_import_error(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ import __import_of_missing_module def test_import(): pass """ ) - result = testdir.runpytest(p1, "--tx=popen", "--tx=popen") + result = pytester.runpytest(p1, "--tx=popen", "--tx=popen") assert result.ret in (1, 2) result.stdout.fnmatch_lines( ["E *Error: No module named *__import_of_missing_module*"] ) - def test_manytests_to_one_popen(self, testdir): - p1 = testdir.makepyfile( + def test_manytests_to_one_popen(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ import pytest def test_fail0(): @@ -97,12 +100,12 @@ class TestDistribution: pytest.skip("hello") """ ) - result = testdir.runpytest(p1, "-v", "-d", "--tx=popen", "--tx=popen") + result = pytester.runpytest(p1, "-v", "-d", "--tx=popen", "--tx=popen") result.stdout.fnmatch_lines(["*1*Python*", "*2 failed, 1 passed, 1 skipped*"]) assert result.ret == 1 - def test_n1_fail_minus_x(self, testdir): - p1 = testdir.makepyfile( + def test_n1_fail_minus_x(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ def test_fail1(): assert 0 @@ -110,25 +113,25 @@ class TestDistribution: assert 0 """ ) - result = testdir.runpytest(p1, "-x", "-v", "-n1") + result = pytester.runpytest(p1, "-x", "-v", "-n1") assert result.ret == 2 result.stdout.fnmatch_lines(["*Interrupted: stopping*1*", "*1 failed*"]) - def test_basetemp_in_subprocesses(self, testdir): - p1 = testdir.makepyfile( + def test_basetemp_in_subprocesses(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ - def test_send(tmpdir): - import py - assert tmpdir.relto(py.path.local(%r)), tmpdir + def test_send(tmp_path): + from pathlib import Path + assert tmp_path.relative_to(Path(%r)), tmp_path """ - % str(testdir.tmpdir) + % str(pytester.path) ) - result = testdir.runpytest_subprocess(p1, "-n1") + result = pytester.runpytest_subprocess(p1, "-n1") assert result.ret == 0 result.stdout.fnmatch_lines(["*1 passed*"]) - def test_dist_ini_specified(self, testdir): - p1 = testdir.makepyfile( + def test_dist_ini_specified(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ import pytest def test_fail0(): @@ -141,22 +144,22 @@ class TestDistribution: pytest.skip("hello") """ ) - testdir.makeini( + pytester.makeini( """ [pytest] addopts = --tx=3*popen """ ) - result = testdir.runpytest(p1, "-d", "-v") + result = pytester.runpytest(p1, "-d", "-v") result.stdout.fnmatch_lines(["*2*Python*", "*2 failed, 1 passed, 1 skipped*"]) assert result.ret == 1 @pytest.mark.xfail("sys.platform.startswith('java')", run=False) - def test_dist_tests_with_crash(self, testdir): + def test_dist_tests_with_crash(self, pytester: pytest.Pytester) -> None: if not hasattr(os, "kill"): pytest.skip("no os.kill") - p1 = testdir.makepyfile( + p1 = pytester.makepyfile( """ import pytest def test_fail0(): @@ -174,7 +177,7 @@ class TestDistribution: os.kill(os.getpid(), 15) """ ) - result = testdir.runpytest(p1, "-v", "-d", "-n1") + result = pytester.runpytest(p1, "-v", "-d", "-n1") result.stdout.fnmatch_lines( [ "*Python*", @@ -185,10 +188,12 @@ class TestDistribution: ) assert result.ret == 1 - def test_distribution_rsyncdirs_example(self, testdir, monkeypatch): + def test_distribution_rsyncdirs_example( + self, pytester: pytest.Pytester, monkeypatch + ) -> None: # use a custom plugin that has a custom command-line option to ensure # this is propagated to workers (see #491) - testdir.makepyfile( + pytester.makepyfile( **{ "myplugin/src/foobarplugin.py": """ from __future__ import print_function @@ -207,18 +212,19 @@ class TestDistribution: """ } ) - assert (testdir.tmpdir / "myplugin/src/foobarplugin.py").check(file=1) + assert (pytester.path / "myplugin/src/foobarplugin.py").is_file() monkeypatch.setenv( - "PYTHONPATH", str(testdir.tmpdir / "myplugin/src"), prepend=os.pathsep + "PYTHONPATH", str(pytester.path / "myplugin/src"), prepend=os.pathsep ) - source = testdir.mkdir("source") - dest = testdir.mkdir("dest") - subdir = source.mkdir("example_pkg") - subdir.ensure("__init__.py") - p = subdir.join("test_one.py") - p.write("def test_5():\n assert not __file__.startswith(%r)" % str(p)) - result = testdir.runpytest_subprocess( + source = pytester.mkdir("source") + dest = pytester.mkdir("dest") + subdir = source / "example_pkg" + subdir.mkdir() + subdir.joinpath("__init__.py").touch() + p = subdir / "test_one.py" + p.write_text("def test_5():\n assert not __file__.startswith(%r)" % str(p)) + result = pytester.runpytest_subprocess( "-v", "-d", "-s", @@ -239,10 +245,10 @@ class TestDistribution: ] ) result.stderr.fnmatch_lines(["--foobar=123 active! *"]) - assert dest.join(subdir.basename).check(dir=1) + assert dest.joinpath(subdir.name).is_dir() - def test_data_exchange(self, testdir): - testdir.makeconftest( + def test_data_exchange(self, pytester: pytest.Pytester) -> None: + pytester.makeconftest( """ # This hook only called on the controlling process. def pytest_configure_node(node): @@ -268,22 +274,22 @@ class TestDistribution: 'calculated result is %s' % calc_result) """ ) - p1 = testdir.makepyfile("def test_func(): pass") - result = testdir.runpytest("-v", p1, "-d", "--tx=popen") + p1 = pytester.makepyfile("def test_func(): pass") + result = pytester.runpytest("-v", p1, "-d", "--tx=popen") result.stdout.fnmatch_lines( ["*0*Python*", "*calculated result is 49*", "*1 passed*"] ) assert result.ret == 0 - def test_keyboardinterrupt_hooks_issue79(self, testdir): - testdir.makepyfile( + def test_keyboardinterrupt_hooks_issue79(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( __init__="", test_one=""" def test_hello(): raise KeyboardInterrupt() """, ) - testdir.makeconftest( + pytester.makeconftest( """ def pytest_sessionfinish(session): # on the worker @@ -296,22 +302,22 @@ class TestDistribution: """ ) args = ["-n1", "--debug"] - result = testdir.runpytest_subprocess(*args) + result = pytester.runpytest_subprocess(*args) s = result.stdout.str() assert result.ret == 2 assert "s2call" in s assert "Interrupted" in s - def test_keyboard_interrupt_dist(self, testdir): + def test_keyboard_interrupt_dist(self, pytester: pytest.Pytester) -> None: # xxx could be refined to check for return code - testdir.makepyfile( + pytester.makepyfile( """ def test_sleep(): import time time.sleep(10) """ ) - child = testdir.spawn_pytest("-n1 -v", expect_timeout=30.0) + child = pytester.spawn_pytest("-n1 -v", expect_timeout=30.0) child.expect(".*test_sleep.*") child.kill(2) # keyboard interrupt child.expect(".*KeyboardInterrupt.*") @@ -319,42 +325,42 @@ class TestDistribution: child.close() # assert ret == 2 - def test_dist_with_collectonly(self, testdir): - p1 = testdir.makepyfile( + def test_dist_with_collectonly(self, pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ def test_ok(): pass """ ) - result = testdir.runpytest(p1, "-n1", "--collect-only") + result = pytester.runpytest(p1, "-n1", "--collect-only") assert result.ret == 0 result.stdout.fnmatch_lines(["*collected 1 item*"]) class TestDistEach: - def test_simple(self, testdir): - testdir.makepyfile( + def test_simple(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ def test_hello(): pass """ ) - result = testdir.runpytest_subprocess("--debug", "--dist=each", "--tx=2*popen") + result = pytester.runpytest_subprocess("--debug", "--dist=each", "--tx=2*popen") assert not result.ret result.stdout.fnmatch_lines(["*2 pass*"]) @pytest.mark.xfail( - run=False, reason="other python versions might not have py.test installed" + run=False, reason="other python versions might not have pytest installed" ) - def test_simple_diffoutput(self, testdir): + def test_simple_diffoutput(self, pytester: pytest.Pytester) -> None: interpreters = [] for name in ("python2.5", "python2.6"): - interp = py.path.local.sysfind(name) + interp = shutil.which(name) if interp is None: pytest.skip("%s not found" % name) interpreters.append(interp) - testdir.makepyfile( + pytester.makepyfile( __init__="", test_one=""" import sys @@ -366,7 +372,7 @@ class TestDistEach: args = ["--dist=each", "-v"] args += ["--tx", "popen//python=%s" % interpreters[0]] args += ["--tx", "popen//python=%s" % interpreters[1]] - result = testdir.runpytest(*args) + result = pytester.runpytest(*args) s = result.stdout.str() assert "2...5" in s assert "2...6" in s @@ -374,8 +380,8 @@ class TestDistEach: class TestTerminalReporting: @pytest.mark.parametrize("verbosity", ["", "-q", "-v"]) - def test_output_verbosity(self, testdir, verbosity): - testdir.makepyfile( + def test_output_verbosity(self, pytester, verbosity: str) -> None: + pytester.makepyfile( """ def test_ok(): pass @@ -384,7 +390,7 @@ class TestTerminalReporting: args = ["-n1"] if verbosity: args.append(verbosity) - result = testdir.runpytest(*args) + result = pytester.runpytest(*args) out = result.stdout.str() if verbosity == "-v": assert "scheduling tests" in out @@ -397,8 +403,8 @@ class TestTerminalReporting: assert "scheduling tests" not in out assert "gw" in out - def test_pass_skip_fail(self, testdir): - testdir.makepyfile( + def test_pass_skip_fail(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ import pytest def test_ok(): @@ -409,7 +415,7 @@ class TestTerminalReporting: assert 0 """ ) - result = testdir.runpytest("-n1", "-v") + result = pytester.runpytest("-n1", "-v") result.stdout.fnmatch_lines_random( [ "*PASS*test_pass_skip_fail.py*test_ok*", @@ -421,14 +427,14 @@ class TestTerminalReporting: ["*def test_func():", "> assert 0", "E assert 0"] ) - def test_fail_platinfo(self, testdir): - testdir.makepyfile( + def test_fail_platinfo(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ def test_func(): assert 0 """ ) - result = testdir.runpytest("-n1", "-v") + result = pytester.runpytest("-n1", "-v") result.stdout.fnmatch_lines( [ "*FAIL*test_fail_platinfo.py*test_func*", @@ -439,31 +445,31 @@ class TestTerminalReporting: ] ) - def test_logfinish_hook(self, testdir): + def test_logfinish_hook(self, pytester: pytest.Pytester) -> None: """Ensure the pytest_runtest_logfinish hook is being properly handled""" from _pytest import hookspec if not hasattr(hookspec, "pytest_runtest_logfinish"): pytest.skip("test requires pytest_runtest_logfinish hook in pytest (3.4+)") - testdir.makeconftest( + pytester.makeconftest( """ def pytest_runtest_logfinish(): print('pytest_runtest_logfinish hook called') """ ) - testdir.makepyfile( + pytester.makepyfile( """ def test_func(): pass """ ) - result = testdir.runpytest("-n1", "-s") + result = pytester.runpytest("-n1", "-s") result.stdout.fnmatch_lines(["*pytest_runtest_logfinish hook called*"]) -def test_teardownfails_one_function(testdir): - p = testdir.makepyfile( +def test_teardownfails_one_function(pytester: pytest.Pytester) -> None: + p = pytester.makepyfile( """ def test_func(): pass @@ -471,15 +477,15 @@ def test_teardownfails_one_function(testdir): assert 0 """ ) - result = testdir.runpytest(p, "-n1", "--tx=popen") + result = pytester.runpytest(p, "-n1", "--tx=popen") result.stdout.fnmatch_lines( ["*def teardown_function(function):*", "*1 passed*1 error*"] ) @pytest.mark.xfail -def test_terminate_on_hangingnode(testdir): - p = testdir.makeconftest( +def test_terminate_on_hangingnode(pytester: pytest.Pytester) -> None: + p = pytester.makeconftest( """ def pytest_sessionfinish(session): if session.nodeid == "my": # running on worker @@ -487,14 +493,14 @@ def test_terminate_on_hangingnode(testdir): time.sleep(3) """ ) - result = testdir.runpytest(p, "--dist=each", "--tx=popen//id=my") + result = pytester.runpytest(p, "--dist=each", "--tx=popen//id=my") assert result.duration < 2.0 result.stdout.fnmatch_lines(["*killed*my*"]) @pytest.mark.xfail(reason="works if run outside test suite", run=False) -def test_session_hooks(testdir): - testdir.makeconftest( +def test_session_hooks(pytester: pytest.Pytester) -> None: + pytester.makeconftest( """ import sys def pytest_sessionstart(session): @@ -511,28 +517,28 @@ def test_session_hooks(testdir): raise ValueError(42) """ ) - p = testdir.makepyfile( + p = pytester.makepyfile( """ import sys def test_hello(): assert hasattr(sys, 'pytestsessionhooks') """ ) - result = testdir.runpytest(p, "--dist=each", "--tx=popen") + result = pytester.runpytest(p, "--dist=each", "--tx=popen") result.stdout.fnmatch_lines(["*ValueError*", "*1 passed*"]) assert not result.ret d = result.parseoutcomes() assert d["passed"] == 1 - assert testdir.tmpdir.join("worker").check() - assert testdir.tmpdir.join("controller").check() + assert pytester.path.joinpath("worker").exists() + assert pytester.path.joinpath("controller").exists() -def test_session_testscollected(testdir): +def test_session_testscollected(pytester: pytest.Pytester) -> None: """ Make sure controller node is updating the session object with the number of tests collected from the workers. """ - testdir.makepyfile( + pytester.makepyfile( test_foo=""" import pytest @pytest.mark.parametrize('i', range(3)) @@ -540,7 +546,7 @@ def test_session_testscollected(testdir): pass """ ) - testdir.makeconftest( + pytester.makeconftest( """ def pytest_sessionfinish(session): collected = getattr(session, 'testscollected', None) @@ -548,15 +554,15 @@ def test_session_testscollected(testdir): f.write('collected = %s' % collected) """ ) - result = testdir.inline_run("-n1") + result = pytester.inline_run("-n1") result.assertoutcome(passed=3) - collected_file = testdir.tmpdir.join("testscollected") - assert collected_file.isfile() - assert collected_file.read() == "collected = 3" + collected_file = pytester.path / "testscollected" + assert collected_file.is_file() + assert collected_file.read_text() == "collected = 3" -def test_fixture_teardown_failure(testdir): - p = testdir.makepyfile( +def test_fixture_teardown_failure(pytester: pytest.Pytester) -> None: + p = pytester.makepyfile( """ import pytest @pytest.fixture(scope="module") @@ -568,14 +574,16 @@ def test_fixture_teardown_failure(testdir): pass """ ) - result = testdir.runpytest_subprocess(p, "-n1") + result = pytester.runpytest_subprocess(p, "-n1") result.stdout.fnmatch_lines(["*ValueError*42*", "*1 passed*1 error*"]) assert result.ret -def test_config_initialization(testdir, monkeypatch, pytestconfig): +def test_config_initialization( + pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch, pytestconfig +) -> None: """Ensure workers and controller are initialized consistently. Integration test for #445""" - testdir.makepyfile( + pytester.makepyfile( **{ "dir_a/test_foo.py": """ def test_1(request): @@ -583,7 +591,7 @@ def test_config_initialization(testdir, monkeypatch, pytestconfig): """ } ) - testdir.makefile( + pytester.makefile( ".ini", myconfig=""" [pytest] @@ -591,17 +599,17 @@ def test_config_initialization(testdir, monkeypatch, pytestconfig): """, ) monkeypatch.setenv("PYTEST_ADDOPTS", "-v") - result = testdir.runpytest("-n2", "-c", "myconfig.ini", "-v") + result = pytester.runpytest("-n2", "-c", "myconfig.ini", "-v") result.stdout.fnmatch_lines(["dir_a/test_foo.py::test_1*", "*= 1 passed in *"]) assert result.ret == 0 @pytest.mark.parametrize("when", ["setup", "call", "teardown"]) -def test_crashing_item(testdir, when): +def test_crashing_item(pytester, when) -> None: """Ensure crashing item is correctly reported during all testing stages""" code = dict(setup="", call="", teardown="") code[when] = "py.process.kill(os.getpid())" - p = testdir.makepyfile( + p = pytester.makepyfile( """ import os import py @@ -624,19 +632,19 @@ def test_crashing_item(testdir, when): ) ) passes = 2 if when == "teardown" else 1 - result = testdir.runpytest("-n2", p) + result = pytester.runpytest("-n2", p) result.stdout.fnmatch_lines( ["*crashed*test_crash*", "*1 failed*%d passed*" % passes] ) -def test_multiple_log_reports(testdir): +def test_multiple_log_reports(pytester: pytest.Pytester) -> None: """ Ensure that pytest-xdist supports plugins that emit multiple logreports (#206). Inspired by pytest-rerunfailures. """ - testdir.makeconftest( + pytester.makeconftest( """ from _pytest.runner import runtestprotocol def pytest_runtest_protocol(item, nextitem): @@ -648,31 +656,31 @@ def test_multiple_log_reports(testdir): return True """ ) - testdir.makepyfile( + pytester.makepyfile( """ def test(): pass """ ) - result = testdir.runpytest("-n1") + result = pytester.runpytest("-n1") result.stdout.fnmatch_lines(["*2 passed*"]) -def test_skipping(testdir): - p = testdir.makepyfile( +def test_skipping(pytester: pytest.Pytester) -> None: + p = pytester.makepyfile( """ import pytest def test_crash(): pytest.skip("hello") """ ) - result = testdir.runpytest("-n1", "-rs", p) + result = pytester.runpytest("-n1", "-rs", p) assert result.ret == 0 result.stdout.fnmatch_lines(["*hello*", "*1 skipped*"]) -def test_fixture_scope_caching_issue503(testdir): - p1 = testdir.makepyfile( +def test_fixture_scope_caching_issue503(pytester: pytest.Pytester) -> None: + p1 = pytester.makepyfile( """ import pytest @@ -690,17 +698,17 @@ def test_fixture_scope_caching_issue503(testdir): pass """ ) - result = testdir.runpytest(p1, "-v", "-n1") + result = pytester.runpytest(p1, "-v", "-n1") assert result.ret == 0 result.stdout.fnmatch_lines(["*2 passed*"]) -def test_issue_594_random_parametrize(testdir): +def test_issue_594_random_parametrize(pytester: pytest.Pytester) -> None: """ Make sure that tests that are randomly parametrized display an appropriate error message, instead of silently skipping the entire test run. """ - p1 = testdir.makepyfile( + p1 = pytester.makepyfile( """ import pytest import random @@ -712,42 +720,42 @@ def test_issue_594_random_parametrize(testdir): assert 1 """ ) - result = testdir.runpytest(p1, "-v", "-n4") + result = pytester.runpytest(p1, "-v", "-n4") assert result.ret == 1 result.stdout.fnmatch_lines(["Different tests were collected between gw* and gw*"]) -def test_tmpdir_disabled(testdir): +def test_tmpdir_disabled(pytester: pytest.Pytester) -> None: """Test xdist doesn't break if internal tmpdir plugin is disabled (#22).""" - p1 = testdir.makepyfile( + p1 = pytester.makepyfile( """ def test_ok(): pass """ ) - result = testdir.runpytest(p1, "-n1", "-p", "no:tmpdir") + result = pytester.runpytest(p1, "-n1", "-p", "no:tmpdir") assert result.ret == 0 result.stdout.fnmatch_lines("*1 passed*") @pytest.mark.parametrize("plugin", ["xdist.looponfail", "xdist.boxed"]) -def test_sub_plugins_disabled(testdir, plugin): +def test_sub_plugins_disabled(pytester, plugin) -> None: """Test that xdist doesn't break if we disable any of its sub-plugins. (#32)""" - p1 = testdir.makepyfile( + p1 = pytester.makepyfile( """ def test_ok(): pass """ ) - result = testdir.runpytest(p1, "-n1", "-p", "no:%s" % plugin) + result = pytester.runpytest(p1, "-n1", "-p", "no:%s" % plugin) assert result.ret == 0 result.stdout.fnmatch_lines("*1 passed*") class TestWarnings: @pytest.mark.parametrize("n", ["-n0", "-n1"]) - def test_warnings(self, testdir, n): - testdir.makepyfile( + def test_warnings(self, pytester, n) -> None: + pytester.makepyfile( """ import warnings, py, pytest @@ -756,19 +764,16 @@ class TestWarnings: warnings.warn(UserWarning('this is a warning')) """ ) - result = testdir.runpytest(n) + result = pytester.runpytest(n) result.stdout.fnmatch_lines(["*this is a warning*", "*1 passed, 1 warning*"]) - def test_warning_captured_deprecated_in_pytest_6(self, testdir): + def test_warning_captured_deprecated_in_pytest_6( + self, pytester: pytest.Pytester + ) -> None: """ Do not trigger the deprecated pytest_warning_captured hook in pytest 6+ (#562) """ - import _pytest.hookspec - - if not hasattr(_pytest.hookspec, "pytest_warning_recorded"): - pytest.skip("test requires pytest 6.0+") - - testdir.makeconftest( + pytester.makeconftest( """ def pytest_warning_captured(warning_message): if warning_message == "my custom worker warning": @@ -778,23 +783,23 @@ class TestWarnings: ).format(warning_message) """ ) - testdir.makepyfile( + pytester.makepyfile( """ import warnings def test(): warnings.warn("my custom worker warning") """ ) - result = testdir.runpytest("-n1") + result = pytester.runpytest("-n1") result.stdout.fnmatch_lines(["*1 passed*"]) result.stdout.no_fnmatch_line("*this hook should not be called in this version") @pytest.mark.parametrize("n", ["-n0", "-n1"]) - def test_custom_subclass(self, testdir, n): + def test_custom_subclass(self, pytester, n) -> None: """Check that warning subclasses that don't honor the args attribute don't break pytest-xdist (#344) """ - testdir.makepyfile( + pytester.makepyfile( """ import warnings, py, pytest @@ -809,33 +814,34 @@ class TestWarnings: warnings.warn(MyWarning("foo", 1)) """ ) - testdir.syspathinsert() - result = testdir.runpytest(n) + pytester.syspathinsert() + result = pytester.runpytest(n) result.stdout.fnmatch_lines(["*MyWarning*", "*1 passed, 1 warning*"]) @pytest.mark.parametrize("n", ["-n0", "-n1"]) - def test_unserializable_arguments(self, testdir, n): + def test_unserializable_arguments(self, pytester, n) -> None: """Check that warnings with unserializable arguments are handled correctly (#349).""" - testdir.makepyfile( + pytester.makepyfile( """ import warnings, pytest - def test_func(tmpdir): - fn = (tmpdir / 'foo.txt').ensure(file=1) + def test_func(tmp_path): + fn = tmp_path / 'foo.txt' + fn.touch() with fn.open('r') as f: warnings.warn(UserWarning("foo", f)) """ ) - testdir.syspathinsert() - result = testdir.runpytest(n) + pytester.syspathinsert() + result = pytester.runpytest(n) result.stdout.fnmatch_lines(["*UserWarning*foo.txt*", "*1 passed, 1 warning*"]) @pytest.mark.parametrize("n", ["-n0", "-n1"]) - def test_unserializable_warning_details(self, testdir, n): + def test_unserializable_warning_details(self, pytester, n) -> None: """Check that warnings with unserializable _WARNING_DETAILS are handled correctly (#379). """ - testdir.makepyfile( + pytester.makepyfile( """ import warnings, pytest import socket @@ -849,28 +855,28 @@ class TestWarnings: # _WARNING_DETAIL. We need to test that it is not serialized # (it can't be, so the test will fail if we try to). @pytest.mark.filterwarnings('always') - def test_func(tmpdir): + def test_func(tmp_path): abuse_socket() gc.collect() """ ) - testdir.syspathinsert() - result = testdir.runpytest(n) + pytester.syspathinsert() + result = pytester.runpytest(n) result.stdout.fnmatch_lines( ["*ResourceWarning*unclosed*", "*1 passed, 1 warning*"] ) class TestNodeFailure: - def test_load_single(self, testdir): - f = testdir.makepyfile( + def test_load_single(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): os._exit(1) def test_b(): pass """ ) - res = testdir.runpytest(f, "-n1") + res = pytester.runpytest(f, "-n1") res.stdout.fnmatch_lines( [ "replacing crashed worker gw*", @@ -879,8 +885,8 @@ class TestNodeFailure: ] ) - def test_load_multiple(self, testdir): - f = testdir.makepyfile( + def test_load_multiple(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): pass @@ -889,7 +895,7 @@ class TestNodeFailure: def test_d(): pass """ ) - res = testdir.runpytest(f, "-n2") + res = pytester.runpytest(f, "-n2") res.stdout.fnmatch_lines( [ "replacing crashed worker gw*", @@ -898,15 +904,15 @@ class TestNodeFailure: ] ) - def test_each_single(self, testdir): - f = testdir.makepyfile( + def test_each_single(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): os._exit(1) def test_b(): pass """ ) - res = testdir.runpytest(f, "--dist=each", "--tx=popen") + res = pytester.runpytest(f, "--dist=each", "--tx=popen") res.stdout.fnmatch_lines( [ "replacing crashed worker gw*", @@ -916,15 +922,15 @@ class TestNodeFailure: ) @pytest.mark.xfail(reason="#20: xdist race condition on node restart") - def test_each_multiple(self, testdir): - f = testdir.makepyfile( + def test_each_multiple(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): os._exit(1) def test_b(): pass """ ) - res = testdir.runpytest(f, "--dist=each", "--tx=2*popen") + res = pytester.runpytest(f, "--dist=each", "--tx=2*popen") res.stdout.fnmatch_lines( [ "*Replacing crashed worker*", @@ -933,8 +939,8 @@ class TestNodeFailure: ] ) - def test_max_worker_restart(self, testdir): - f = testdir.makepyfile( + def test_max_worker_restart(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): pass @@ -943,7 +949,7 @@ class TestNodeFailure: def test_d(): pass """ ) - res = testdir.runpytest(f, "-n4", "--max-worker-restart=1") + res = pytester.runpytest(f, "-n4", "--max-worker-restart=1") res.stdout.fnmatch_lines( [ "replacing crashed worker*", @@ -954,15 +960,15 @@ class TestNodeFailure: ] ) - def test_max_worker_restart_tests_queued(self, testdir): - f = testdir.makepyfile( + def test_max_worker_restart_tests_queued(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os, pytest @pytest.mark.parametrize('i', range(10)) def test(i): os._exit(1) """ ) - res = testdir.runpytest(f, "-n2", "--max-worker-restart=3") + res = pytester.runpytest(f, "-n2", "--max-worker-restart=3") res.stdout.fnmatch_lines( [ "replacing crashed worker*", @@ -975,14 +981,14 @@ class TestNodeFailure: ) assert "INTERNALERROR" not in res.stdout.str() - def test_max_worker_restart_die(self, testdir): - f = testdir.makepyfile( + def test_max_worker_restart_die(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os os._exit(1) """ ) - res = testdir.runpytest(f, "-n4", "--max-worker-restart=0") + res = pytester.runpytest(f, "-n4", "--max-worker-restart=0") res.stdout.fnmatch_lines( [ "* xdist: worker gw* crashed and worker restarting disabled *", @@ -990,8 +996,8 @@ class TestNodeFailure: ] ) - def test_disable_restart(self, testdir): - f = testdir.makepyfile( + def test_disable_restart(self, pytester: pytest.Pytester) -> None: + f = pytester.makepyfile( """ import os def test_a(): pass @@ -999,7 +1005,7 @@ class TestNodeFailure: def test_c(): pass """ ) - res = testdir.runpytest(f, "-n4", "--max-worker-restart=0") + res = pytester.runpytest(f, "-n4", "--max-worker-restart=0") res.stdout.fnmatch_lines( [ "worker gw* crashed and worker restarting disabled", @@ -1011,10 +1017,10 @@ class TestNodeFailure: @pytest.mark.parametrize("n", [0, 2]) -def test_worker_id_fixture(testdir, n): +def test_worker_id_fixture(pytester, n) -> None: import glob - f = testdir.makepyfile( + f = pytester.makepyfile( """ import pytest @pytest.mark.parametrize("run_num", range(2)) @@ -1023,10 +1029,10 @@ def test_worker_id_fixture(testdir, n): f.write(worker_id) """ ) - result = testdir.runpytest(f, "-n%d" % n) + result = pytester.runpytest(f, "-n%d" % n) result.stdout.fnmatch_lines("* 2 passed in *") worker_ids = set() - for fname in glob.glob(str(testdir.tmpdir.join("*.txt"))): + for fname in glob.glob(str(pytester.path / "*.txt")): with open(fname) as f: worker_ids.add(f.read().strip()) if n == 0: @@ -1036,10 +1042,10 @@ def test_worker_id_fixture(testdir, n): @pytest.mark.parametrize("n", [0, 2]) -def test_testrun_uid_fixture(testdir, n): +def test_testrun_uid_fixture(pytester, n) -> None: import glob - f = testdir.makepyfile( + f = pytester.makepyfile( """ import pytest @pytest.mark.parametrize("run_num", range(2)) @@ -1048,10 +1054,10 @@ def test_testrun_uid_fixture(testdir, n): f.write(testrun_uid) """ ) - result = testdir.runpytest(f, "-n%d" % n) + result = pytester.runpytest(f, "-n%d" % n) result.stdout.fnmatch_lines("* 2 passed in *") testrun_uids = set() - for fname in glob.glob(str(testdir.tmpdir.join("*.txt"))): + for fname in glob.glob(str(pytester.path / "*.txt")): with open(fname) as f: testrun_uids.add(f.read().strip()) assert len(testrun_uids) == 1 @@ -1059,21 +1065,21 @@ def test_testrun_uid_fixture(testdir, n): @pytest.mark.parametrize("tb", ["auto", "long", "short", "no", "line", "native"]) -def test_error_report_styles(testdir, tb): - testdir.makepyfile( +def test_error_report_styles(pytester, tb) -> None: + pytester.makepyfile( """ import pytest def test_error_report_styles(): raise RuntimeError('some failure happened') """ ) - result = testdir.runpytest("-n1", "--tb=%s" % tb) + result = pytester.runpytest("-n1", "--tb=%s" % tb) if tb != "no": result.stdout.fnmatch_lines("*some failure happened*") result.assert_outcomes(failed=1) -def test_color_yes_collection_on_non_atty(testdir, request): +def test_color_yes_collection_on_non_atty(pytester, request) -> None: """skip collect progress report when working on non-terminals. Similar to pytest-dev/pytest#1397 @@ -1081,7 +1087,7 @@ def test_color_yes_collection_on_non_atty(testdir, request): tr = request.config.pluginmanager.getplugin("terminalreporter") if not hasattr(tr, "isatty"): pytest.skip("only valid for newer pytest versions") - testdir.makepyfile( + pytester.makepyfile( """ import pytest @pytest.mark.parametrize('i', range(10)) @@ -1090,34 +1096,34 @@ def test_color_yes_collection_on_non_atty(testdir, request): """ ) args = ["--color=yes", "-n2"] - result = testdir.runpytest(*args) + result = pytester.runpytest(*args) assert "test session starts" in result.stdout.str() assert "\x1b[1m" in result.stdout.str() assert "gw0 [10] / gw1 [10]" in result.stdout.str() assert "gw0 C / gw1 C" not in result.stdout.str() -def test_without_terminal_plugin(testdir, request): +def test_without_terminal_plugin(pytester, request) -> None: """ No output when terminal plugin is disabled """ - testdir.makepyfile( + pytester.makepyfile( """ def test_1(): pass """ ) - result = testdir.runpytest("-p", "no:terminal", "-n2") + result = pytester.runpytest("-p", "no:terminal", "-n2") assert result.stdout.str() == "" assert result.stderr.str() == "" assert result.ret == 0 -def test_internal_error_with_maxfail(testdir): +def test_internal_error_with_maxfail(pytester: pytest.Pytester) -> None: """ Internal error when using --maxfail option (#62, #65). """ - testdir.makepyfile( + pytester.makepyfile( """ import pytest @@ -1131,33 +1137,33 @@ def test_internal_error_with_maxfail(testdir): pass """ ) - result = testdir.runpytest_subprocess("--maxfail=1", "-n1") + result = pytester.runpytest_subprocess("--maxfail=1", "-n1") result.stdout.fnmatch_lines(["* 1 error in *"]) assert "INTERNALERROR" not in result.stderr.str() -def test_internal_errors_propagate_to_controller(testdir): - testdir.makeconftest( +def test_internal_errors_propagate_to_controller(pytester: pytest.Pytester) -> None: + pytester.makeconftest( """ def pytest_collection_modifyitems(): raise RuntimeError("Some runtime error") """ ) - testdir.makepyfile("def test(): pass") - result = testdir.runpytest("-n1") + pytester.makepyfile("def test(): pass") + result = pytester.runpytest("-n1") result.stdout.fnmatch_lines(["*RuntimeError: Some runtime error*"]) class TestLoadScope: - def test_by_module(self, testdir): + def test_by_module(self, pytester: pytest.Pytester) -> None: test_file = """ import pytest @pytest.mark.parametrize('i', range(10)) def test(i): pass """ - testdir.makepyfile(test_a=test_file, test_b=test_file) - result = testdir.runpytest("-n2", "--dist=loadscope", "-v") + pytester.makepyfile(test_a=test_file, test_b=test_file) + result = pytester.runpytest("-n2", "--dist=loadscope", "-v") assert get_workers_and_test_count_by_prefix( "test_a.py::test", result.outlines ) in ({"gw0": 10}, {"gw1": 10}) @@ -1165,8 +1171,8 @@ class TestLoadScope: "test_b.py::test", result.outlines ) in ({"gw0": 10}, {"gw1": 10}) - def test_by_class(self, testdir): - testdir.makepyfile( + def test_by_class(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( test_a=""" import pytest class TestA: @@ -1180,7 +1186,7 @@ class TestLoadScope: pass """ ) - result = testdir.runpytest("-n2", "--dist=loadscope", "-v") + result = pytester.runpytest("-n2", "--dist=loadscope", "-v") assert get_workers_and_test_count_by_prefix( "test_a.py::TestA", result.outlines ) in ({"gw0": 10}, {"gw1": 10}) @@ -1188,7 +1194,7 @@ class TestLoadScope: "test_a.py::TestB", result.outlines ) in ({"gw0": 10}, {"gw1": 10}) - def test_module_single_start(self, testdir): + def test_module_single_start(self, pytester: pytest.Pytester) -> None: """Fix test suite never finishing in case all workers start with a single test (#277).""" test_file1 = """ import pytest @@ -1202,8 +1208,8 @@ class TestLoadScope: def test_2(): pass """ - testdir.makepyfile(test_a=test_file1, test_b=test_file1, test_c=test_file2) - result = testdir.runpytest("-n2", "--dist=loadscope", "-v") + pytester.makepyfile(test_a=test_file1, test_b=test_file1, test_c=test_file2) + result = pytester.runpytest("-n2", "--dist=loadscope", "-v") a = get_workers_and_test_count_by_prefix("test_a.py::test", result.outlines) b = get_workers_and_test_count_by_prefix("test_b.py::test", result.outlines) c1 = get_workers_and_test_count_by_prefix("test_c.py::test_1", result.outlines) @@ -1215,7 +1221,7 @@ class TestLoadScope: class TestFileScope: - def test_by_module(self, testdir): + def test_by_module(self, pytester: pytest.Pytester) -> None: test_file = """ import pytest class TestA: @@ -1228,8 +1234,8 @@ class TestFileScope: def test(self, i): pass """ - testdir.makepyfile(test_a=test_file, test_b=test_file) - result = testdir.runpytest("-n2", "--dist=loadfile", "-v") + pytester.makepyfile(test_a=test_file, test_b=test_file) + result = pytester.runpytest("-n2", "--dist=loadfile", "-v") test_a_workers_and_test_count = get_workers_and_test_count_by_prefix( "test_a.py::TestA", result.outlines ) @@ -1254,8 +1260,8 @@ class TestFileScope: or test_b_workers_and_test_count in ({"gw0": 0}, {"gw1": 10}) ) - def test_by_class(self, testdir): - testdir.makepyfile( + def test_by_class(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( test_a=""" import pytest class TestA: @@ -1269,7 +1275,7 @@ class TestFileScope: pass """ ) - result = testdir.runpytest("-n2", "--dist=loadfile", "-v") + result = pytester.runpytest("-n2", "--dist=loadfile", "-v") test_a_workers_and_test_count = get_workers_and_test_count_by_prefix( "test_a.py::TestA", result.outlines ) @@ -1294,7 +1300,7 @@ class TestFileScope: or test_b_workers_and_test_count in ({"gw0": 0}, {"gw1": 10}) ) - def test_module_single_start(self, testdir): + def test_module_single_start(self, pytester: pytest.Pytester) -> None: """Fix test suite never finishing in case all workers start with a single test (#277).""" test_file1 = """ import pytest @@ -1308,8 +1314,8 @@ class TestFileScope: def test_2(): pass """ - testdir.makepyfile(test_a=test_file1, test_b=test_file1, test_c=test_file2) - result = testdir.runpytest("-n2", "--dist=loadfile", "-v") + pytester.makepyfile(test_a=test_file1, test_b=test_file1, test_c=test_file2) + result = pytester.runpytest("-n2", "--dist=loadfile", "-v") a = get_workers_and_test_count_by_prefix("test_a.py::test", result.outlines) b = get_workers_and_test_count_by_prefix("test_b.py::test", result.outlines) c1 = get_workers_and_test_count_by_prefix("test_c.py::test_1", result.outlines) @@ -1353,24 +1359,24 @@ class TestLocking: ) @pytest.mark.parametrize("scope", ["each", "load", "loadscope", "loadfile", "no"]) - def test_single_file(self, testdir, scope): - testdir.makepyfile(test_a=self.test_file1) - result = testdir.runpytest("-n2", "--dist=%s" % scope, "-v") + def test_single_file(self, pytester, scope) -> None: + pytester.makepyfile(test_a=self.test_file1) + result = pytester.runpytest("-n2", "--dist=%s" % scope, "-v") result.assert_outcomes(passed=(12 if scope != "each" else 12 * 2)) @pytest.mark.parametrize("scope", ["each", "load", "loadscope", "loadfile", "no"]) - def test_multi_file(self, testdir, scope): - testdir.makepyfile( + def test_multi_file(self, pytester, scope) -> None: + pytester.makepyfile( test_a=self.test_file1, test_b=self.test_file1, test_c=self.test_file1, test_d=self.test_file1, ) - result = testdir.runpytest("-n2", "--dist=%s" % scope, "-v") + result = pytester.runpytest("-n2", "--dist=%s" % scope, "-v") result.assert_outcomes(passed=(48 if scope != "each" else 48 * 2)) -def parse_tests_and_workers_from_output(lines): +def parse_tests_and_workers_from_output(lines: List[str]) -> List[Tuple[str, str, str]]: result = [] for line in lines: # example match: "[gw0] PASSED test_a.py::test[7]" @@ -1391,8 +1397,10 @@ def parse_tests_and_workers_from_output(lines): return result -def get_workers_and_test_count_by_prefix(prefix, lines, expected_status="PASSED"): - result = {} +def get_workers_and_test_count_by_prefix( + prefix: str, lines: List[str], expected_status: str = "PASSED" +) -> Dict[str, int]: + result: Dict[str, int] = {} for worker, status, nodeid in parse_tests_and_workers_from_output(lines): if expected_status == status and nodeid.startswith(prefix): result[worker] = result.get(worker, 0) + 1 @@ -1417,13 +1425,12 @@ class TestAPI: return FakeRequest() - def test_is_xdist_worker(self, fake_request): + def test_is_xdist_worker(self, fake_request) -> None: assert xdist.is_xdist_worker(fake_request) del fake_request.config.workerinput assert not xdist.is_xdist_worker(fake_request) - def test_is_xdist_controller(self, fake_request): - + def test_is_xdist_controller(self, fake_request) -> None: assert not xdist.is_xdist_master(fake_request) assert not xdist.is_xdist_controller(fake_request) @@ -1435,7 +1442,7 @@ class TestAPI: assert not xdist.is_xdist_master(fake_request) assert not xdist.is_xdist_controller(fake_request) - def test_get_xdist_worker_id(self, fake_request): + def test_get_xdist_worker_id(self, fake_request) -> None: assert xdist.get_xdist_worker_id(fake_request) == "gw5" del fake_request.config.workerinput assert xdist.get_xdist_worker_id(fake_request) == "master" diff --git a/testing/conftest.py b/testing/conftest.py index 52f0308..dd7293d 100644 --- a/testing/conftest.py +++ b/testing/conftest.py @@ -1,12 +1,13 @@ -import py -import pytest import execnet +import pytest +import shutil +from typing import List pytest_plugins = "pytester" @pytest.fixture(autouse=True) -def _divert_atexit(request, monkeypatch): +def _divert_atexit(request, monkeypatch: pytest.MonkeyPatch): import atexit finalizers = [] @@ -23,7 +24,7 @@ def _divert_atexit(request, monkeypatch): func(*args, **kwargs) -def pytest_addoption(parser): +def pytest_addoption(parser) -> None: parser.addoption( "--gx", action="append", @@ -33,28 +34,28 @@ def pytest_addoption(parser): @pytest.fixture -def specssh(request): +def specssh(request) -> str: return getspecssh(request.config) # configuration information for tests -def getgspecs(config): +def getgspecs(config) -> List[execnet.XSpec]: return [execnet.XSpec(spec) for spec in config.getvalueorskip("gspecs")] -def getspecssh(config): +def getspecssh(config) -> str: # type: ignore[return] xspecs = getgspecs(config) for spec in xspecs: if spec.ssh: - if not py.path.local.sysfind("ssh"): - py.test.skip("command not found: ssh") + if not shutil.which("ssh"): + pytest.skip("command not found: ssh") return str(spec) - py.test.skip("need '--gx ssh=...'") + pytest.skip("need '--gx ssh=...'") -def getsocketspec(config): +def getsocketspec(config) -> execnet.XSpec: xspecs = getgspecs(config) for spec in xspecs: if spec.socket: return spec - py.test.skip("need '--gx socket=...'") + pytest.skip("need '--gx socket=...'") diff --git a/testing/test_dsession.py b/testing/test_dsession.py index b015c75..464045e 100644 --- a/testing/test_dsession.py +++ b/testing/test_dsession.py @@ -1,59 +1,44 @@ from xdist.dsession import DSession, get_default_max_worker_restart from xdist.report import report_collection_diff from xdist.scheduler import EachScheduling, LoadScheduling +from typing import Optional -import py import pytest import execnet -XSpec = execnet.XSpec - - -def run(item, node, excinfo=None): - runner = item.config.pluginmanager.getplugin("runner") - rep = runner.ItemTestReport(item=item, excinfo=excinfo, when="call") - rep.node = node - return rep - class MockGateway: - _count = 0 - - def __init__(self): + def __init__(self) -> None: + self._count = 0 self.id = str(self._count) self._count += 1 class MockNode: - def __init__(self): - self.sent = [] + def __init__(self) -> None: + self.sent = [] # type: ignore[var-annotated] self.gateway = MockGateway() self._shutdown = False - def send_runtest_some(self, indices): + def send_runtest_some(self, indices) -> None: self.sent.extend(indices) - def send_runtest_all(self): + def send_runtest_all(self) -> None: self.sent.append("ALL") - def shutdown(self): + def shutdown(self) -> None: self._shutdown = True @property - def shutting_down(self): + def shutting_down(self) -> bool: return self._shutdown -def dumpqueue(queue): - while queue.qsize(): - print(queue.get()) - - class TestEachScheduling: - def test_schedule_load_simple(self, testdir): + def test_schedule_load_simple(self, pytester: pytest.Pytester) -> None: node1 = MockNode() node2 = MockNode() - config = testdir.parseconfig("--tx=2*popen") + config = pytester.parseconfig("--tx=2*popen") sched = EachScheduling(config) sched.add_node(node1) sched.add_node(node2) @@ -74,9 +59,9 @@ class TestEachScheduling: sched.mark_test_complete(node2, 0) assert sched.tests_finished - def test_schedule_remove_node(self, testdir): + def test_schedule_remove_node(self, pytester: pytest.Pytester) -> None: node1 = MockNode() - config = testdir.parseconfig("--tx=popen") + config = pytester.parseconfig("--tx=popen") sched = EachScheduling(config) sched.add_node(node1) collection = ["a.py::test_1"] @@ -93,8 +78,8 @@ class TestEachScheduling: class TestLoadScheduling: - def test_schedule_load_simple(self, testdir): - config = testdir.parseconfig("--tx=2*popen") + def test_schedule_load_simple(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfig("--tx=2*popen") sched = LoadScheduling(config) sched.add_node(MockNode()) sched.add_node(MockNode()) @@ -117,8 +102,8 @@ class TestLoadScheduling: sched.mark_test_complete(node1, node1.sent[0]) assert sched.tests_finished - def test_schedule_batch_size(self, testdir): - config = testdir.parseconfig("--tx=2*popen") + def test_schedule_batch_size(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfig("--tx=2*popen") sched = LoadScheduling(config) sched.add_node(MockNode()) sched.add_node(MockNode()) @@ -144,8 +129,8 @@ class TestLoadScheduling: assert node1.sent == [0, 2, 4, 5] assert not sched.pending - def test_schedule_fewer_tests_than_nodes(self, testdir): - config = testdir.parseconfig("--tx=2*popen") + def test_schedule_fewer_tests_than_nodes(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfig("--tx=2*popen") sched = LoadScheduling(config) sched.add_node(MockNode()) sched.add_node(MockNode()) @@ -164,8 +149,10 @@ class TestLoadScheduling: assert sent3 == [] assert not sched.pending - def test_schedule_fewer_than_two_tests_per_node(self, testdir): - config = testdir.parseconfig("--tx=2*popen") + def test_schedule_fewer_than_two_tests_per_node( + self, pytester: pytest.Pytester + ) -> None: + config = pytester.parseconfig("--tx=2*popen") sched = LoadScheduling(config) sched.add_node(MockNode()) sched.add_node(MockNode()) @@ -184,9 +171,9 @@ class TestLoadScheduling: assert sent3 == [2] assert not sched.pending - def test_add_remove_node(self, testdir): + def test_add_remove_node(self, pytester: pytest.Pytester) -> None: node = MockNode() - config = testdir.parseconfig("--tx=popen") + config = pytester.parseconfig("--tx=popen") sched = LoadScheduling(config) sched.add_node(node) collection = ["test_file.py::test_func"] @@ -197,7 +184,7 @@ class TestLoadScheduling: crashitem = sched.remove_node(node) assert crashitem == collection[0] - def test_different_tests_collected(self, testdir): + def test_different_tests_collected(self, pytester: pytest.Pytester) -> None: """ Test that LoadScheduling is reporting collection errors when different test ids are collected by workers. @@ -215,7 +202,7 @@ class TestLoadScheduling: self.reports.append(report) collect_hook = CollectHook() - config = testdir.parseconfig("--tx=2*popen") + config = pytester.parseconfig("--tx=2*popen") config.pluginmanager.register(collect_hook, "collect_hook") node1 = MockNode() node2 = MockNode() @@ -231,9 +218,9 @@ class TestLoadScheduling: class TestDistReporter: - @py.test.mark.xfail - def test_rsync_printing(self, testdir, linecomp): - config = testdir.parseconfig() + @pytest.mark.xfail + def test_rsync_printing(self, pytester: pytest.Pytester, linecomp) -> None: + config = pytester.parseconfig() from _pytest.pytest_terminal import TerminalReporter rep = TerminalReporter(config, file=linecomp.stringio) @@ -258,21 +245,21 @@ class TestDistReporter: # linecomp.assert_contains_lines([ # "*X1*popen*xyz*2.5*" # ]) - dsession.pytest_xdist_rsyncstart(source="hello", gateways=[gw1, gw2]) + dsession.pytest_xdist_rsyncstart(source="hello", gateways=[gw1, gw2]) # type: ignore[attr-defined] linecomp.assert_contains_lines(["[X1,X2] rsyncing: hello"]) -def test_report_collection_diff_equal(): +def test_report_collection_diff_equal() -> None: """Test reporting of equal collections.""" from_collection = to_collection = ["aaa", "bbb", "ccc"] assert report_collection_diff(from_collection, to_collection, 1, 2) is None -def test_default_max_worker_restart(): +def test_default_max_worker_restart() -> None: class config: class option: - maxworkerrestart = None - numprocesses = 0 + maxworkerrestart: Optional[str] = None + numprocesses: int = 0 assert get_default_max_worker_restart(config) is None @@ -286,7 +273,7 @@ def test_default_max_worker_restart(): assert get_default_max_worker_restart(config) == 0 -def test_report_collection_diff_different(): +def test_report_collection_diff_different() -> None: """Test reporting of different collections.""" from_collection = ["aaa", "bbb", "ccc", "YYY"] to_collection = ["aZa", "bbb", "XXX", "ccc"] @@ -311,8 +298,8 @@ def test_report_collection_diff_different(): @pytest.mark.xfail(reason="duplicate test ids not supported yet") -def test_pytest_issue419(testdir): - testdir.makepyfile( +def test_pytest_issue419(pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ import pytest @@ -321,6 +308,6 @@ def test_pytest_issue419(testdir): pass """ ) - reprec = testdir.inline_run("-n1") + reprec = pytester.inline_run("-n1") reprec.assertoutcome(passed=2) assert 0 diff --git a/testing/test_looponfail.py b/testing/test_looponfail.py index 4b69a85..02a1f59 100644 --- a/testing/test_looponfail.py +++ b/testing/test_looponfail.py @@ -1,90 +1,106 @@ import py import pytest -from pkg_resources import parse_version +import shutil +import textwrap +from pathlib import Path from xdist.looponfail import RemoteControl from xdist.looponfail import StatRecorder +PYTEST_GTE_7 = hasattr(pytest, "version_tuple") and pytest.version_tuple >= (7, 0) # type: ignore[attr-defined] + + class TestStatRecorder: - def test_filechange(self, tmpdir): - tmp = tmpdir - hello = tmp.ensure("hello.py") - sd = StatRecorder([tmp]) + def test_filechange(self, tmp_path: Path) -> None: + tmp = tmp_path + hello = tmp / "hello.py" + hello.touch() + sd = StatRecorder([py.path.local(tmp)]) changed = sd.check() assert not changed - hello.write("world") + hello.write_text("world") changed = sd.check() assert changed - (hello + "c").write("hello") + hello.with_suffix(".pyc").write_text("hello") changed = sd.check() assert not changed - p = tmp.ensure("new.py") + p = tmp / "new.py" + p.touch() changed = sd.check() assert changed - p.remove() + p.unlink() changed = sd.check() assert changed - tmp.join("a", "b", "c.py").ensure() + tmp.joinpath("a", "b").mkdir(parents=True) + tmp.joinpath("a", "b", "c.py").touch() changed = sd.check() assert changed - tmp.join("a", "c.txt").ensure() + tmp.joinpath("a", "c.txt").touch() changed = sd.check() assert changed changed = sd.check() assert not changed - tmp.join("a").remove() + shutil.rmtree(str(tmp.joinpath("a"))) changed = sd.check() assert changed - def test_dirchange(self, tmpdir): - tmp = tmpdir - tmp.ensure("dir", "hello.py") - sd = StatRecorder([tmp]) - assert not sd.fil(tmp.join("dir")) + def test_dirchange(self, tmp_path: Path) -> None: + tmp = tmp_path + tmp.joinpath("dir").mkdir() + tmp.joinpath("dir", "hello.py").touch() + sd = StatRecorder([py.path.local(tmp)]) + assert not sd.fil(py.path.local(tmp / "dir")) - def test_filechange_deletion_race(self, tmpdir, monkeypatch): - tmp = tmpdir - sd = StatRecorder([tmp]) + def test_filechange_deletion_race( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + tmp = tmp_path + pytmp = py.path.local(tmp) + sd = StatRecorder([pytmp]) changed = sd.check() assert not changed - p = tmp.ensure("new.py") + p = tmp.joinpath("new.py") + p.touch() changed = sd.check() assert changed - p.remove() + p.unlink() # make check()'s visit() call return our just removed # path as if we were in a race condition - monkeypatch.setattr(tmp, "visit", lambda *args: [p]) + monkeypatch.setattr(pytmp, "visit", lambda *args: [py.path.local(p)]) changed = sd.check() assert changed - def test_pycremoval(self, tmpdir): - tmp = tmpdir - hello = tmp.ensure("hello.py") - sd = StatRecorder([tmp]) + def test_pycremoval(self, tmp_path: Path) -> None: + tmp = tmp_path + hello = tmp / "hello.py" + hello.touch() + sd = StatRecorder([py.path.local(tmp)]) changed = sd.check() assert not changed - pycfile = hello + "c" - pycfile.ensure() - hello.write("world") + pycfile = hello.with_suffix(".pyc") + pycfile.touch() + hello.write_text("world") changed = sd.check() assert changed - assert not pycfile.check() + assert not pycfile.exists() - def test_waitonchange(self, tmpdir, monkeypatch): - tmp = tmpdir - sd = StatRecorder([tmp]) + def test_waitonchange( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + tmp = tmp_path + sd = StatRecorder([py.path.local(tmp)]) ret_values = [True, False] monkeypatch.setattr(StatRecorder, "check", lambda self: ret_values.pop()) @@ -93,63 +109,74 @@ class TestStatRecorder: class TestRemoteControl: - def test_nofailures(self, testdir): - item = testdir.getitem("def test_func(): pass\n") + def test_nofailures(self, pytester: pytest.Pytester) -> None: + item = pytester.getitem("def test_func(): pass\n") control = RemoteControl(item.config) control.setup() topdir, failures = control.runsession()[:2] assert not failures - def test_failures_somewhere(self, testdir): - item = testdir.getitem("def test_func():\n assert 0\n") + def test_failures_somewhere(self, pytester: pytest.Pytester) -> None: + item = pytester.getitem("def test_func():\n assert 0\n") control = RemoteControl(item.config) control.setup() failures = control.runsession() assert failures control.setup() - item.fspath.write("def test_func():\n assert 1\n") - removepyc(item.fspath) + item_path = item.path if PYTEST_GTE_7 else Path(item.fspath) # type: ignore[attr-defined] + item_path.write_text("def test_func():\n assert 1\n") + removepyc(item_path) topdir, failures = control.runsession()[:2] assert not failures - def test_failure_change(self, testdir): - modcol = testdir.getitem( - """ - def test_func(): - assert 0 - """ + def test_failure_change(self, pytester: pytest.Pytester) -> None: + modcol = pytester.getitem( + textwrap.dedent( + """ + def test_func(): + assert 0 + """ + ) ) control = RemoteControl(modcol.config) control.loop_once() assert control.failures - modcol.fspath.write( - py.code.Source( + modcol_path = modcol.path if PYTEST_GTE_7 else Path(modcol.fspath) # type: ignore[attr-defined] + modcol_path.write_text( + textwrap.dedent( + """ + def test_func(): + assert 1 + def test_new(): + assert 0 """ - def test_func(): - assert 1 - def test_new(): - assert 0 - """ ) ) - removepyc(modcol.fspath) + removepyc(modcol_path) control.loop_once() assert not control.failures control.loop_once() assert control.failures assert str(control.failures).find("test_new") != -1 - def test_failure_subdir_no_init(self, testdir): - modcol = testdir.getitem( - """ - def test_func(): - assert 0 - """ + def test_failure_subdir_no_init( + self, pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch + ) -> None: + modcol = pytester.getitem( + textwrap.dedent( + """ + def test_func(): + assert 0 + """ + ) ) - parent = modcol.fspath.dirpath().dirpath() - parent.chdir() + if PYTEST_GTE_7: + parent = modcol.path.parent.parent # type: ignore[attr-defined] + else: + parent = Path(modcol.fspath.dirpath().dirpath()) + monkeypatch.chdir(parent) modcol.config.args = [ - py.path.local(x).relto(parent) for x in modcol.config.args + str(Path(x).relative_to(parent)) for x in modcol.config.args ] control = RemoteControl(modcol.config) control.loop_once() @@ -159,57 +186,63 @@ class TestRemoteControl: class TestLooponFailing: - def test_looponfail_from_fail_to_ok(self, testdir): - modcol = testdir.getmodulecol( - """ - def test_one(): - x = 0 - assert x == 1 - def test_two(): - assert 1 - """ + def test_looponfail_from_fail_to_ok(self, pytester: pytest.Pytester) -> None: + modcol = pytester.getmodulecol( + textwrap.dedent( + """ + def test_one(): + x = 0 + assert x == 1 + def test_two(): + assert 1 + """ + ) ) remotecontrol = RemoteControl(modcol.config) remotecontrol.loop_once() assert len(remotecontrol.failures) == 1 - modcol.fspath.write( - py.code.Source( + modcol_path = modcol.path if PYTEST_GTE_7 else Path(modcol.fspath) + modcol_path.write_text( + textwrap.dedent( + """ + def test_one(): + assert 1 + def test_two(): + assert 1 """ - def test_one(): - assert 1 - def test_two(): - assert 1 - """ ) ) - removepyc(modcol.fspath) + removepyc(modcol_path) remotecontrol.loop_once() assert not remotecontrol.failures - def test_looponfail_from_one_to_two_tests(self, testdir): - modcol = testdir.getmodulecol( - """ - def test_one(): - assert 0 - """ + def test_looponfail_from_one_to_two_tests(self, pytester: pytest.Pytester) -> None: + modcol = pytester.getmodulecol( + textwrap.dedent( + """ + def test_one(): + assert 0 + """ + ) ) remotecontrol = RemoteControl(modcol.config) remotecontrol.loop_once() assert len(remotecontrol.failures) == 1 assert "test_one" in remotecontrol.failures[0] - modcol.fspath.write( - py.code.Source( + modcol_path = modcol.path if PYTEST_GTE_7 else Path(modcol.fspath) + modcol_path.write_text( + textwrap.dedent( + """ + def test_one(): + assert 1 # passes now + def test_two(): + assert 0 # new and fails """ - def test_one(): - assert 1 # passes now - def test_two(): - assert 0 # new and fails - """ ) ) - removepyc(modcol.fspath) + removepyc(modcol_path) remotecontrol.loop_once() assert len(remotecontrol.failures) == 0 remotecontrol.loop_once() @@ -217,47 +250,49 @@ class TestLooponFailing: assert "test_one" not in remotecontrol.failures[0] assert "test_two" in remotecontrol.failures[0] - @pytest.mark.xfail( - parse_version(pytest.__version__) >= parse_version("3.1"), - reason="broken by pytest 3.1+", - strict=True, - ) - def test_looponfail_removed_test(self, testdir): - modcol = testdir.getmodulecol( - """ - def test_one(): - assert 0 - def test_two(): - assert 0 - """ + @pytest.mark.xfail(reason="broken by pytest 3.1+", strict=True) + def test_looponfail_removed_test(self, pytester: pytest.Pytester) -> None: + modcol = pytester.getmodulecol( + textwrap.dedent( + """ + def test_one(): + assert 0 + def test_two(): + assert 0 + """ + ) ) remotecontrol = RemoteControl(modcol.config) remotecontrol.loop_once() assert len(remotecontrol.failures) == 2 - modcol.fspath.write( - py.code.Source( + modcol.path.write_text( + textwrap.dedent( + """ + def test_xxx(): # renamed test + assert 0 + def test_two(): + assert 1 # pass now """ - def test_xxx(): # renamed test - assert 0 - def test_two(): - assert 1 # pass now - """ ) ) - removepyc(modcol.fspath) + removepyc(modcol.path) remotecontrol.loop_once() assert len(remotecontrol.failures) == 0 remotecontrol.loop_once() assert len(remotecontrol.failures) == 1 - def test_looponfail_multiple_errors(self, testdir, monkeypatch): - modcol = testdir.getmodulecol( - """ - def test_one(): - assert 0 - """ + def test_looponfail_multiple_errors( + self, pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch + ) -> None: + modcol = pytester.getmodulecol( + textwrap.dedent( + """ + def test_one(): + assert 0 + """ + ) ) remotecontrol = RemoteControl(modcol.config) orig_runsession = remotecontrol.runsession @@ -274,55 +309,59 @@ class TestLooponFailing: class TestFunctional: - def test_fail_to_ok(self, testdir): - p = testdir.makepyfile( - """ - def test_one(): - x = 0 - assert x == 1 - """ + def test_fail_to_ok(self, pytester: pytest.Pytester) -> None: + p = pytester.makepyfile( + textwrap.dedent( + """ + def test_one(): + x = 0 + assert x == 1 + """ + ) ) - # p = testdir.mkdir("sub").join(p1.basename) + # p = pytester.mkdir("sub").join(p1.basename) # p1.move(p) - child = testdir.spawn_pytest("-f %s --traceconfig" % p, expect_timeout=30.0) + child = pytester.spawn_pytest("-f %s --traceconfig" % p, expect_timeout=30.0) child.expect("def test_one") child.expect("x == 1") child.expect("1 failed") child.expect("### LOOPONFAILING ####") child.expect("waiting for changes") - p.write( - py.code.Source( + p.write_text( + textwrap.dedent( """ - def test_one(): - x = 1 - assert x == 1 - """ - ) + def test_one(): + x = 1 + assert x == 1 + """ + ), ) child.expect(".*1 passed.*") child.kill(15) - def test_xfail_passes(self, testdir): - p = testdir.makepyfile( - """ - import py - @py.test.mark.xfail - def test_one(): - pass - """ + def test_xfail_passes(self, pytester: pytest.Pytester) -> None: + p = pytester.makepyfile( + textwrap.dedent( + """ + import pytest + @pytest.mark.xfail + def test_one(): + pass + """ + ) ) - child = testdir.spawn_pytest("-f %s" % p, expect_timeout=30.0) + child = pytester.spawn_pytest("-f %s" % p, expect_timeout=30.0) child.expect("1 xpass") # child.expect("### LOOPONFAILING ####") child.expect("waiting for changes") child.kill(15) -def removepyc(path): +def removepyc(path: Path) -> None: # XXX damn those pyc files - pyc = path + "c" - if pyc.check(): - pyc.remove() - c = path.dirpath("__pycache__") - if c.check(): - c.remove() + pyc = path.with_suffix(".pyc") + if pyc.exists(): + pyc.unlink() + c = path.parent / "__pycache__" + if c.exists(): + shutil.rmtree(c) diff --git a/testing/test_newhooks.py b/testing/test_newhooks.py index d2c2878..012f1ea 100644 --- a/testing/test_newhooks.py +++ b/testing/test_newhooks.py @@ -3,8 +3,8 @@ import pytest class TestHooks: @pytest.fixture(autouse=True) - def create_test_file(self, testdir): - testdir.makepyfile( + def create_test_file(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ import os def test_a(): pass @@ -13,11 +13,11 @@ class TestHooks: """ ) - def test_runtest_logreport(self, testdir): + def test_runtest_logreport(self, pytester: pytest.Pytester) -> None: """Test that log reports from pytest_runtest_logreport when running with xdist contain "node", "nodeid", "worker_id", and "testrun_uid" attributes. (#8) """ - testdir.makeconftest( + pytester.makeconftest( """ def pytest_runtest_logreport(report): if hasattr(report, 'node'): @@ -35,7 +35,7 @@ class TestHooks: % (report.nodeid, report.worker_id, report.testrun_uid)) """ ) - res = testdir.runpytest("-n1", "-s") + res = pytester.runpytest("-n1", "-s") res.stdout.fnmatch_lines( [ "*HOOK: test_runtest_logreport.py::test_a gw0 *", @@ -45,9 +45,9 @@ class TestHooks: ] ) - def test_node_collection_finished(self, testdir): + def test_node_collection_finished(self, pytester: pytest.Pytester) -> None: """Test pytest_xdist_node_collection_finished hook (#8).""" - testdir.makeconftest( + pytester.makeconftest( """ def pytest_xdist_node_collection_finished(node, ids): workerid = node.workerinput['workerid'] @@ -55,7 +55,7 @@ class TestHooks: print("HOOK: %s %s" % (workerid, ', '.join(stripped_ids))) """ ) - res = testdir.runpytest("-n2", "-s") + res = pytester.runpytest("-n2", "-s") res.stdout.fnmatch_lines_random( ["*HOOK: gw0 test_a, test_b, test_c", "*HOOK: gw1 test_a, test_b, test_c"] ) @@ -64,8 +64,8 @@ class TestHooks: class TestCrashItem: @pytest.fixture(autouse=True) - def create_test_file(self, testdir): - testdir.makepyfile( + def create_test_file(self, pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ import os def test_a(): pass @@ -75,9 +75,9 @@ class TestCrashItem: """ ) - def test_handlecrashitem(self, testdir): + def test_handlecrashitem(self, pytester: pytest.Pytester) -> None: """Test pytest_handlecrashitem hook.""" - testdir.makeconftest( + pytester.makeconftest( """ test_runs = 0 @@ -91,6 +91,6 @@ class TestCrashItem: print("HOOK: pytest_handlecrashitem") """ ) - res = testdir.runpytest("-n2", "-s") + res = pytester.runpytest("-n2", "-s") res.stdout.fnmatch_lines_random(["*HOOK: pytest_handlecrashitem"]) res.stdout.fnmatch_lines(["*3 passed*"]) diff --git a/testing/test_plugin.py b/testing/test_plugin.py index 5870676..e50c0cd 100644 --- a/testing/test_plugin.py +++ b/testing/test_plugin.py @@ -1,44 +1,46 @@ from contextlib import suppress +from pathlib import Path -import py import execnet from xdist.workermanage import NodeManager import pytest -def test_dist_incompatibility_messages(testdir): - result = testdir.runpytest("--pdb", "--looponfail") +def test_dist_incompatibility_messages(pytester: pytest.Pytester) -> None: + result = pytester.runpytest("--pdb", "--looponfail") assert result.ret != 0 - result = testdir.runpytest("--pdb", "-n", "3") + result = pytester.runpytest("--pdb", "-n", "3") assert result.ret != 0 assert "incompatible" in result.stderr.str() - result = testdir.runpytest("--pdb", "-d", "--tx", "popen") + result = pytester.runpytest("--pdb", "-d", "--tx", "popen") assert result.ret != 0 assert "incompatible" in result.stderr.str() -def test_dist_options(testdir): +def test_dist_options(pytester: pytest.Pytester) -> None: from xdist.plugin import pytest_cmdline_main as check_options - config = testdir.parseconfigure("-n 2") + config = pytester.parseconfigure("-n 2") check_options(config) assert config.option.dist == "load" assert config.option.tx == ["popen"] * 2 - config = testdir.parseconfigure("--numprocesses", "2") + config = pytester.parseconfigure("--numprocesses", "2") check_options(config) assert config.option.dist == "load" assert config.option.tx == ["popen"] * 2 - config = testdir.parseconfigure("--numprocesses", "3", "--maxprocesses", "2") + config = pytester.parseconfigure("--numprocesses", "3", "--maxprocesses", "2") check_options(config) assert config.option.dist == "load" assert config.option.tx == ["popen"] * 2 - config = testdir.parseconfigure("-d") + config = pytester.parseconfigure("-d") check_options(config) assert config.option.dist == "load" -def test_auto_detect_cpus(testdir, monkeypatch): +def test_auto_detect_cpus( + pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch +) -> None: import os from xdist.plugin import pytest_cmdline_main as check_options @@ -56,20 +58,20 @@ def test_auto_detect_cpus(testdir, monkeypatch): monkeypatch.setattr(multiprocessing, "cpu_count", lambda: 99) - config = testdir.parseconfigure("-n2") + config = pytester.parseconfigure("-n2") assert config.getoption("numprocesses") == 2 - config = testdir.parseconfigure("-nauto") + config = pytester.parseconfigure("-nauto") check_options(config) assert config.getoption("numprocesses") == 99 - config = testdir.parseconfigure("-nauto", "--pdb") + config = pytester.parseconfigure("-nauto", "--pdb") check_options(config) assert config.getoption("usepdb") assert config.getoption("numprocesses") == 0 assert config.getoption("dist") == "no" - config = testdir.parseconfigure("-nlogical", "--pdb") + config = pytester.parseconfigure("-nlogical", "--pdb") check_options(config) assert config.getoption("usepdb") assert config.getoption("numprocesses") == 0 @@ -77,91 +79,95 @@ def test_auto_detect_cpus(testdir, monkeypatch): monkeypatch.delattr(os, "sched_getaffinity", raising=False) monkeypatch.setenv("TRAVIS", "true") - config = testdir.parseconfigure("-nauto") + config = pytester.parseconfigure("-nauto") check_options(config) assert config.getoption("numprocesses") == 2 -def test_auto_detect_cpus_psutil(testdir, monkeypatch): +def test_auto_detect_cpus_psutil( + pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch +) -> None: from xdist.plugin import pytest_cmdline_main as check_options psutil = pytest.importorskip("psutil") monkeypatch.setattr(psutil, "cpu_count", lambda logical=True: 84 if logical else 42) - config = testdir.parseconfigure("-nauto") + config = pytester.parseconfigure("-nauto") check_options(config) assert config.getoption("numprocesses") == 42 - config = testdir.parseconfigure("-nlogical") + config = pytester.parseconfigure("-nlogical") check_options(config) assert config.getoption("numprocesses") == 84 -def test_hook_auto_num_workers(testdir, monkeypatch): +def test_hook_auto_num_workers( + pytester: pytest.Pytester, monkeypatch: pytest.MonkeyPatch +) -> None: from xdist.plugin import pytest_cmdline_main as check_options - testdir.makeconftest( + pytester.makeconftest( """ def pytest_xdist_auto_num_workers(): return 42 """ ) - config = testdir.parseconfigure("-nauto") + config = pytester.parseconfigure("-nauto") check_options(config) assert config.getoption("numprocesses") == 42 - config = testdir.parseconfigure("-nlogical") + config = pytester.parseconfigure("-nlogical") check_options(config) assert config.getoption("numprocesses") == 42 -def test_boxed_with_collect_only(testdir): +def test_boxed_with_collect_only(pytester: pytest.Pytester) -> None: from xdist.plugin import pytest_cmdline_main as check_options - config = testdir.parseconfigure("-n1", "--boxed") + config = pytester.parseconfigure("-n1", "--boxed") check_options(config) assert config.option.forked - config = testdir.parseconfigure("-n1", "--collect-only") + config = pytester.parseconfigure("-n1", "--collect-only") check_options(config) assert not config.option.forked - config = testdir.parseconfigure("-n1", "--boxed", "--collect-only") + config = pytester.parseconfigure("-n1", "--boxed", "--collect-only") check_options(config) assert config.option.forked -def test_dsession_with_collect_only(testdir): +def test_dsession_with_collect_only(pytester: pytest.Pytester) -> None: from xdist.plugin import pytest_cmdline_main as check_options from xdist.plugin import pytest_configure as configure - config = testdir.parseconfigure("-n1") + config = pytester.parseconfigure("-n1") check_options(config) configure(config) assert config.pluginmanager.hasplugin("dsession") - config = testdir.parseconfigure("-n1", "--collect-only") + config = pytester.parseconfigure("-n1", "--collect-only") check_options(config) configure(config) assert not config.pluginmanager.hasplugin("dsession") -def test_testrunuid_provided(testdir): - config = testdir.parseconfigure("--testrunuid", "test123", "--tx=popen") +def test_testrunuid_provided(pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--testrunuid", "test123", "--tx=popen") nm = NodeManager(config) assert nm.testrunuid == "test123" -def test_testrunuid_generated(testdir): - config = testdir.parseconfigure("--tx=popen") +def test_testrunuid_generated(pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--tx=popen") nm = NodeManager(config) assert len(nm.testrunuid) == 32 class TestDistOptions: - def test_getxspecs(self, testdir): - config = testdir.parseconfigure("--tx=popen", "--tx", "ssh=xyz") + def test_getxspecs(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--tx=popen", "--tx", "ssh=xyz") nodemanager = NodeManager(config) xspecs = nodemanager._getxspecs() assert len(xspecs) == 2 @@ -169,39 +175,39 @@ class TestDistOptions: assert xspecs[0].popen assert xspecs[1].ssh == "xyz" - def test_xspecs_multiplied(self, testdir): - config = testdir.parseconfigure("--tx=3*popen") + def test_xspecs_multiplied(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--tx=3*popen") xspecs = NodeManager(config)._getxspecs() assert len(xspecs) == 3 assert xspecs[1].popen - def test_getrsyncdirs(self, testdir): - config = testdir.parseconfigure("--rsyncdir=" + str(testdir.tmpdir)) + def test_getrsyncdirs(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--rsyncdir=" + str(pytester.path)) nm = NodeManager(config, specs=[execnet.XSpec("popen")]) assert not nm._getrsyncdirs() nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")]) assert nm.roots - assert testdir.tmpdir in nm.roots + assert pytester.path in nm.roots - def test_getrsyncignore(self, testdir): - config = testdir.parseconfigure("--rsyncignore=fo*") + def test_getrsyncignore(self, pytester: pytest.Pytester) -> None: + config = pytester.parseconfigure("--rsyncignore=fo*") nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=qwe")]) assert "fo*" in nm.rsyncoptions["ignores"] - def test_getrsyncdirs_with_conftest(self, testdir): - p = py.path.local() - for bn in "x y z".split(): - p.mkdir(bn) - testdir.makeini( + def test_getrsyncdirs_with_conftest(self, pytester: pytest.Pytester) -> None: + p = Path.cwd() + for bn in ("x", "y", "z"): + p.joinpath(bn).mkdir() + pytester.makeini( """ [pytest] rsyncdirs= x """ ) - config = testdir.parseconfigure(testdir.tmpdir, "--rsyncdir=y", "--rsyncdir=z") + config = pytester.parseconfigure(pytester.path, "--rsyncdir=y", "--rsyncdir=z") nm = NodeManager(config, specs=[execnet.XSpec("popen//chdir=xyz")]) roots = nm._getrsyncdirs() # assert len(roots) == 3 + 1 # pylib - assert py.path.local("y") in roots - assert py.path.local("z") in roots - assert testdir.tmpdir.join("x") in roots + assert Path("y").resolve() in roots + assert Path("z").resolve() in roots + assert pytester.path.joinpath("x") in roots diff --git a/testing/test_remote.py b/testing/test_remote.py index 31a6a2a..348febc 100644 --- a/testing/test_remote.py +++ b/testing/test_remote.py @@ -1,5 +1,5 @@ -import py import pprint +import py import pytest import sys import uuid @@ -32,18 +32,16 @@ class EventCall: class WorkerSetup: use_callback = False - def __init__(self, request, testdir): + def __init__(self, request, pytester: pytest.Pytester) -> None: self.request = request - self.testdir = testdir - self.events = Queue() + self.pytester = pytester + self.events = Queue() # type: ignore[var-annotated] - def setup( - self, - ): - self.testdir.chdir() + def setup(self) -> None: + self.pytester.chdir() # import os ; os.environ['EXECNET_DEBUG'] = "2" self.gateway = execnet.makegateway() - self.config = config = self.testdir.parseconfigure() + self.config = config = self.pytester.parseconfigure() putevent = self.use_callback and self.events.put or None class DummyMananger: @@ -70,15 +68,15 @@ class WorkerSetup: @pytest.fixture -def worker(request, testdir): - return WorkerSetup(request, testdir) +def worker(request, pytester: pytest.Pytester) -> WorkerSetup: + return WorkerSetup(request, pytester) @pytest.mark.xfail(reason="#59") -def test_remoteinitconfig(testdir): +def test_remoteinitconfig(pytester: pytest.Pytester) -> None: from xdist.remote import remote_initconfig - config1 = testdir.parseconfig() + config1 = pytester.parseconfig() config2 = remote_initconfig(config1.option.__dict__, config1.args) assert config2.option.__dict__ == config1.option.__dict__ assert config2.pluginmanager.getplugin("terminal") in (-1, None) @@ -94,8 +92,10 @@ class TestWorkerInteractor: return unserialize - def test_basic_collect_and_runtests(self, worker, unserialize_report): - worker.testdir.makepyfile( + def test_basic_collect_and_runtests( + self, worker: WorkerSetup, unserialize_report + ) -> None: + worker.pytester.makepyfile( """ def test_func(): pass @@ -108,7 +108,7 @@ class TestWorkerInteractor: assert ev.name == "collectionstart" assert not ev.kwargs ev = worker.popevent("collectionfinish") - assert ev.kwargs["topdir"] == worker.testdir.tmpdir + assert ev.kwargs["topdir"] == py.path.local(worker.pytester.path) ids = ev.kwargs["ids"] assert len(ids) == 1 worker.sendcommand("runtests", indices=list(range(len(ids)))) @@ -126,8 +126,8 @@ class TestWorkerInteractor: ev = worker.popevent("workerfinished") assert "workeroutput" in ev.kwargs - def test_remote_collect_skip(self, worker, unserialize_report): - worker.testdir.makepyfile( + def test_remote_collect_skip(self, worker: WorkerSetup, unserialize_report) -> None: + worker.pytester.makepyfile( """ import pytest pytest.skip("hello", allow_module_level=True) @@ -144,8 +144,8 @@ class TestWorkerInteractor: ev = worker.popevent("collectionfinish") assert not ev.kwargs["ids"] - def test_remote_collect_fail(self, worker, unserialize_report): - worker.testdir.makepyfile("""aasd qwe""") + def test_remote_collect_fail(self, worker: WorkerSetup, unserialize_report) -> None: + worker.pytester.makepyfile("""aasd qwe""") worker.setup() ev = worker.popevent("collectionstart") assert not ev.kwargs @@ -156,8 +156,8 @@ class TestWorkerInteractor: ev = worker.popevent("collectionfinish") assert not ev.kwargs["ids"] - def test_runtests_all(self, worker, unserialize_report): - worker.testdir.makepyfile( + def test_runtests_all(self, worker: WorkerSetup, unserialize_report) -> None: + worker.pytester.makepyfile( """ def test_func(): pass def test_func2(): pass @@ -183,17 +183,19 @@ class TestWorkerInteractor: ev = worker.popevent("workerfinished") assert "workeroutput" in ev.kwargs - def test_happy_run_events_converted(self, testdir, worker): - py.test.xfail("implement a simple test for event production") - assert not worker.use_callback - worker.testdir.makepyfile( + def test_happy_run_events_converted( + self, pytester: pytest.Pytester, worker: WorkerSetup + ) -> None: + pytest.xfail("implement a simple test for event production") + assert not worker.use_callback # type: ignore[unreachable] + worker.pytester.makepyfile( """ def test_func(): pass """ ) worker.setup() - hookrec = testdir.getreportrecorder(worker.config) + hookrec = pytester.getreportrecorder(worker.config) for data in worker.slp.channel: worker.slp.process_from_remote(data) worker.slp.process_from_remote(worker.slp.ENDMARK) @@ -209,7 +211,9 @@ class TestWorkerInteractor: ] ) - def test_process_from_remote_error_handling(self, worker, capsys): + def test_process_from_remote_error_handling( + self, worker: WorkerSetup, capsys: pytest.CaptureFixture[str] + ) -> None: worker.use_callback = True worker.setup() worker.slp.process_from_remote(("", ())) @@ -219,8 +223,8 @@ class TestWorkerInteractor: assert ev.name == "errordown" -def test_remote_env_vars(testdir): - testdir.makepyfile( +def test_remote_env_vars(pytester: pytest.Pytester) -> None: + pytester.makepyfile( """ import os def test(): @@ -229,13 +233,13 @@ def test_remote_env_vars(testdir): assert os.environ['PYTEST_XDIST_WORKER_COUNT'] == '2' """ ) - result = testdir.runpytest("-n2", "--max-worker-restart=0") + result = pytester.runpytest("-n2", "--max-worker-restart=0") assert result.ret == 0 -def test_remote_inner_argv(testdir): +def test_remote_inner_argv(pytester: pytest.Pytester) -> None: """Test/document the behavior due to execnet using `python -c`.""" - testdir.makepyfile( + pytester.makepyfile( """ import sys @@ -243,14 +247,14 @@ def test_remote_inner_argv(testdir): assert sys.argv == ["-c"] """ ) - result = testdir.runpytest("-n1") + result = pytester.runpytest("-n1") assert result.ret == 0 -def test_remote_mainargv(testdir): +def test_remote_mainargv(pytester: pytest.Pytester) -> None: outer_argv = sys.argv - testdir.makepyfile( + pytester.makepyfile( """ def test_mainargv(request): assert request.config.workerinput["mainargv"] == {!r} @@ -258,14 +262,14 @@ def test_remote_mainargv(testdir): outer_argv ) ) - result = testdir.runpytest("-n1") + result = pytester.runpytest("-n1") assert result.ret == 0 -def test_remote_usage_prog(testdir, request): +def test_remote_usage_prog(pytester: pytest.Pytester, request) -> None: if not hasattr(request.config._parser, "prog"): pytest.skip("prog not available in config parser") - testdir.makeconftest( + pytester.makeconftest( """ import pytest @@ -280,7 +284,7 @@ def test_remote_usage_prog(testdir, request): config_parser = config._parser """ ) - testdir.makepyfile( + pytester.makepyfile( """ import sys @@ -289,14 +293,14 @@ def test_remote_usage_prog(testdir, request): """ ) - result = testdir.runpytest_subprocess("-n1") + result = pytester.runpytest_subprocess("-n1") assert result.ret == 1 result.stdout.fnmatch_lines(["*usage: *", "*error: my_usage_error"]) -def test_remote_sys_path(testdir): +def test_remote_sys_path(pytester: pytest.Pytester) -> None: """Work around sys.path differences due to execnet using `python -c`.""" - testdir.makepyfile( + pytester.makepyfile( """ import sys @@ -304,5 +308,5 @@ def test_remote_sys_path(testdir): assert "" not in sys.path """ ) - result = testdir.runpytest("-n1") + result = pytester.runpytest("-n1") assert result.ret == 0 diff --git a/testing/test_workermanage.py b/testing/test_workermanage.py index 3cf19a8..fae0601 100644 --- a/testing/test_workermanage.py +++ b/testing/test_workermanage.py @@ -1,39 +1,42 @@ +import execnet import py import pytest +import shutil import textwrap -import execnet -from _pytest.pytester import HookRecorder -from xdist import workermanage, newhooks +from pathlib import Path +from xdist import workermanage from xdist.workermanage import HostRSync, NodeManager pytest_plugins = "pytester" @pytest.fixture -def hookrecorder(request, config): - hookrecorder = HookRecorder(config.pluginmanager) - if hasattr(hookrecorder, "start_recording"): - hookrecorder.start_recording(newhooks) - request.addfinalizer(hookrecorder.finish_recording) +def hookrecorder(request, config, pytester: pytest.Pytester): + hookrecorder = pytester.make_hook_recorder(config.pluginmanager) return hookrecorder @pytest.fixture -def config(testdir): - return testdir.parseconfig() +def config(pytester: pytest.Pytester): + return pytester.parseconfig() @pytest.fixture -def mysetup(tmpdir): - class mysetup: - source = tmpdir.mkdir("source") - dest = tmpdir.mkdir("dest") - - return mysetup() +def source(tmp_path: Path) -> Path: + source = tmp_path / "source" + source.mkdir() + return source @pytest.fixture -def workercontroller(monkeypatch): +def dest(tmp_path: Path) -> Path: + dest = tmp_path / "dest" + dest.mkdir() + return dest + + +@pytest.fixture +def workercontroller(monkeypatch: pytest.MonkeyPatch): class MockController: def __init__(self, *args): pass @@ -46,18 +49,20 @@ def workercontroller(monkeypatch): class TestNodeManagerPopen: - def test_popen_no_default_chdir(self, config): + def test_popen_no_default_chdir(self, config) -> None: gm = NodeManager(config, ["popen"]) assert gm.specs[0].chdir is None - def test_default_chdir(self, config): + def test_default_chdir(self, config) -> None: specs = ["ssh=noco", "socket=xyz"] for spec in NodeManager(config, specs).specs: assert spec.chdir == "pyexecnetcache" for spec in NodeManager(config, specs, defaultchdir="abc").specs: assert spec.chdir == "abc" - def test_popen_makegateway_events(self, config, hookrecorder, workercontroller): + def test_popen_makegateway_events( + self, config, hookrecorder, workercontroller + ) -> None: hm = NodeManager(config, ["popen"] * 2) hm.setup_nodes(None) call = hookrecorder.popcall("pytest_xdist_setupnodes") @@ -72,15 +77,16 @@ class TestNodeManagerPopen: hm.teardown_nodes() assert not len(hm.group) - def test_popens_rsync(self, config, mysetup, workercontroller): - source = mysetup.source + def test_popens_rsync( + self, config, source: Path, dest: Path, workercontroller + ) -> None: hm = NodeManager(config, ["popen"] * 2) hm.setup_nodes(None) assert len(hm.group) == 2 for gw in hm.group: class pseudoexec: - args = [] + args = [] # type: ignore[var-annotated] def __init__(self, *args): self.args.extend(args) @@ -97,30 +103,37 @@ class TestNodeManagerPopen: assert not len(hm.group) assert "sys.path.insert" in gw.remote_exec.args[0] - def test_rsync_popen_with_path(self, config, mysetup, workercontroller): - source, dest = mysetup.source, mysetup.dest + def test_rsync_popen_with_path( + self, config, source: Path, dest: Path, workercontroller + ) -> None: hm = NodeManager(config, ["popen//chdir=%s" % dest] * 1) hm.setup_nodes(None) - source.ensure("dir1", "dir2", "hello") + source.joinpath("dir1", "dir2").mkdir(parents=True) + source.joinpath("dir1", "dir2", "hello").touch() notifications = [] for gw in hm.group: hm.rsync(gw, source, notify=lambda *args: notifications.append(args)) assert len(notifications) == 1 assert notifications[0] == ("rsyncrootready", hm.group["gw0"].spec, source) hm.teardown_nodes() - dest = dest.join(source.basename) - assert dest.join("dir1").check() - assert dest.join("dir1", "dir2").check() - assert dest.join("dir1", "dir2", "hello").check() + dest = dest.joinpath(source.name) + assert dest.joinpath("dir1").exists() + assert dest.joinpath("dir1", "dir2").exists() + assert dest.joinpath("dir1", "dir2", "hello").exists() def test_rsync_same_popen_twice( - self, config, mysetup, hookrecorder, workercontroller - ): - source, dest = mysetup.source, mysetup.dest + self, + config, + source: Path, + dest: Path, + hookrecorder, + workercontroller, + ) -> None: hm = NodeManager(config, ["popen//chdir=%s" % dest] * 2) hm.roots = [] hm.setup_nodes(None) - source.ensure("dir1", "dir2", "hello") + source.joinpath("dir1", "dir2").mkdir(parents=True) + source.joinpath("dir1", "dir2", "hello").touch() gw = hm.group[0] hm.rsync(gw, source) call = hookrecorder.popcall("pytest_xdist_rsyncstart") @@ -131,83 +144,98 @@ class TestNodeManagerPopen: class TestHRSync: - def test_hrsync_filter(self, mysetup): - source, _ = mysetup.source, mysetup.dest # noqa - source.ensure("dir", "file.txt") - source.ensure(".svn", "entries") - source.ensure(".somedotfile", "moreentries") - source.ensure("somedir", "editfile~") + def test_hrsync_filter(self, source: Path, dest: Path) -> None: + source.joinpath("dir").mkdir() + source.joinpath("dir", "file.txt").touch() + source.joinpath(".svn").mkdir() + source.joinpath(".svn", "entries").touch() + source.joinpath(".somedotfile").mkdir() + source.joinpath(".somedotfile", "moreentries").touch() + source.joinpath("somedir").mkdir() + source.joinpath("somedir", "editfile~").touch() syncer = HostRSync(source, ignores=NodeManager.DEFAULT_IGNORES) - files = list(source.visit(rec=syncer.filter, fil=syncer.filter)) + files = list(py.path.local(source).visit(rec=syncer.filter, fil=syncer.filter)) assert len(files) == 3 basenames = [x.basename for x in files] assert "dir" in basenames assert "file.txt" in basenames assert "somedir" in basenames - def test_hrsync_one_host(self, mysetup): - source, dest = mysetup.source, mysetup.dest + def test_hrsync_one_host(self, source: Path, dest: Path) -> None: gw = execnet.makegateway("popen//chdir=%s" % dest) finished = [] rsync = HostRSync(source) rsync.add_target_host(gw, finished=lambda: finished.append(1)) - source.join("hello.py").write("world") + source.joinpath("hello.py").write_text("world") rsync.send() gw.exit() - assert dest.join(source.basename, "hello.py").check() + assert dest.joinpath(source.name, "hello.py").exists() assert len(finished) == 1 class TestNodeManager: - @py.test.mark.xfail(run=False) - def test_rsync_roots_no_roots(self, testdir, mysetup): - mysetup.source.ensure("dir1", "file1").write("hello") - config = testdir.parseconfig(mysetup.source) - nodemanager = NodeManager(config, ["popen//chdir=%s" % mysetup.dest]) + @pytest.mark.xfail(run=False) + def test_rsync_roots_no_roots( + self, pytester: pytest.Pytester, source: Path, dest: Path + ) -> None: + source.joinpath("dir1").mkdir() + source.joinpath("dir1", "file1").write_text("hello") + config = pytester.parseconfig(source) + nodemanager = NodeManager(config, ["popen//chdir=%s" % dest]) # assert nodemanager.config.topdir == source == config.topdir - nodemanager.makegateways() - nodemanager.rsync_roots() - (p,) = nodemanager.gwmanager.multi_exec( + nodemanager.makegateways() # type: ignore[attr-defined] + nodemanager.rsync_roots() # type: ignore[call-arg] + (p,) = nodemanager.gwmanager.multi_exec( # type: ignore[attr-defined] "import os ; channel.send(os.getcwd())" ).receive_each() - p = py.path.local(p) + p = Path(p) print("remote curdir", p) - assert p == mysetup.dest.join(config.topdir.basename) - assert p.join("dir1").check() - assert p.join("dir1", "file1").check() + assert p == dest.joinpath(config.rootpath.name) + assert p.joinpath("dir1").check() + assert p.joinpath("dir1", "file1").check() - def test_popen_rsync_subdir(self, testdir, mysetup, workercontroller): - source, dest = mysetup.source, mysetup.dest - dir1 = mysetup.source.mkdir("dir1") - dir2 = dir1.mkdir("dir2") - dir2.ensure("hello") + def test_popen_rsync_subdir( + self, pytester: pytest.Pytester, source: Path, dest: Path, workercontroller + ) -> None: + dir1 = source / "dir1" + dir1.mkdir() + dir2 = dir1 / "dir2" + dir2.mkdir() + dir2.joinpath("hello").touch() for rsyncroot in (dir1, source): - dest.remove() + shutil.rmtree(str(dest), ignore_errors=True) nodemanager = NodeManager( - testdir.parseconfig( + pytester.parseconfig( "--tx", "popen//chdir=%s" % dest, "--rsyncdir", rsyncroot, source ) ) nodemanager.setup_nodes(None) # calls .rsync_roots() if rsyncroot == source: - dest = dest.join("source") - assert dest.join("dir1").check() - assert dest.join("dir1", "dir2").check() - assert dest.join("dir1", "dir2", "hello").check() + dest = dest.joinpath("source") + assert dest.joinpath("dir1").exists() + assert dest.joinpath("dir1", "dir2").exists() + assert dest.joinpath("dir1", "dir2", "hello").exists() nodemanager.teardown_nodes() @pytest.mark.parametrize( "flag, expects_report", [("-q", False), ("", False), ("-v", True)] ) def test_rsync_report( - self, testdir, mysetup, workercontroller, capsys, flag, expects_report - ): - source, dest = mysetup.source, mysetup.dest - dir1 = mysetup.source.mkdir("dir1") - args = "--tx", "popen//chdir=%s" % dest, "--rsyncdir", dir1, source + self, + pytester: pytest.Pytester, + source: Path, + dest: Path, + workercontroller, + capsys: pytest.CaptureFixture[str], + flag: str, + expects_report: bool, + ) -> None: + dir1 = source / "dir1" + dir1.mkdir() + args = ["--tx", "popen//chdir=%s" % dest, "--rsyncdir", str(dir1), str(source)] if flag: - args += (flag,) - nodemanager = NodeManager(testdir.parseconfig(*args)) + args.append(flag) + nodemanager = NodeManager(pytester.parseconfig(*args)) nodemanager.setup_nodes(None) # calls .rsync_roots() out, _ = capsys.readouterr() if expects_report: @@ -215,77 +243,86 @@ class TestNodeManager: else: assert "<= pytest/__init__.py" not in out - def test_init_rsync_roots(self, testdir, mysetup, workercontroller): - source, dest = mysetup.source, mysetup.dest - dir2 = source.ensure("dir1", "dir2", dir=1) - source.ensure("dir1", "somefile", dir=1) - dir2.ensure("hello") - source.ensure("bogusdir", "file") - source.join("tox.ini").write( + def test_init_rsync_roots( + self, pytester: pytest.Pytester, source: Path, dest: Path, workercontroller + ) -> None: + dir2 = source.joinpath("dir1", "dir2") + dir2.mkdir(parents=True) + source.joinpath("dir1", "somefile").mkdir() + dir2.joinpath("hello").touch() + source.joinpath("bogusdir").mkdir() + source.joinpath("bogusdir", "file").touch() + source.joinpath("tox.ini").write_text( textwrap.dedent( """ - [pytest] - rsyncdirs=dir1/dir2 - """ + [pytest] + rsyncdirs=dir1/dir2 + """ ) ) - config = testdir.parseconfig(source) + config = pytester.parseconfig(source) nodemanager = NodeManager(config, ["popen//chdir=%s" % dest]) nodemanager.setup_nodes(None) # calls .rsync_roots() - assert dest.join("dir2").check() - assert not dest.join("dir1").check() - assert not dest.join("bogus").check() + assert dest.joinpath("dir2").exists() + assert not dest.joinpath("dir1").exists() + assert not dest.joinpath("bogus").exists() - def test_rsyncignore(self, testdir, mysetup, workercontroller): - source, dest = mysetup.source, mysetup.dest - dir2 = source.ensure("dir1", "dir2", dir=1) - source.ensure("dir5", "dir6", "bogus") - source.ensure("dir5", "file") - dir2.ensure("hello") - source.ensure("foo", "bar") - source.ensure("bar", "foo") - source.join("tox.ini").write( + def test_rsyncignore( + self, pytester: pytest.Pytester, source: Path, dest: Path, workercontroller + ) -> None: + dir2 = source.joinpath("dir1", "dir2") + dir2.mkdir(parents=True) + source.joinpath("dir5", "dir6").mkdir(parents=True) + source.joinpath("dir5", "dir6", "bogus").touch() + source.joinpath("dir5", "file").touch() + dir2.joinpath("hello").touch() + source.joinpath("foo").mkdir() + source.joinpath("foo", "bar").touch() + source.joinpath("bar").mkdir() + source.joinpath("bar", "foo").touch() + source.joinpath("tox.ini").write_text( textwrap.dedent( """ - [pytest] - rsyncdirs = dir1 dir5 - rsyncignore = dir1/dir2 dir5/dir6 foo* - """ + [pytest] + rsyncdirs = dir1 dir5 + rsyncignore = dir1/dir2 dir5/dir6 foo* + """ ) ) - config = testdir.parseconfig(source) + config = pytester.parseconfig(source) config.option.rsyncignore = ["bar"] nodemanager = NodeManager(config, ["popen//chdir=%s" % dest]) nodemanager.setup_nodes(None) # calls .rsync_roots() - assert dest.join("dir1").check() - assert not dest.join("dir1", "dir2").check() - assert dest.join("dir5", "file").check() - assert not dest.join("dir6").check() - assert not dest.join("foo").check() - assert not dest.join("bar").check() + assert dest.joinpath("dir1").exists() + assert not dest.joinpath("dir1", "dir2").exists() + assert dest.joinpath("dir5", "file").exists() + assert not dest.joinpath("dir6").exists() + assert not dest.joinpath("foo").exists() + assert not dest.joinpath("bar").exists() - def test_optimise_popen(self, testdir, mysetup, workercontroller): - source = mysetup.source + def test_optimise_popen( + self, pytester: pytest.Pytester, source: Path, dest: Path, workercontroller + ) -> None: specs = ["popen"] * 3 - source.join("conftest.py").write("rsyncdirs = ['a']") - source.ensure("a", dir=1) - config = testdir.parseconfig(source) + source.joinpath("conftest.py").write_text("rsyncdirs = ['a']") + source.joinpath("a").mkdir() + config = pytester.parseconfig(source) nodemanager = NodeManager(config, specs) nodemanager.setup_nodes(None) # calls .rysnc_roots() for gwspec in nodemanager.specs: assert gwspec._samefilesystem() assert not gwspec.chdir - def test_ssh_setup_nodes(self, specssh, testdir): - testdir.makepyfile( + def test_ssh_setup_nodes(self, specssh: str, pytester: pytest.Pytester) -> None: + pytester.makepyfile( __init__="", test_x=""" def test_one(): pass """, ) - reprec = testdir.inline_run( - "-d", "--rsyncdir=%s" % testdir.tmpdir, "--tx", specssh, testdir.tmpdir + reprec = pytester.inline_run( + "-d", "--rsyncdir=%s" % pytester.path, "--tx", specssh, pytester.path ) (rep,) = reprec.getreports("pytest_runtest_logreport") assert rep.passed diff --git a/tox.ini b/tox.ini index 9d63503..3ab7109 100644 --- a/tox.ini +++ b/tox.ini @@ -18,7 +18,6 @@ commands= extras = testing psutil -deps = pytest commands = pytest {posargs:-k psutil}