-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathAppServerTest.py
More file actions
106 lines (95 loc) · 3.96 KB
/
Copy pathAppServerTest.py
File metadata and controls
106 lines (95 loc) · 3.96 KB
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
import os
import time
import unittest
from re import compile as reCompile
from threading import Thread
from Queue import Queue, Empty
from urllib import urlopen
class AppServerTest(unittest.TestCase):
def setUp(self):
"""Setup fixture and test starting."""
workdir = self.workDir()
dirname = os.path.dirname
webwaredir = dirname(dirname(dirname(workdir)))
launch = os.path.join(workdir, 'Launch.py')
cmd = ("python %s --webware-dir=%s --work-dir=%s"
" ThreadedAppServer" % (launch, webwaredir, workdir))
self._output = os.popen(cmd)
# Setting the appserver output to non-blocking mode
# could be done as follows on Unix systems:
# fcntl.fcntl(self._output.fileno(), fcntl.F_SETFL, os.O_NONBLOCK)
# But, since this does not work on Windows systems,
# we will pull the output in a separate thread:
def pullStream(stream, queue):
while stream:
line = stream.readline()
if not line:
break
queue.put(line)
self._queue = Queue()
self._thread = Thread(target=pullStream,
args=(self._output, self._queue))
self._thread.start()
self.assertAppServerSays('^WebKit AppServer ')
self.assertAppServerSays(' Webware for Python.$')
self.assertAppServerSays(' by Chuck Esterbrook.')
self.assertAppServerSays('^WebKit and Webware are open source.$')
self.assertAppServerSays('^EnableHTTP\s*=\s*(True|1)$')
self.assertAppServerSays('^HTTPPort\s*=\s*8080$')
self.assertAppServerSays('^Host\s*=\s*(localhost|127.0.0.1)$')
self.assertAppServerSays('^Ready (.*).$')
# We will also test the built-in HTTP server with this:
try:
data = urlopen('http://localhost:8080').read()
except IOError:
data = '<h2>Could not read page.</h2>'
self.assertTrue('<h1>Welcome to Webware!</h1>' in data,
'Built-in HTTP server failure. Anything using port 8080 already?')
self.assertTrue('<h2>Test passed.</h2>' in data)
def workDir(self):
raise NotImplementedError("Subclasses must implement this method.")
def assertAppServerSays(self, pattern, wait=5):
"""Check that the appserver output contains the specified pattern.
If the appserver does not output the pattern within the given number
of seconds, an assertion is raised.
"""
self.assertTrue(self.waitForAppServer(pattern, wait),
"Expected appserver to say '%s',\n"
"but after waiting %d seconds it said:\n%s"
% (pattern, wait, self._actualAppServerOutput))
def waitForAppServer(self, pattern, wait=5):
"""Check that the appserver output contains the specified pattern.
Returns True or False depending on whether the pattern was seen.
"""
start = time.time()
comp = reCompile(pattern)
lines = []
found = False
while 1:
try:
line = self._queue.get_nowait()
except Empty:
line = None
if line is None:
now = time.time()
if now - start > wait:
break
time.sleep(0.2)
else:
if len(lines) > 9:
del lines[0]
lines.append(line)
if comp.search(line):
found = True
break
self._actualAppServerOutput = ''.join(lines)
return found
def tearDown(self):
"""Teardown fixture and test stopping."""
try:
data = urlopen('http://localhost:8080/stop').read()
except IOError:
data = '<h2>Could not read page.</h2>'
self.assertTrue('<h2>The AppServer has been stopped.</h2>' in data)
self.assertAppServerSays('^AppServer has been shutdown.$')
self._output = None