fixes, tests and basic Popen support

src/netns/subprocess.py

@@ -22,9 +22,8 @@ def spawn(executable, argv = None, cwd = None, env = None,
 
     The standard input, output, and error of the created process will be
     redirected to the file descriptors specified by `stdin`, `stdout`, and
-    `stderr`, respectively. These parameters must be integers or None, in which
-    case, no redirection will occur. If the value is negative, the respective
-    file descriptor is closed in the executed program.
+    `stderr`, respectively. These parameters must be open file objects,
+    integers or None, in which case, no redirection will occur.
 
     Note that the original descriptors are not closed, and that piping should
     be handled externally.
@@ -34,9 +33,12 @@ def spawn(executable, argv = None, cwd = None, env = None,
 
     userfd = [stdin, stdout, stderr]
     filtered_userfd = filter(lambda x: x != None and x >= 0, userfd)
-    sysfd = [x.fileno() for x in sys.stdin, sys.stdout, sys.stderr]
+    for i in range(3):
+        if userfd[i] != None and not isinstance(userfd[i], int):
+            userfd[i] = userfd[i].fileno()
+
     # Verify there is no clash
-    assert not (set(sysfd) & set(filtered_userfd))
+    assert not (set([0, 1, 2]) & set(filtered_userfd))
 
     if user != None:
         if str(user).isdigit():
@@ -61,13 +63,13 @@ def spawn(executable, argv = None, cwd = None, env = None,
             # Set up stdio piping
             for i in range(3):
                 if userfd[i] != None and userfd[i] >= 0:
-                    os.dup2(userfd[i], sysfd[i])
-                    os.close(userfd[i]) # only in child!
-                if userfd[i] != None and userfd[i] < 0:
-                    os.close(sysfd[i])
+                    os.dup2(userfd[i], i)
+                    if userfd[i] != i and userfd[i] not in userfd[0:i]:
+                        _eintr_wrapper(os.close, userfd[i]) # only in child!
             # Set up special control pipe
-            os.close(r)
-            fcntl.fcntl(w, fcntl.F_SETFD, fcntl.FD_CLOEXEC)
+            _eintr_wrapper(os.close, r)
+            flags = fcntl.fcntl(w, fcntl.F_GETFD)
+            fcntl.fcntl(w, fcntl.F_SETFD, flags | fcntl.FD_CLOEXEC)
             if user != None:
                 # Change user
                 os.setgid(gid)
@@ -95,45 +97,34 @@ def spawn(executable, argv = None, cwd = None, env = None,
                 # subprocess.py
                 v.child_traceback = "".join(
                         traceback.format_exception(t, v, tb))
-                os.write(w, pickle.dumps(v))
-                os.close(w)
+                _eintr_wrapper(os.write, w, pickle.dumps(v))
+                _eintr_wrapper(os.close, w)
+                #traceback.print_exc()
             except:
                 traceback.print_exc()
             os._exit(1)
 
-    os.close(w)
+    _eintr_wrapper(os.close, w)
 
     # read EOF for success, or a string as error info
     s = ""
     while True:
-        s1 = os.read(r, 4096)
+        s1 = _eintr_wrapper(os.read, r, 4096)
         if s1 == "":
             break
         s += s1
-    os.close(r)
+    _eintr_wrapper(os.close, r)
 
     if s == "":
         return pid
 
     # It was an error
-    os.waitpid(pid, 0)
+    _eintr_wrapper(os.waitpid, pid, 0)
     exc = pickle.loads(s)
     # XXX: sys.excepthook
     #print exc.child_traceback
     raise exc
 
-# Used to print extra info in nested exceptions
-def _custom_hook(t, v, tb):
-    sys.stderr.write("wee\n")
-    if hasattr(v, "child_traceback"):
-        sys.stderr.write("Nested exception, original traceback " +
-                "(most recent call last):\n")
-        sys.stderr.write(v.child_traceback + ("-" * 70) + "\n")
-    sys.__excepthook__(t, v, tb)
-
-# XXX: somebody kill me, I deserve it :)
-sys.excepthook = _custom_hook
-
 def poll(pid):
     """Check if the process already died. Returns the exit code or None if
     the process is still alive."""
@@ -144,7 +135,7 @@ def poll(pid):
 
 def wait(pid):
     """Wait for process to die and return the exit code."""
-    return os.waitpid(pid, 0)[1]
+    return _eintr_wrapper(os.waitpid, pid, 0)[1]
 
 class Subprocess(object):
     # FIXME: this is the visible interface; documentation should move here.
@@ -167,6 +158,7 @@ class Subprocess(object):
                 user = user)
 
         node._add_subprocess(self)
+        self._returncode = None
 
     @property
     def pid(self):
@@ -176,14 +168,91 @@ class Subprocess(object):
         r = self._slave.poll(self._pid)
         if r != None:
             del self._pid
-            self.return_value = r
-        return r
+            self._returncode = r
+        return self.returncode
 
     def wait(self):
-        r = self._slave.wait(self._pid)
+        self._returncode = self._slave.wait(self._pid)
         del self._pid
-        self.return_value = r
-        return r
+        return self.returncode
 
     def signal(self, sig = signal.SIGTERM):
         return self._slave.signal(self._pid, sig)
+
+    @property
+    def returncode(self):
+        if self._returncode == None:
+            return None
+        if os.WIFSIGNALED(self._returncode):
+            return -os.WTERMSIG(self._returncode)
+        if os.WIFEXITED(self._returncode):
+            return os.EXITSTATUS(self._returncode)
+        raise RuntimeError("Invalid return code")
+
+    # FIXME: do we have any other way to deal with this than having explicit
+    # destroy?
+    def destroy(self):
+        pass
+
+PIPE = -1
+STDOUT = -2
+class Popen(Subprocess):
+    def __init__(self, node, executable, argv = None, cwd = None, env = None,
+            stdin = None, stdout = None, stderr = None, user = None,
+            bufsize = 0):
+        self.stdin = self.stdout = self.stderr = None
+        fdmap = { "stdin": stdin, "stdout": stdout, "stderr": stderr }
+        # if PIPE: all should be closed at the end
+        for k, v in fdmap:
+            if v == None:
+                continue
+            if v == PIPE:
+                r, w = os.pipe()
+                if k == "stdin":
+                    setattr(self, k, os.fdopen(w, 'wb', bufsize))
+                    fdmap[k] = r
+                else:
+                    setattr(self, k, os.fdopen(w, 'rb', bufsize))
+                    fdmap[k] = w
+            elif isinstance(v, int):
+                pass
+            else:
+                fdmap[k] = v.fileno()
+        if stderr == STDOUT:
+            fdmap['stderr'] = fdmap['stdout']
+
+        #print fdmap
+
+        super(Popen, self).__init__(node, executable, argv = argv, cwd = cwd,
+                env = env, stdin = fdmap['stdin'], stdout = fdmap['stdout'],
+                stderr = fdmap['stderr'], user = user)
+
+        # Close pipes, they have been dup()ed to the child
+        for k, v in fdmap:
+            if getattr(self, k) != None:
+                _eintr_wrapper(os.close, v)
+
+#    def comunicate(self, input = None)
+
+def _eintr_wrapper(f, *args):
+    "Wraps some callable with a loop that retries on EINTR"
+    while True:
+        try:
+            return f(*args)
+        except OSError, e:
+            if e.errno == errno.EINTR:
+                continue
+            else:
+                raise
+
+# Used to print extra info in nested exceptions
+def _custom_hook(t, v, tb):
+    if hasattr(v, "child_traceback"):
+        sys.stderr.write("Nested exception, original traceback " +
+                "(most recent call last):\n")
+        sys.stderr.write(v.child_traceback + ("-" * 70) + "\n")
+    sys.__excepthook__(t, v, tb)
+
+# XXX: somebody kill me, I deserve it :)
+sys.excepthook = _custom_hook
+

t/test_subprocess.py

@@ -22,6 +22,21 @@ def _getpwuid(uid):
     except:
         return None
 
+def _readall(fd):
+    s = ""
+    while True:
+        try:
+            s1 = os.read(fd, 4096)
+        except OSError, e:
+            if e.errno == errno.EINTR:
+                continue
+            else:
+                raise
+        if s1 == "":
+            break
+        s += s1
+    return s
+
 class TestSubprocess(unittest.TestCase):
     def _check_ownership(self, user, pid):
         uid = pwd.getpwnam(user)[2]
@@ -51,8 +66,6 @@ class TestSubprocess(unittest.TestCase):
         while _stat(self.nofile):
             self.nofile += '_'
 
-    # XXX: unittest still cannot skip tests
-    #@unittest.skipUnless(os.getuid() == 0, "Test requires root privileges")
     @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
     def test_spawn_chuser(self):
         user = 'nobody'
@@ -62,6 +75,16 @@ class TestSubprocess(unittest.TestCase):
         os.kill(pid, signal.SIGTERM)
         self.assertEquals(netns.subprocess.wait(pid), signal.SIGTERM)
 
+    @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
+    def test_Subprocess_chuser(self):
+        node = netns.Node(nonetns = True)
+        user = 'nobody'
+        p = netns.subprocess.Subprocess(node, '/bin/sleep',
+                ['/bin/sleep', '1000'], user = user)
+        self._check_ownership(user, p.pid)
+        p.signal()
+        self.assertEquals(p.wait(), -signal.SIGTERM)
+
     def test_spawn_basic(self):
         # User does not exist
         self.assertRaises(ValueError, netns.subprocess.spawn,
@@ -82,13 +105,16 @@ class TestSubprocess(unittest.TestCase):
         # uses a default search path
         self.assertRaises(OSError, netns.subprocess.spawn,
                 'sleep', env = {'PATH': ''})
-        #p = netns.subprocess.spawn(None, '/bin/sleep', ['/bin/sleep', '1000'],
-        #        cwd = '/', env = [])
-        # FIXME: tests fds
 
-    @test_util.skipUnless(os.getuid() == 0, "Test requires root privileges")
+        r, w = os.pipe()
+        p = netns.subprocess.spawn('/bin/echo', ['echo', 'hello world'],
+                stdout = w)
+        os.close(w)
+        self.assertEquals(_readall(r), "hello world\n")
+        os.close(r)
+
     def test_Subprocess_basic(self):
-        node = netns.Node()
+        node = netns.Node(nonetns = True) #, debug = True)
         # User does not exist
         self.assertRaises(RuntimeError, netns.subprocess.Subprocess, node,
                 '/bin/sleep', ['/bin/sleep', '1000'], user = self.nouser)
@@ -108,10 +134,15 @@ class TestSubprocess(unittest.TestCase):
         # uses a default search path
         self.assertRaises(RuntimeError, netns.subprocess.Subprocess, node,
                 'sleep', env = {'PATH': ''})
-        #p = netns.subprocess.Subprocess(None, '/bin/sleep', ['/bin/sleep', '1000'], cwd = '/', env = [])
         # FIXME: tests fds
+        r, w = os.pipe()
+        p = netns.subprocess.Subprocess(node, '/bin/echo',
+                ['echo', 'hello world'], stdout = w)
+        os.close(w)
+        self.assertEquals(_readall(r), "hello world\n")
+        os.close(r)
 
-
+# FIXME: tests for Popen!
 if __name__ == '__main__':
     unittest.main()