Merge pull request #824 from nicoddemus/drop-py-path

Replace py.path.local usages by pathlib.Path

.pre-commit-config.yaml

@@ -1,4 +1,9 @@
 repos:
+-   repo: https://github.com/PyCQA/autoflake
+    rev: v1.7.6
+    hooks:
+    -   id: autoflake
+        args: ["--in-place", "--remove-unused-variables", "--remove-all-unused-imports"]
 -   repo: https://github.com/psf/black
     rev: 22.3.0
     hooks:

changelog/824.trivial.rst

@@ -0,0 +1 @@
+Replace internal usages of ``py.path.local`` by ``pathlib.Path``.

docs/known-limitations.rst

@@ -17,7 +17,8 @@ Example:
 
     import pytest
 
-    @pytest.mark.parametrize("param", {"a","b"})
+
+    @pytest.mark.parametrize("param", {"a", "b"})
     def test_pytest_parametrize_unordered(param):
         pass
 
@@ -37,6 +38,7 @@ Some solutions:
 
     import pytest
 
+
     @pytest.mark.parametrize("param", ["a", "b"])
     def test_pytest_parametrize_unordered(param):
         pass
@@ -47,6 +49,7 @@ Some solutions:
 
     import pytest
 
+
     @pytest.mark.parametrize("param", sorted({"a", "b"}))
     def test_pytest_parametrize_unordered(param):
         pass

src/xdist/_path.py

@@ -0,0 +1,19 @@
+import os
+from itertools import chain
+from pathlib import Path
+from typing import Callable, Iterator
+
+
+def visit_path(
+    path: Path, *, filter: Callable[[Path], bool], recurse: Callable[[Path], bool]
+) -> Iterator[Path]:
+    """
+    Implements the interface of ``py.path.local.visit()`` for Path objects,
+    to simplify porting the code over from ``py.path.local``.
+    """
+    for dirpath, dirnames, filenames in os.walk(path):
+        dirnames[:] = [x for x in dirnames if recurse(Path(dirpath, x))]
+        for name in chain(dirnames, filenames):
+            p = Path(dirpath, name)
+            if filter(p):
+                yield p

src/xdist/looponfail.py

@@ -6,11 +6,17 @@
     processes) otherwise changes to source code can crash
     the controlling process which should best never happen.
 """
-import py
+import os
+from pathlib import Path
+from typing import Dict, Sequence
+
 import pytest
 import sys
 import time
 import execnet
+from _pytest._io import TerminalWriter
+
+from xdist._path import visit_path
 
 
 @pytest.hookimpl
@@ -38,9 +44,9 @@ def pytest_cmdline_main(config):
         return 2  # looponfail only can get stop with ctrl-C anyway
 
 
-def looponfail_main(config):
+def looponfail_main(config: pytest.Config) -> None:
     remotecontrol = RemoteControl(config)
-    rootdirs = [py.path.local(root) for root in config.getini("looponfailroots")]
+    rootdirs = [Path(root) for root in config.getini("looponfailroots")]
     statrecorder = StatRecorder(rootdirs)
     try:
         while 1:
@@ -71,7 +77,7 @@ class RemoteControl:
 
     def setup(self, out=None):
         if out is None:
-            out = py.io.TerminalWriter()
+            out = TerminalWriter()
         if hasattr(self, "gateway"):
             raise ValueError("already have gateway %r" % self.gateway)
         self.trace("setting up worker session")
@@ -129,7 +135,7 @@ class RemoteControl:
 
 
 def repr_pytest_looponfailinfo(failreports, rootdirs):
-    tr = py.io.TerminalWriter()
+    tr = TerminalWriter()
     if failreports:
         tr.sep("#", "LOOPONFAILING", bold=True)
         for report in failreports:
@@ -225,16 +231,16 @@ class WorkerFailSession:
 
 
 class StatRecorder:
-    def __init__(self, rootdirlist):
+    def __init__(self, rootdirlist: Sequence[Path]) -> None:
         self.rootdirlist = rootdirlist
-        self.statcache = {}
+        self.statcache: Dict[Path, os.stat_result] = {}
         self.check()  # snapshot state
 
-    def fil(self, p):
+    def fil(self, p: Path) -> bool:
-        return p.check(file=1, dotfile=0) and p.ext != ".pyc"
+        return p.is_file() and not p.name.startswith(".") and p.suffix != ".pyc"
 
-    def rec(self, p):
+    def rec(self, p: Path) -> bool:
-        return p.check(dotfile=0)
+        return not p.name.startswith(".") and p.exists()
 
     def waitonchange(self, checkinterval=1.0):
         while 1:
@@ -243,34 +249,34 @@ class StatRecorder:
                 return
             time.sleep(checkinterval)
 
-    def check(self, removepycfiles=True):  # noqa, too complex
+    def check(self, removepycfiles: bool = True) -> bool:  # noqa, too complex
         changed = False
-        statcache = self.statcache
+        newstat: Dict[Path, os.stat_result] = {}
-        newstat = {}
         for rootdir in self.rootdirlist:
-            for path in rootdir.visit(self.fil, self.rec):
+            for path in visit_path(rootdir, filter=self.fil, recurse=self.rec):
-                oldstat = statcache.pop(path, None)
+                oldstat = self.statcache.pop(path, None)
                 try:
-                    newstat[path] = curstat = path.stat()
+                    curstat = path.stat()
-                except py.error.ENOENT:
+                except OSError:
                     if oldstat:
                         changed = True
                 else:
-                    if oldstat:
+                    newstat[path] = curstat
+                    if oldstat is not None:
                         if (
-                            oldstat.mtime != curstat.mtime
+                            oldstat.st_mtime != curstat.st_mtime
-                            or oldstat.size != curstat.size
+                            or oldstat.st_size != curstat.st_size
                         ):
                             changed = True
                             print("# MODIFIED", path)
-                            if removepycfiles and path.ext == ".py":
+                            if removepycfiles and path.suffix == ".py":
-                                pycfile = path + "c"
+                                pycfile = path.with_suffix(".pyc")
-                                if pycfile.check():
+                                if pycfile.is_file():
-                                    pycfile.remove()
+                                    os.unlink(pycfile)
 
                     else:
                         changed = True
-        if statcache:
+        if self.statcache:
             changed = True
         self.statcache = newstat
         return changed

src/xdist/plugin.py

@@ -3,7 +3,6 @@ import uuid
 import sys
 from pathlib import Path
 
-import py
 import pytest
 
 
@@ -165,7 +164,7 @@ def pytest_addoption(parser):
         "looponfailroots",
         type="paths" if PYTEST_GTE_7 else "pathlist",
         help="directories to check for changes",
-        default=[Path.cwd() if PYTEST_GTE_7 else py.path.local()],
+        default=[Path.cwd()],
     )
 
 

src/xdist/workermanage.py

@@ -3,6 +3,8 @@ import os
 import re
 import sys
 import uuid
+from pathlib import Path
+from typing import List, Union, Sequence, Optional, Any, Tuple, Set
 
 import py
 import pytest
@@ -33,7 +35,7 @@ class NodeManager:
     EXIT_TIMEOUT = 10
     DEFAULT_IGNORES = [".*", "*.pyc", "*.pyo", "*~"]
 
-    def __init__(self, config, specs=None, defaultchdir="pyexecnetcache"):
+    def __init__(self, config, specs=None, defaultchdir="pyexecnetcache") -> None:
         self.config = config
         self.trace = self.config.trace.get("nodemanager")
         self.testrunuid = self.config.getoption("testrunuid")
@@ -52,7 +54,7 @@ class NodeManager:
             self.specs.append(spec)
         self.roots = self._getrsyncdirs()
         self.rsyncoptions = self._getrsyncoptions()
-        self._rsynced_specs = set()
+        self._rsynced_specs: Set[Tuple[Any, Any]] = set()
 
     def rsync_roots(self, gateway):
         """Rsync the set of roots to the node's gateway cwd."""
