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
125
126
127
128
129
130
131
132
133
134
|
# -*- coding: utf-8 -*-
import unittest
import time
from tools import tob
import sys
import os
import signal
import socket
from subprocess import Popen, PIPE
import tools
from bottle import _e
try:
from urllib.request import urlopen
except:
from urllib2 import urlopen
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 '''
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
s.connect((server, port))
return True
except socket.error:
return False
finally:
s.close()
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 range(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 too 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("Server exited with error code %d" % rv)
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 urlopen('http://127.0.0.1:%d/%s' % (self.port, url)).read()
except Exception:
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 MeinheldServer(TestServer):
server = 'meinheld'
class TestBjoernServer(TestServer):
server = 'bjoern'
class TestAiohttpServer(TestServer):
server = 'aiohttp'
class TestAiohttpServer(TestServer):
server = 'uvloop'
|
