IQ.Lvbs/pytest-xdist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace py.path.local usages by pathlib.Path

Browse Source
This commit is contained in:
Bruno Oliveira
2022-10-21 22:09:38 -03:00
parent f04cd22fd6
commit 25cc1a4119
10 changed files with 121 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
src/xdist/_path.py Normal file
View File

@@ -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

58
src/xdist/looponfail.py
View File

@@ -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):
return p.check(file=1, dotfile=0) and p.ext != ".pyc"
def fil(self, p: Path) -> bool:
return p.is_file() and not p.name.startswith(".") and p.suffix != ".pyc"
def rec(self, p):
return p.check(dotfile=0)
def rec(self, p: Path) -> bool:
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 = {}
newstat: Dict[Path, os.stat_result] = {}
for rootdir in self.rootdirlist:
for path in rootdir.visit(self.fil, self.rec):
oldstat = statcache.pop(path, None)
for path in visit_path(rootdir, filter=self.fil, recurse=self.rec):
oldstat = self.statcache.pop(path, None)
try:
newstat[path] = curstat = path.stat()
except py.error.ENOENT:
curstat = path.stat()
except OSError:
if oldstat:
changed = True
else:
if oldstat:
newstat[path] = curstat
if oldstat is not None:
if (
oldstat.mtime != curstat.mtime
or oldstat.size != curstat.size
oldstat.st_mtime != curstat.st_mtime
or oldstat.st_size != curstat.st_size
):
changed = True
print("# MODIFIED", path)
if removepycfiles and path.ext == ".py":
pycfile = path + "c"
if pycfile.check():
pycfile.remove()
if removepycfiles and path.suffix == ".py":
pycfile = path.with_suffix(".pyc")
if pycfile.is_file():
os.unlink(pycfile)
else:
changed = True
if statcache:
if self.statcache:
changed = True
self.statcache = newstat
return changed

3
src/xdist/plugin.py
View File

@@ -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()],
)

56
src/xdist/workermanage.py
View File

@@ -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()
if not root.check():
root = Path(root).resolve()
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):
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)
PathLike = Union[str, "os.PathLike[str]"]
def filter(self, path):
path = py.path.local(path)
def __init__(
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])
if not fspath.exists():
fspath = Path(parts[0])
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))