1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
# -*- coding: utf-8 -*-
import unittest
import bottle
import urllib2
import time
from tools import tob
import sys
import os
import signal
import socket
from subprocess import Popen, PIPE
import tools
serverscript = os.path.join(os.path.dirname(__file__), 'servertest.py')
def ping(server, port):
''' Check if a server accepts connections on a specific TCP port '''
try:
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((server, port))
s.close()
return True
except socket.error, e:
return False
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
# 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):
if self.skip: return
if self.p.poll() == None:
os.kill(self.p.pid, signal.SIGINT)
time.sleep(0.5)
while self.p.poll() == None:
os.kill(self.p.pid, signal.SIGTERM)
time.sleep(1)
for stream in (self.p.stdout, self.p.stderr):
for line in stream:
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:
return urllib2.urlopen('http://127.0.0.1:%d/%s' % (self.port, url)).read()
except Exception, e:
return repr(e)
def test_simple(self):
''' Test a simple static page with this server adapter. '''
if self.skip: return
self.assertEqual(tob('OK'), self.fetch('test'))
class TestCherryPyServer(TestServer):
server = 'cherrypy'
class TestPasteServer(TestServer):
server = 'paste'
class TestTornadoServer(TestServer):
server = 'tornado'
class TestTwistedServer(TestServer):
server = 'twisted'
class TestDieselServer(TestServer):
server = 'diesel'
class TestGunicornServer(TestServer):
server = 'gunicorn'
class TestGeventServer(TestServer):
server = 'gevent'
class TestEventletServer(TestServer):
server = 'eventlet'
class TestRocketServer(TestServer):
server = 'rocket'
class TestFapwsServer(TestServer):
server = 'fapws3'
class TestFapwsServer(TestServer):
server = 'meinheld'
class TestBjoernServer(TestServer):
server = 'bjoern'
if __name__ == '__main__': #pragma: no cover
unittest.main()
|
