A few more tests

test/test_protocol.py

@@ -11,6 +11,16 @@ class TestServer(unittest.TestCase):
         (s0, s1) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM, 0)
         (s2, s3) = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM, 0)
 
+        def test_help(fd):
+            fd.write("HELP\n")
+            # should be more than one line
+            self.assertEquals(fd.readline()[0:4], "200-")
+            while True:
+                l = fd.readline()
+                self.assertEquals(l[0:3], "200")
+                if l[3] == ' ':
+                    break
+
         def run_server():
             srv = netns.protocol.Server(s0, s0)
             srv.run()
@@ -22,11 +32,13 @@ class TestServer(unittest.TestCase):
 
         s = os.fdopen(s1.fileno(), "r+", 1)
         self.assertEquals(s.readline()[0:4], "220 ")
+        test_help(s)
         s.close()
         s0.close()
 
         s = os.fdopen(s3.fileno(), "r+", 1)
         self.assertEquals(s.readline()[0:4], "220 ")
+        test_help(s)
         s.close()
         s2.close()
         t.join()

test/test_subprocess.py

@@ -172,6 +172,28 @@ class TestSubprocess(unittest.TestCase):
         self.assertEquals(p.wait(), -signal.SIGTERM) # no-op
         self.assertEquals(p.poll(), -signal.SIGTERM) # no-op
 
+        # destroy
+        p = Subprocess(node, ['sleep', '100'])
+        pid = p.pid
+        os.kill(pid, 0) # verify process still there
+        p.destroy()
+        self.assertRaises(OSError, os.kill, pid, 0) # should be dead by now
+
+        # forceful destroy
+        # Command: ignore SIGTERM, write \n to synchronise and then sleep while
+        # closing stdout (so _readall finishes)
+        cmd = 'trap "" SIGTERM; echo; exec sleep 100 > /dev/null'
+
+        r, w = os.pipe()
+        p = Subprocess(node, cmd, shell = True, stdout = w)
+        os.close(w)
+        self.assertEquals(_readall(r), "\n") # wait for trap to be installed
+        os.close(r)
+        pid = p.pid
+        os.kill(pid, 0) # verify process still there
+        p.destroy()
+        self.assertRaises(OSError, os.kill, pid, 0) # should be dead by now
+
         p = Subprocess(node, ['sleep', '100'])
         os.kill(p.pid, signal.SIGTERM)
         time.sleep(0.2)