@@ -81,7 +83,7 @@ class NodeManager:
     def _getxspecs(self):
         return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
 
-    def _getrsyncdirs(self):
+    def _getrsyncdirs(self) -> List[Path]:
         for spec in self.specs:
             if not spec.popen or spec.chdir:
                 break
@@ -108,8 +110,8 @@ class NodeManager:
             candidates.extend(rsyncroots)
         roots = []
         for root in candidates:
-            root = py.path.local(root).realpath()
+            root = Path(root).resolve()
-            if not root.check():
+            if not root.exists():
                 raise pytest.UsageError("rsyncdir doesn't exist: {!r}".format(root))
             if root not in roots:
                 roots.append(root)
@@ -160,18 +162,24 @@ class NodeManager:
 class HostRSync(execnet.RSync):
     """RSyncer that filters out common files"""
 
-    def __init__(self, sourcedir, *args, **kwargs):
+    PathLike = Union[str, "os.PathLike[str]"]
-        self._synced = {}
-        ignores = kwargs.pop("ignores", None) or []
-        self._ignores = [
-            re.compile(fnmatch.translate(getattr(x, "strpath", x))) for x in ignores
-        ]
-        super().__init__(sourcedir=sourcedir, **kwargs)
 
-    def filter(self, path):
+    def __init__(
-        path = py.path.local(path)
+        self,
+        sourcedir: PathLike,
+        *,
+        ignores: Optional[Sequence[PathLike]] = None,
+        **kwargs: object
+    ) -> None:
+        if ignores is None:
+            ignores = []
+        self._ignores = [re.compile(fnmatch.translate(os.fspath(x))) for x in ignores]
+        super().__init__(sourcedir=Path(sourcedir), **kwargs)
+
+    def filter(self, path: PathLike) -> bool:
+        path = Path(path)
         for cre in self._ignores:
-            if cre.match(path.basename) or cre.match(path.strpath):
+            if cre.match(path.name) or cre.match(str(path)):
                 return False
         else:
             return True
@@ -187,20 +195,28 @@ class HostRSync(execnet.RSync):
             print("{}:{} <= {}".format(gateway.spec, remotepath, path))
 
 
