Update test suite to modern pytest

- Use pytest>=6.2 features (same as install_requires).
  When we can require >=7, can fix some more typing omissions and
  version checks.
- Replace testdir with pytester
- Replace py.test with pytest
- Replace tmpdir with tmp_path
- Remove (almost) all other uses of py
- Add some type annotations (not checked yet)

Ref #722.

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