doc,testing: drop use of py & tmpdir

Use modern alternatives instead.

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
 
 

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(

testing/test_basics.py

@@ -2,10 +2,10 @@
 import os
 import subprocess
 import sys
+import textwrap
 from io import BytesIO
 
 import execnet
-import py
 import pytest
 from execnet import gateway
 from execnet import gateway_base
@@ -29,20 +29,20 @@ def test_serializer_api(self, val):
         val2 = execnet.loads(dumped)
         assert val == val2
 
-    def test_mmap(self, tmpdir, val):
+    def test_mmap(self, tmp_path, val):
         mmap = pytest.importorskip("mmap").mmap
-        p = tmpdir.join("data")
+        p = tmp_path / "data"
         with p.open("wb") as f:
             f.write(execnet.dumps(val))
-        f = p.open("r+b")
-        m = mmap(f.fileno(), 0)
-        val2 = execnet.load(m)
+        with p.open("r+b") as f:
+            m = mmap(f.fileno(), 0)
+            val2 = execnet.load(m)
         assert val == val2
 
     def test_bytesio(self, val):
-        f = py.io.BytesIO()
+        f = BytesIO()
         execnet.dump(f, val)
-        read = py.io.BytesIO(f.getvalue())
+        read = BytesIO(f.getvalue())
         val2 = execnet.load(read)
         assert val == val2
 
@@ -81,10 +81,8 @@ def receive():
         return popen.stdout.readline()
 
     try:
-        source = py.code.Source(read_write_loop, "read_write_loop()")
-        repr_source = repr(str(source)) + "\n"
-        sendline = repr_source
-        send(sendline)
+        source = inspect.getsource(read_write_loop) + "read_write_loop()"
+        send(repr(source) + "\n")
         s = receive()
         assert s == "ok\n"
         send("hello\n")
@@ -114,11 +112,11 @@ def read_write_loop():
             break
 
 
-def test_io_message(anypython, tmpdir, execmodel):
-    check = tmpdir.join("check.py")
-    check.write(
-        py.code.Source(
-            gateway_base,
+def test_io_message(anypython, tmp_path, execmodel):
+    check = tmp_path / "check.py"
+    check.write_text(
+        inspect.getsource(gateway_base)
+        + textwrap.dedent(
             """
         from io import BytesIO
         import tempfile
@@ -146,24 +144,25 @@ def test_io_message(anypython, tmpdir, execmodel):
             ),
         )
     )
-    # out = py.process.cmdexec("%s %s" %(executable,check))
-    out = anypython.sysexec(check)
+    out = subprocess.run(
+        [str(anypython), str(check)], text=True, capture_output=True, check=True
+    ).stdout
     print(out)
     assert "all passed" in out
 
 
-def test_popen_io(anypython, tmpdir, execmodel):
-    check = tmpdir.join("check.py")
-    check.write(
-        py.code.Source(
-            gateway_base,
+def test_popen_io(anypython, tmp_path, execmodel):
+    check = tmp_path / "check.py"
+    check.write_text(
+        inspect.getsource(gateway_base)
+        + textwrap.dedent(
             f"""
         io = init_popen_io(get_execmodel({execmodel.backend!r}))
         io.write("hello".encode('ascii'))
         s = io.read(1)
         assert s == "x".encode('ascii')
-    """,
-        )
+    """
+        ),
     )
     from subprocess import Popen, PIPE
 
@@ -191,32 +190,36 @@ def newread(numbytes):
     assert result == b"tes"
 
 
-def test_rinfo_source(anypython, tmpdir):
-    check = tmpdir.join("check.py")
-    check.write(
-        py.code.Source(
+def test_rinfo_source(anypython, tmp_path):
+    check = tmp_path / "check.py"
+    check.write_text(
+        textwrap.dedent(
             """
         class Channel:
             def send(self, data):
                 assert eval(repr(data), {}) == data
         channel = Channel()
-        """,
-            gateway.rinfo_source,
+        """
+        )
+        + inspect.getsource(gateway.rinfo_source)
+        + textwrap.dedent(
             """
         print ('all passed')
-    """,
+        """
         )
     )
-    out = anypython.sysexec(check)
+    out = subprocess.run(
+        [str(anypython), str(check)], text=True, capture_output=True, check=True
+    ).stdout
     print(out)
     assert "all passed" in out
 
 
-def test_geterrortext(anypython, tmpdir):
-    check = tmpdir.join("check.py")
-    check.write(
-        py.code.Source(
-            gateway_base,
+def test_geterrortext(anypython, tmp_path):
+    check = tmp_path / "check.py"
+    check.write_text(
+        inspect.getsource(gateway_base)
+        + textwrap.dedent(
             """
         class Arg:
             pass
@@ -230,21 +233,22 @@ class Arg:
             s = geterrortext(excinfo)
             assert "17" in s
             print ("all passed")
-    """,
+    """
         )
     )
-    out = anypython.sysexec(check)
+    out = subprocess.run(
+        [str(anypython), str(check)], text=True, capture_output=True, check=True
+    ).stdout
     print(out)
     assert "all passed" in out
 
 
-@pytest.mark.skipif("not hasattr(os, 'dup')")
-def test_stdouterrin_setnull(execmodel):
-    cap = py.io.StdCaptureFD()
+@pytest.mark.skipif(not hasattr(os, "dup"), reason="no os.dup")
+def test_stdouterrin_setnull(execmodel, capfd):
     gateway_base.init_popen_io(execmodel)
     os.write(1, b"hello")
     os.read(0, 1)
-    out, err = cap.reset()
+    out, err = capfd.readouterr()
     assert not out
     assert not err
 
@@ -267,7 +271,7 @@ def close(self, errortext=None):
 
 
 def test_exectask(execmodel):
-    io = py.io.BytesIO()
+    io = BytesIO()
     io.execmodel = execmodel
     gw = gateway_base.WorkerGateway(io, id="something")
     ch = PseudoChannel()
@@ -278,10 +282,10 @@ def test_exectask(execmodel):
 class TestMessage:
     def test_wire_protocol(self):
         for i, handler in enumerate(Message._types):
-            one = py.io.BytesIO()
+            one = BytesIO()
             data = b"23"
             Message(i, 42, data).to_io(one)
-            two = py.io.BytesIO(one.getvalue())
+            two = BytesIO(one.getvalue())
             msg = Message.from_io(two)
             assert msg.msgcode == i
             assert isinstance(msg, Message)
@@ -338,7 +342,7 @@ def prototype(wrong):
     def test_function_without_known_source_fails(self):
         # this one won't be able to find the source
         mess = {}
-        py.builtin.exec_("def fail(channel): pass", mess, mess)
+        exec("def fail(channel): pass", mess, mess)
         print(inspect.getsourcefile(mess["fail"]))
         pytest.raises(ValueError, gateway._source_of_function, mess["fail"])
 
@@ -361,9 +365,9 @@ def working(channel):
 
 class TestGlobalFinder:
     def check(self, func):
-        src = py.code.Source(func)
-        code = py.code.Code(func)
-        return gateway._find_non_builtin_globals(str(src), code.raw)
+        src = textwrap.dedent(inspect.getsource(func))
+        code = func.__code__
+        return gateway._find_non_builtin_globals(src, code)
 
     def test_local(self):
         def f(a, b, c):