diff options
Diffstat (limited to 'test/test_server.py')
-rw-r--r-- | test/test_server.py | 67 |
1 files changed, 40 insertions, 27 deletions
diff --git a/test/test_server.py b/test/test_server.py index b251b3c..db64a6d 100644 --- a/test/test_server.py +++ b/test/test_server.py @@ -9,6 +9,7 @@ import os import signal import socket from subprocess import Popen, PIPE +import tools serverscript = os.path.join(os.path.dirname(__file__), 'servertest.py') @@ -24,38 +25,51 @@ def ping(server, port): class TestServer(unittest.TestCase): server = 'wsgiref' + skip = False def setUp(self): + if self.skip: return # Find a free port for port in range(8800, 8900): self.port = port - if not ping('127.0.0.1', port): break - else: - raise ValueError("Could not find a free port to test networking.") - # Start servertest.py in a subprocess - cmd = [sys.executable, serverscript, self.server, str(self.port)] - cmd += sys.argv[1:] # pass cmdline arguments to subprocesses - self.p = Popen(cmd, stdout=PIPE, stderr=PIPE) - # Wait for the socket to accept connections - for i in xrange(100): - time.sleep(0.1) - # Check if the process has died for some reason - if self.p.poll() != None: break - if ping('127.0.0.1', self.port): break - else: - raise AssertionError("Server took to long to start up.") + # Start servertest.py in a subprocess + cmd = [sys.executable, serverscript, self.server, str(port)] + cmd += sys.argv[1:] # pass cmdline arguments to subprocesses + self.p = Popen(cmd, stdout=PIPE, stderr=PIPE) + # Wait for the socket to accept connections + for i in xrange(100): + time.sleep(0.1) + # Accepts connections? + if ping('127.0.0.1', port): return + # Server died for some reason... + if not self.p.poll() is None: break + rv = self.p.poll() + if rv is None: + raise AssertionError("Server took to long to start up.") + if rv is 128: # Import error + tools.warn("Skipping %r test (ImportError)." % self.server) + self.skip = True + return + if rv is 3: # Port in use + continue + raise AssertionError("Could not find a free port to test server.") def tearDown(self): - while self.p.poll() is None: + if self.skip: return + for i in range(10): + if self.p.poll() != None: break os.kill(self.p.pid, signal.SIGINT) - time.sleep(0.1) - if self.p.poll() is None: - os.kill(self.p.pid, signal.SIGTERM) + time.sleep(0.1*i) + for i in range(10): + if self.p.poll() != None: break + os.kill(self.p.pid, signal.SIGINT) + time.sleep(i) for stream in (self.p.stdout, self.p.stderr): for line in stream: - if tob('Warning') in line \ - or tob('Error') in line or True: - print line.strip().decode('utf8') + if tob('warning') in line.lower(): + tools.warn(line.strip().decode('utf8')) + elif tob('error') in line.lower(): + raise AssertionError(line.strip().decode('utf8')) def fetch(self, url): try: @@ -63,12 +77,11 @@ class TestServer(unittest.TestCase): except Exception, e: return repr(e) - def test_test(self): + def test_simple(self): ''' Test a simple static page with this server adapter. ''' - if self.p.poll() == None: - self.assertEqual(tob('OK'), self.fetch('test')) - #else: - # self.assertTrue(False, "Server process down") + if self.skip: return + self.assertEqual(tob('OK'), self.fetch('test')) + class TestCherryPyServer(TestServer): |