[Zodb-checkins] CVS: Zope/lib/python/zdaemon/tests - donothing.sh:1.1.2.1 parent.py:1.1.2.1 testzdrun.py:1.5.16.1

[Chris McDonough]

Update of /cvs-repository/Zope/lib/python/zdaemon/tests
In directory cvs.zope.org:/tmp/cvs-serv17213/lib/python/zdaemon/tests

Modified Files:
      Tag: Zope-2_7-branch
	testzdrun.py 
Added Files:
      Tag: Zope-2_7-branch
	donothing.sh parent.py 
Log Message:
Merge changes from HEAD since the release of Zope 2.7a1 into the Zope-2_7-branch in preparation for release of Zope 2.7b1.


=== Added File Zope/lib/python/zdaemon/tests/donothing.sh ===
#!/bin/sh
while [ "1" -ne "2" ]; do
   sleep 10
done

    


=== Added File Zope/lib/python/zdaemon/tests/parent.py ===
import time
import os
import sys
import signal

def main():
    # dummy zdctl startup of zdrun
    shutup()
    file = os.path.normpath(os.path.abspath(sys.argv[0]))
    dir = os.path.dirname(file)
    zctldir = os.path.dirname(dir)
    zdrun = os.path.join(zctldir, 'zdrun.py')
    args = [sys.executable, zdrun]
    args += ['-d', '-b', '10', '-s', os.path.join(dir, 'testsock'),
             '-x', '0,2', '-z', dir, os.path.join(dir, 'donothing.sh')]
    flag = os.P_NOWAIT
    #cmd = ' '.join([sys.executable] + args)
    #print cmd
    os.spawnvp(flag, args[0], args)
    while 1:
        # wait to be signaled
        time.sleep(1)

def shutup():
    os.close(0)
    sys.stdin = sys.__stdin__ = open("/dev/null")
    os.close(1)
    sys.stdout = sys.__stdout__ = open("/dev/null", "w")
    os.close(2)
    sys.stderr = sys.__stderr__ = open("/dev/null", "w")

if __name__ == '__main__':
    main()
    


=== Zope/lib/python/zdaemon/tests/testzdrun.py 1.5 => 1.5.16.1 ===
--- Zope/lib/python/zdaemon/tests/testzdrun.py:1.5	Tue Jan 21 13:19:40 2003
+++ Zope/lib/python/zdaemon/tests/testzdrun.py	Mon Jul 21 12:38:42 2003
@@ -6,8 +6,8 @@
 import signal
 import tempfile
 import unittest
+import socket
 from StringIO import StringIO
-
 from zdaemon import zdrun, zdctl
 
 class ZDaemonTests(unittest.TestCase):
@@ -145,6 +145,49 @@
         self.assertEqual(os.WTERMSIG(wsts), signal.SIGTERM)
         proc.setstatus(wsts)
         self.assertEqual(proc.pid, 0)
+
+    def testRunIgnoresParentSignals(self):
+        # Spawn a process which will in turn spawn a zdrun process.
+        # We make sure that the zdrun process is still running even if
+        # its parent process receives an interrupt signal (it should
+        # not be passed to zdrun).
+        zdrun_socket = os.path.join(self.here, 'testsock')
+        zdctlpid = os.spawnvp(
+            os.P_NOWAIT,
+            sys.executable,
+            [sys.executable, os.path.join(self.here, 'parent.py')]
+            )
+        time.sleep(2) # race condition possible here
+        os.kill(zdctlpid, signal.SIGINT)
+        try:
+            response = send_action('status\n', zdrun_socket) or ''
+        except socket.error, msg:
+            response = ''
+        params = response.split('\n')
+        self.assert_(len(params) > 1)
+        # kill the process
+        send_action('exit\n', zdrun_socket)
+        
+def send_action(action, sockname):
+    """Send an action to the zdrun server and return the response.
+
+    Return None if the server is not up or any other error happened.
+    """
+    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        sock.connect(sockname)
+        sock.send(action + "\n")
+        sock.shutdown(1) # We're not writing any more
+        response = ""
+        while 1:
+            data = sock.recv(1000)
+            if not data:
+                break
+            response += data
+        sock.close()
+        return response
+    except socket.error, msg:
+        return None
 
 def test_suite():
     suite = unittest.TestSuite()