Anonymous avatar Anonymous committed 7a268f2

Port test to pypy's _posixsubprocess test for py3k

Comments (0)

Files changed (2)

pypy/module/_posixsubprocess/test/fd_status.py

+"""When called as a script, print a comma-separated list of the open
+file descriptors on stdout."""
+
+import errno
+import os
+
+try:
+    _MAXFD = os.sysconf("SC_OPEN_MAX")
+except:
+    _MAXFD = 256
+
+if __name__ == "__main__":
+    fds = []
+    for fd in range(0, _MAXFD):
+        try:
+            st = os.fstat(fd)
+        except OSError as e:
+            if e.errno == errno.EBADF:
+                continue
+            raise
+        # Ignore Solaris door files
+        if st.st_mode & 0xF000 != 0xd000:
+            fds.append(fd)
+    print(','.join(map(str, fds)))

pypy/module/_posixsubprocess/test/test_subprocess.py

+import sys
+import os
+from os.path import dirname
+from pypy.tool.pytest.objspace import gettestobjspace
+
+def setup_module(mod):
+    usemodules = ['_posixsubprocess', 'signal']
+    # py3k os.open uses subprocess, requiring the following per platform
+    if os.name != 'nt':
+        usemodules += ['fcntl', 'select']
+    else:
+        usemodules += ['_rawffi', 'thread']
+    mod.space = gettestobjspace(usemodules=usemodules)
+
 class AppTestSubprocess:
-    spaceconfig = dict(usemodules=('_posixsubprocess',))
+    # XXX write more tests
 
-    # XXX write more tests
+    def setup_class(cls):
+        cls.space = space
+        cls.w_dir = space.wrap(dirname(__file__))
 
     def test_cloexec_pipe(self):
         import _posixsubprocess, os
         assert 0 <= fd2 < 4096
         os.close(fd1)
         os.close(fd2)
+
+    def test_close_fds_true(self):
+        import subprocess
+        import os.path
+        import os
+
+        fds = os.pipe()
+        #self.addCleanup(os.close, fds[0])
+        #self.addCleanup(os.close, fds[1])
+
+        open_fds = set(fds)
+        # add a bunch more fds
+        for _ in range(9):
+            fd = os.open("/dev/null", os.O_RDONLY)
+            #self.addCleanup(os.close, fd)
+            open_fds.add(fd)
+
+        p = subprocess.Popen(['/usr/bin/env', 'python', os.path.join(self.dir, 'fd_status.py')], stdout=subprocess.PIPE, close_fds=True)
+        output, ignored = p.communicate()
+        remaining_fds = set(map(int, output.split(b',')))
+
+        assert not (remaining_fds & open_fds), "Some fds were left open"
+        assert 1 in remaining_fds, "Subprocess failed"
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.