Merge pull request #184 from bluetech/testing-rm-py

testing: some cleanups

Author: bluetech

CHANGELOG.rst

@@ -6,6 +6,7 @@
   * minimal mypy fixes and python2 support code drop
 * migrate packaging to hatch
 * drop deprecated apis of old makegateway names
+* Removed ``py`` testing dependency
 
 
 

doc/example/conftest.py

@@ -1,13 +1,13 @@
+import pathlib
 import sys
 
-import py
 
-# make execnet and example code importable
-cand = py.path.local(__file__).dirpath().dirpath().dirpath()
-if cand.join("execnet", "__init__.py").check():
+# Make execnet and example code importable.
+cand = pathlib.Path(__file__).parent.parent.parent
+if cand.joinpath("execnet", "__init__.py").exists():
     if str(cand) not in sys.path:
         sys.path.insert(0, str(cand))
-cand = py.path.local(__file__).dirpath()
+cand = pathlib.Path(__file__).parent
 if str(cand) not in sys.path:
     sys.path.insert(0, str(cand))
 

doc/example/svn-sync-repo.py

@@ -6,10 +6,11 @@
 
 """
 import os
+import pathlib
+import subprocess
 import sys
 
 import execnet
-import py
 
 
 def usage():
@@ -19,8 +20,8 @@ def usage():
 
 def main(args):
     remote = args[0]
-    localrepo = py.path.local(args[1])
-    if not localrepo.check(dir=1):
+    localrepo = pathlib.Path(args[1])
+    if not localrepo.is_dir():
         raise SystemExit(f"localrepo {localrepo} does not exist")
     if len(args) == 3:
         configfile = args[2]
@@ -39,12 +40,18 @@ def main(args):
     # 4. client goes back to step 1
     c = gw.remote_exec(
         """
-        import py
         import os
+        import subprocess
         import time
+
         remote_rev, repopath = channel.receive()
-        while 1:
-            rev = py.process.cmdexec('svnlook youngest "%s"' % repopath)
+        while True:
+            rev = subprocess.run(
+                ["svnlook", "youngest", repopath],
+                check=True,
+                capture_output=True,
+                text=True,
+            ).stdout
             rev = int(rev)
             if rev > remote_rev:
                 revrange = (remote_rev+1, rev)
@@ -103,12 +110,17 @@ def svn_load(repo, dumpchannel, maxcount=100):
         if count <= 0:
             dumpchannel.send(maxcount)
             count = maxcount
-    print >> sys.stdout
+    print()
     f.close()
 
 
 def get_svn_youngest(repo):
-    rev = py.process.cmdexec('svnlook youngest "%s"' % repo)
+    rev = subprocess.run(
+        ["svnlook", "youngest", repo],
+        check=True,
+        capture_output=True,
+        text=True,
+    ).stdout
     return int(rev)
 
 

doc/example/sysinfo.py

@@ -6,11 +6,11 @@
 (c) Holger Krekel, MIT license
 """
 import optparse
+import pathlib
 import re
 import sys
 
 import execnet
-import py
 
 
 parser = optparse.OptionParser(usage=__doc__)
@@ -34,14 +34,15 @@
 
 
 def parsehosts(path):
-    path = py.path.local(path)
+    path = pathlib.Path(path)
     l = []
     rex = re.compile(r"Host\s*(\S+)")
-    for line in path.readlines():
-        m = rex.match(line)
-        if m is not None:
-            (sshname,) = m.groups()
-            l.append(sshname)
+    with path.open() as f:
+        for line in f:
+            m = rex.match(line)
+            if m is not None:
+                (sshname,) = m.groups()
+                l.append(sshname)
     return l
 
 

execnet/gateway.py

@@ -169,19 +169,15 @@ def rinfo_source(channel):
 
 def _find_non_builtin_globals(source, codeobj):
     import ast
-
-    try:
-        import __builtin__
-    except ImportError:
-        import builtins as __builtin__
+    import builtins
 
     vars = dict.fromkeys(codeobj.co_varnames)
     return [
         node.id
         for node in ast.walk(ast.parse(source))
         if isinstance(node, ast.Name)
         and node.id not in vars
-        and node.id not in __builtin__.__dict__
+        and node.id not in builtins.__dict__
     ]
 
 

execnet/rsync.py

@@ -173,10 +173,26 @@ def _send_directory(self, path):
             self._send_directory_structure(p)
 
     def _send_link_structure(self, path):
-        linkpoint = os.readlink(path)
+        sourcedir = self._sourcedir
         basename = path[len(self._sourcedir) + 1 :]
-        if linkpoint.startswith(self._sourcedir):
-            self._send_link("linkbase", basename, linkpoint[len(self._sourcedir) + 1 :])
+        linkpoint = os.readlink(path)
+        # On Windows, readlink returns an extended path (//?/) for
+        # absolute links, but relpath doesn't like mixing extended
+        # and non-extended paths. So fix it up ourselves.
+        if (
+            os.path.__name__ == "ntpath"
+            and linkpoint.startswith("\\\\?\\")
+            and not self._sourcedir.startswith("\\\\?\\")
+        ):
+            sourcedir = "\\\\?\\" + self._sourcedir
+        try:
+            relpath = os.path.relpath(linkpoint, sourcedir)
+        except ValueError:
+            relpath = None
+        if relpath not in (None, os.curdir, os.pardir) and not relpath.startswith(
+            os.pardir + os.sep
+        ):
+            self._send_link("linkbase", basename, relpath)
         else:
             # relative or absolute link, just send it
             self._send_link("link", basename, linkpoint)

testing/conftest.py

@@ -1,8 +1,9 @@
+import pathlib
+import shutil
 import subprocess
 import sys
 
 import execnet
-import py
 import pytest
 from execnet.gateway_base import get_execmodel
 from execnet.gateway_base import WorkerPool
@@ -76,7 +77,7 @@ def getspecssh(config):
     xspecs = getgspecs(config)
     for spec in xspecs:
         if spec.ssh:
-            if not py.path.local.sysfind("ssh"):
+            if not shutil.which("ssh"):
                 pytest.skip("command not found: ssh")
             return spec
     pytest.skip("need '--gx ssh=...'")
@@ -113,8 +114,9 @@ def getexecutable(name, cache={}):
         return cache[name]
     except KeyError:
         if name == "sys.executable":
-            return py.path.local(sys.executable)
-        executable = py.path.local.sysfind(name)
+            return pathlib.Path(sys.executable)
+        path = shutil.which(name)
+        executable = pathlib.Path(path) if path is not None else None
         if executable:
             if name == "jython":
                 popen = subprocess.Popen(