-def make_reltoroot(roots, args):
+def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
     # XXX introduce/use public API for splitting pytest args
     splitcode = "::"
     result = []
     for arg in args:
         parts = arg.split(splitcode)
-        fspath = py.path.local(parts[0])
+        fspath = Path(parts[0])
-        if not fspath.exists():
+        try:
+            exists = fspath.exists()
+        except OSError:
+            exists = False
+        if not exists:
             result.append(arg)
             continue
         for root in roots:
-            x = fspath.relto(root)
+            x: Optional[Path]
+            try:
+                x = fspath.relative_to(root)
+            except ValueError:
+                x = None
             if x or fspath == root:
-                parts[0] = root.basename + "/" + x
+                parts[0] = root.name + "/" + str(x)
                 break
         else:
             raise ValueError("arg {} not relative to an rsync root".format(arg))

testing/test_looponfail.py

@@ -1,4 +1,6 @@
-import py
+import unittest.mock
+from typing import List
+
 import pytest
 import shutil
 import textwrap
@@ -16,7 +18,7 @@ class TestStatRecorder:
         tmp = tmp_path
         hello = tmp / "hello.py"
         hello.touch()
-        sd = StatRecorder([py.path.local(tmp)])
+        sd = StatRecorder([tmp])
         changed = sd.check()
         assert not changed
 
@@ -56,15 +58,12 @@ class TestStatRecorder:
         tmp = tmp_path
         tmp.joinpath("dir").mkdir()
         tmp.joinpath("dir", "hello.py").touch()
-        sd = StatRecorder([py.path.local(tmp)])
+        sd = StatRecorder([tmp])
-        assert not sd.fil(py.path.local(tmp / "dir"))
+        assert not sd.fil(tmp / "dir")
 
-    def test_filechange_deletion_race(
+    def test_filechange_deletion_race(self, tmp_path: Path) -> None:
-        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-    ) -> None:
         tmp = tmp_path
-        pytmp = py.path.local(tmp)
+        sd = StatRecorder([tmp])
-        sd = StatRecorder([pytmp])
         changed = sd.check()
         assert not changed
 
@@ -76,8 +75,12 @@ class TestStatRecorder:
         p.unlink()
         # make check()'s visit() call return our just removed
         # path as if we were in a race condition
-        monkeypatch.setattr(pytmp, "visit", lambda *args: [py.path.local(p)])
+        dirname = str(tmp)
-
+        dirnames: List[str] = []
+        filenames = [str(p)]
+        with unittest.mock.patch(
+            "os.walk", return_value=[(dirname, dirnames, filenames)], autospec=True
+        ):
             changed = sd.check()
         assert changed
 
@@ -85,7 +88,7 @@ class TestStatRecorder:
         tmp = tmp_path
         hello = tmp / "hello.py"
         hello.touch()
-        sd = StatRecorder([py.path.local(tmp)])
+        sd = StatRecorder([tmp])
         changed = sd.check()
         assert not changed
 
@@ -100,7 +103,7 @@ class TestStatRecorder:
         self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         tmp = tmp_path
-        sd = StatRecorder([py.path.local(tmp)])
+        sd = StatRecorder([tmp])
 
         ret_values = [True, False]
         monkeypatch.setattr(StatRecorder, "check", lambda self: ret_values.pop())

testing/test_remote.py

@@ -1,5 +1,4 @@
 import pprint
-import py
 import pytest
 import sys
 import uuid
@@ -108,7 +107,7 @@ class TestWorkerInteractor:
         assert ev.name == "collectionstart"
         assert not ev.kwargs
         ev = worker.popevent("collectionfinish")
-        assert ev.kwargs["topdir"] == py.path.local(worker.pytester.path)
+        assert ev.kwargs["topdir"] == str(worker.pytester.path)
         ids = ev.kwargs["ids"]
         assert len(ids) == 1
         worker.sendcommand("runtests", indices=list(range(len(ids))))

testing/test_workermanage.py

@@ -1,5 +1,4 @@
 import execnet
-import py
 import pytest
 import shutil
 import textwrap
@@ -7,6 +6,7 @@ import warnings
 from pathlib import Path
 from util import generate_warning
 from xdist import workermanage
+from xdist._path import visit_path
 from xdist.remote import serialize_warning_message
 from xdist.workermanage import HostRSync, NodeManager, unserialize_warning_message
 
@@ -157,12 +157,9 @@ class TestHRSync:
         source.joinpath("somedir").mkdir()
         source.joinpath("somedir", "editfile~").touch()
         syncer = HostRSync(source, ignores=NodeManager.DEFAULT_IGNORES)
-        files = list(py.path.local(source).visit(rec=syncer.filter, fil=syncer.filter))
+        files = list(visit_path(source, recurse=syncer.filter, filter=syncer.filter))
-        assert len(files) == 3
+        names = {x.name for x in files}
-        basenames = [x.basename for x in files]
+        assert names == {"dir", "file.txt", "somedir"}
-        assert "dir" in basenames
-        assert "file.txt" in basenames
-        assert "somedir" in basenames
 
     def test_hrsync_one_host(self, source: Path, dest: Path) -> None:
         gw = execnet.makegateway("popen//chdir=%s" % dest)